fix: parallelize text + media LLM analysis instead of sequential

- text-only and media analysis now run concurrently via Promise.all
- text no longer blocks on media download + vision analysis
- each path independently saves to DB when its own results are ready
- same batch still uses single context fetch + attachment lookup
This commit is contained in:
MythEclipse
2026-06-22 17:42:46 +07:00
parent dfabdc85cd
commit e8286247d6
2 changed files with 94 additions and 76 deletions
@@ -180,14 +180,18 @@ async function processBatch(job: {
const allRows: MessageRecord[] = []; const allRows: MessageRecord[] = [];
// Phase 1: Text-only → save immediately (fast) // ── Parallel: text-only + media analysis run concurrently ──────────
if (textOnly.length > 0) { // Text-only → fast LLM call. Media → download + vision + LLM.
const textResult = await runModerationAnalysis({ // Running both in parallel means media downloads overlap with text LLM call.
// Each path saves to DB as soon as its own results are ready.
// ────────────────────────────────────────────────────────────────────
const textPromise = textOnly.length > 0
? runModerationAnalysis({
targets: textOnly, targets: textOnly,
contextText: contextLines.join("\n"), contextText: contextLines.join("\n"),
attachments, attachments,
}); }).then((result) => {
const textUpdates = textResult.results.map((analysisResult) => ({ const updates = result.results.map((analysisResult) => ({
messageId: analysisResult.messageId, messageId: analysisResult.messageId,
result: { result: {
status: analysisResult.status, status: analysisResult.status,
@@ -202,24 +206,25 @@ async function processBatch(job: {
error: null, error: null,
}, },
})); }));
if (textUpdates.length > 0) { if (updates.length > 0) {
const rows = await updateMessagesAIAnalysisBulk(textUpdates); return updateMessagesAIAnalysisBulk(updates).then((rows) => {
allRows.push(...rows); allRows.push(...rows);
logger.info( logger.info(
{ count: textUpdates.length, conversationKey }, { count: updates.length, conversationKey },
"Text-only batch saved — media analysis still in progress", "Text-only batch saved — media analysis still in progress",
); );
});
} }
} })
: Promise.resolve();
// Phase 2: Media → save when done (slow: download + vision) const mediaPromise = media.length > 0
if (media.length > 0) { ? runModerationAnalysis({
const mediaResult = await runModerationAnalysis({
targets: media, targets: media,
contextText: contextLines.join("\n"), contextText: contextLines.join("\n"),
attachments, attachments,
}); }).then((result) => {
const mediaUpdates = mediaResult.results.map((analysisResult) => ({ const updates = result.results.map((analysisResult) => ({
messageId: analysisResult.messageId, messageId: analysisResult.messageId,
result: { result: {
status: analysisResult.status, status: analysisResult.status,
@@ -234,11 +239,16 @@ async function processBatch(job: {
error: null, error: null,
}, },
})); }));
if (mediaUpdates.length > 0) { if (updates.length > 0) {
const rows = await updateMessagesAIAnalysisBulk(mediaUpdates); return updateMessagesAIAnalysisBulk(updates).then((rows) => {
allRows.push(...rows); allRows.push(...rows);
});
} }
} })
: Promise.resolve();
// Wait for both to complete
await Promise.all([textPromise, mediaPromise]);
logger.info( logger.info(
{ total: messages.length, textOnly: textOnly.length, media: media.length, saved: allRows.length }, { total: messages.length, textOnly: textOnly.length, media: media.length, saved: allRows.length },
@@ -707,12 +707,13 @@ async function runTextOnlyBatch(
const maxBatchSize = config.AI_LLM_TEXT_BATCH_SIZE ?? 20; const maxBatchSize = config.AI_LLM_TEXT_BATCH_SIZE ?? 20;
const timeoutMs = config.AI_LLM_MEDIA_ANALYSIS_TIMEOUT_MS ?? 60000; const timeoutMs = config.AI_LLM_MEDIA_ANALYSIS_TIMEOUT_MS ?? 60000;
// ── Phase A: Prepare context in parallel ──────────────────
// URL fetching (web_content) and SearXNG (web_searches) are independent —
// both just enrich the LLM prompt. Run them concurrently so SearXNG
// doesn't block on slow URLs (or vice versa).
//
// ── Fetch web content from URLs in text-only messages ── // ── Fetch web content from URLs in text-only messages ──
// Prevents LLM from guessing based on domain name alone (e.g., false "scam" flags). const urlFetchPromise = (async () => {
// The fetched text content is injected into the message XML so the LLM can
// analyze the actual page rather than pattern-match the URL string.
const urlFetchMap = new Map<string, string>(); // url → fetched text content
{
const allUrls = new Set<string>(); const allUrls = new Set<string>();
for (const msg of targets) { for (const msg of targets) {
const content = msg.edited_content ?? msg.content; const content = msg.edited_content ?? msg.content;
@@ -720,7 +721,7 @@ async function runTextOnlyBatch(
allUrls.add(url); allUrls.add(url);
} }
} }
const urlArr = Array.from(allUrls).slice(0, 10); // cap to 10 fetches per batch const urlArr = Array.from(allUrls).slice(0, 10);
if (urlArr.length > 0) { if (urlArr.length > 0) {
log.debug( log.debug(
{ urlCount: urlArr.length }, { urlCount: urlArr.length },
@@ -729,6 +730,7 @@ async function runTextOnlyBatch(
const results = await Promise.allSettled( const results = await Promise.allSettled(
urlArr.map((url) => fetchUrlSafely(url)), urlArr.map((url) => fetchUrlSafely(url)),
); );
const map = new Map<string, string>();
for (let i = 0; i < urlArr.length; i++) { for (let i = 0; i < urlArr.length; i++) {
const r = results[i]; const r = results[i];
if ( if (
@@ -736,19 +738,16 @@ async function runTextOnlyBatch(
r.value.type === "text" && r.value.type === "text" &&
r.value.textContent r.value.textContent
) { ) {
urlFetchMap.set(urlArr[i], r.value.textContent); map.set(urlArr[i], r.value.textContent);
}
} }
} }
return map;
} }
return new Map<string, string>();
})();
// ── SearXNG enrichment for suspicious/ambiguous content ────────── // ── SearXNG enrichment ──
// Uses SearXNG to look up references mentioned in messages (e.g. anime const searxngPromise = (async () => {
// titles, drug names). Results are injected as <web_search> XML tags
// so the LLM can make informed decisions instead of guessing.
// All messages are searched — Redis cache prevents redundant lookups.
const searxngResults = new Map<string, string>(); // query → formatted XML
{
const queries = new Set<string>(); const queries = new Set<string>();
for (const msg of targets) { for (const msg of targets) {
const content = msg.edited_content ?? msg.content; const content = msg.edited_content ?? msg.content;
@@ -757,7 +756,7 @@ async function runTextOnlyBatch(
} }
} }
if (queries.size > 0) { if (queries.size > 0) {
const queryArr = Array.from(queries).slice(0, 3); // cap to 3 searches per batch const queryArr = Array.from(queries).slice(0, 3);
log.debug( log.debug(
{ searchQueries: queryArr }, { searchQueries: queryArr },
"Running SearXNG enrichment for batch", "Running SearXNG enrichment for batch",
@@ -765,14 +764,23 @@ async function runTextOnlyBatch(
const results = await Promise.allSettled( const results = await Promise.allSettled(
queryArr.map((q) => searchSearxng(q)), queryArr.map((q) => searchSearxng(q)),
); );
const map = new Map<string, string>();
for (let i = 0; i < queryArr.length; i++) { for (let i = 0; i < queryArr.length; i++) {
const r = results[i]; const r = results[i];
if (r.status === "fulfilled" && r.value.length > 0) { if (r.status === "fulfilled" && r.value.length > 0) {
searxngResults.set(queryArr[i], formatSearchResults(r.value)); map.set(queryArr[i], formatSearchResults(r.value));
}
} }
} }
return map;
} }
return new Map<string, string>();
})();
// Wait for BOTH concurrently
const [urlFetchMap, searxngResults] = await Promise.all([
urlFetchPromise,
searxngPromise,
]);
// ── Group identical short messages (< 20 chars) to reduce redundant analysis ── // ── Group identical short messages (< 20 chars) to reduce redundant analysis ──
// Messages with identical normalized content share a single representative. // Messages with identical normalized content share a single representative.