From 0dff7770a1cc4327f17c68070e91bc8fd75becb7 Mon Sep 17 00:00:00 2001 From: asepharyana Date: Sun, 16 Aug 2026 18:51:13 +0700 Subject: [PATCH] perf(ai-moderation): speed up analysis queue (ramai + sepi) - Parallelize per-user reputation/profile fetches in textBatchProcessor (was a serial ~2N DB/Redis round-trip loop per sub-batch; now Promise.all over unique users). Cuts per-batch latency, biggest win on small/quiet batches. - Make the LLM concurrency semaphore dynamic (cached per config value) instead of frozen at import time, so AI_LLM_MAX_CONCURRENT is tunable without code change and reflects current config. - Bump AI_LLM_MAX_CONCURRENT default 5 -> 8 (gemini-flash-lite is cheap; helps throughput when busy). - Lower AI_ANALYSIS_DEBOUNCE_MS 500 -> 250 (snappier first-message analysis when quiet). - Lower AI_ANALYSIS_RECOVERY_INTERVAL_MS 15000 -> 10000 (stuck/errored messages re-analyze sooner). tsc, biome, vitest (129) all clean. --- .../src/modules/ai-moderation/llmClient.ts | 17 +++++++-- .../ai-moderation/textBatchProcessor.ts | 36 +++++++++++-------- .../src/shared/config/index.ts | 6 ++-- 3 files changed, 40 insertions(+), 19 deletions(-) diff --git a/services/discord-gateway/src/modules/ai-moderation/llmClient.ts b/services/discord-gateway/src/modules/ai-moderation/llmClient.ts index efbc58e..89e7f64 100644 --- a/services/discord-gateway/src/modules/ai-moderation/llmClient.ts +++ b/services/discord-gateway/src/modules/ai-moderation/llmClient.ts @@ -18,7 +18,20 @@ const log = createChildLogger("llm-client"); // Concurrency limiter for LLM API calls (inlined from concurrencyLimiter.ts) // --------------------------------------------------------------------------- -const llmSemaphore = pLimit(config.AI_LLM_MAX_CONCURRENT ?? 5); +// The limiter is cached per configured concurrency value so it can be tuned +// (env / BWS) without a code change and always reflects the current config — +// a module-level `pLimit(config.X)` would freeze the cap at import time. +let llmSemaphore = pLimit(config.AI_LLM_MAX_CONCURRENT ?? 5); +let llmSemaphoreLimit = config.AI_LLM_MAX_CONCURRENT ?? 5; + +function getLlmSemaphore() { + const wanted = config.AI_LLM_MAX_CONCURRENT ?? 5; + if (wanted !== llmSemaphoreLimit) { + llmSemaphore = pLimit(wanted); + llmSemaphoreLimit = wanted; + } + return llmSemaphore; +} let activeCount = 0; let pendingCount = 0; @@ -30,7 +43,7 @@ export async function withLlmConcurrency(fn: () => Promise): Promise { "Queuing LLM request", ); - return llmSemaphore(async () => { + return getLlmSemaphore()(async () => { pendingCount--; activeCount++; diff --git a/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts b/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts index 7277e80..4def40b 100644 --- a/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts +++ b/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts @@ -214,20 +214,28 @@ export async function runTextOnlyBatch( asOf?: number | null; } >(); - for (const msg of batch) { - if (!userContexts.has(msg.user_id)) { - const rep = await initializeUserReputation(msg.user_id, msg.guild_id); - const repAttrs = formatReputationAttrs(rep); - const repXml = ``; - userContexts.set(msg.user_id, repXml); - } - if (!userProfiles.has(msg.user_id)) { - const profile = await getUserProfile(msg.user_id); - userProfiles.set(msg.user_id, { - text: profile?.profile_summary ?? "", - asOf: profile?.last_analyzed_at ?? null, - }); - } + // ── Per-user reputation + profile context (fetched ONCE per unique user, + // in parallel — was a serial per-message loop that cost ~2N sequential + // DB/Redis round-trips per sub-batch and dominated latency on small + // batches). ───────────────────────────────────────────────────────── + const uniqueUserIds = [...new Set(batch.map((m) => m.user_id))]; + const batchGuildId = batch[0]?.guild_id ?? ""; + const userFetches = await Promise.all( + uniqueUserIds.map(async (uid) => { + const [rep, profile] = await Promise.all([ + initializeUserReputation(uid, batchGuildId), + getUserProfile(uid), + ]); + return { uid, rep, profile }; + }), + ); + for (const { uid, rep, profile } of userFetches) { + const repAttrs = formatReputationAttrs(rep); + userContexts.set(uid, ``); + userProfiles.set(uid, { + text: profile?.profile_summary ?? "", + asOf: profile?.last_analyzed_at ?? null, + }); } const userProfilesBlock = buildUserProfilesBlock(userProfiles); diff --git a/services/discord-gateway/src/shared/config/index.ts b/services/discord-gateway/src/shared/config/index.ts index 0a9f7f8..b692e0b 100644 --- a/services/discord-gateway/src/shared/config/index.ts +++ b/services/discord-gateway/src/shared/config/index.ts @@ -177,7 +177,7 @@ export const configSchema = z QDRANT_URL: z.string().optional(), QDRANT_COLLECTION: z.string().default("gmw_text_moderation"), QDRANT_API_KEY: z.string().optional(), - AI_LLM_MAX_CONCURRENT: z.coerce.number().int().positive().default(5), + AI_LLM_MAX_CONCURRENT: z.coerce.number().int().positive().default(8), AI_LLM_IMAGE_MAX_DIMENSION: z.coerce .number() .int() @@ -226,11 +226,11 @@ export const configSchema = z .default(5), // ── AI Analysis Timing ────────────────────────────────────────────── - AI_ANALYSIS_DEBOUNCE_MS: z.coerce.number().positive().default(500), + AI_ANALYSIS_DEBOUNCE_MS: z.coerce.number().positive().default(250), AI_ANALYSIS_RECOVERY_INTERVAL_MS: z.coerce .number() .positive() - .default(15000), + .default(10000), AI_ANALYSIS_ERROR_COOLDOWN_MS: z.coerce.number().positive().default(30000), // ── AI Analysis Batch ───────────────────────────────────────────────