diff --git a/.env.example b/.env.example index ccf2112..6e37291 100644 --- a/.env.example +++ b/.env.example @@ -48,6 +48,8 @@ AI_ANALYSIS_ENABLED=false AI_LLM_API_KEY=your_9router_key_here AI_LLM_BASE_URL=https://9router.asepharyana.tech/v1 AI_LLM_MODEL=free +# Vision model for image/video moderation (falls back to AI_LLM_MODEL if unset) +AI_LLM_VISION_MODEL=multimodal # NVIDIA Nemotron Content Safety Configuration NVIDIA_NEMOTRON_API_KEY=your_nvidia_api_key_here diff --git a/src/config.ts b/src/config.ts index 47dc446..0beda09 100644 --- a/src/config.ts +++ b/src/config.ts @@ -71,14 +71,18 @@ const configSchema = z .string() .url() .default("https://9router.asepharyana.tech/v1"), - AI_LLM_MODEL: z.string().default("free"), + /** Model used for text-only moderation (messages, badword analysis). */ + AI_LLM_MODEL: z.string().default("text"), + /** Model used for image/video moderation (vision-capable model). */ + AI_LLM_VISION_MODEL: z.string().optional(), AI_ANALYSIS_DEBOUNCE_MS: z.coerce.number().positive().default(500), AI_ANALYSIS_RECOVERY_INTERVAL_MS: z.coerce .number() .positive() .default(15000), AI_ANALYSIS_ERROR_COOLDOWN_MS: z.coerce.number().positive().default(30000), - AI_ANALYSIS_MAX_BATCH_SIZE: z.coerce.number().int().positive().default(25), + /** Max messages fetched per conversation batch (token budget is the real constraint). */ + AI_ANALYSIS_MAX_BATCH_SIZE: z.coerce.number().int().positive().default(200), AI_ANALYSIS_MAX_CONTEXT_TOKENS: z.coerce.number().positive().default(8000), /** Token budget for target messages specifically (separate from context window). */ AI_ANALYSIS_MAX_TARGET_TOKENS: z.coerce.number().positive().default(4000), @@ -96,15 +100,12 @@ const configSchema = z .number() .positive() .default(120000), - /** - * Maximum number of concurrent individual-fallback LLM calls. - * Prevents OOM/connection exhaustion when many messages miss a batch. - */ + /** Max concurrent individual-fallback LLM calls (effectively unlimited). */ AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT: z.coerce .number() .int() .positive() - .default(20), + .default(1000), /** * How many consecutive individual-fallback errors trigger the individual * circuit breaker (separate from the batch circuit breaker). @@ -113,7 +114,7 @@ const configSchema = z .number() .int() .positive() - .default(10), + .default(50), /** NVIDIA Nemotron-3 Content Safety API key for badword detection. */ NVIDIA_NEMOTRON_API_KEY: z.string().optional(), /** NVIDIA Nemotron model identifier. */ diff --git a/src/moderation/aiAnalyzer.ts b/src/moderation/aiAnalyzer.ts index 2772a57..24c06fe 100644 --- a/src/moderation/aiAnalyzer.ts +++ b/src/moderation/aiAnalyzer.ts @@ -486,34 +486,15 @@ function enqueueIndividualFallbacks(messages: MessageRecord[]): void { const newMessages = messages.filter((m) => !individualInFlight.has(m.id)); if (newMessages.length === 0) return; - // FIX #1: Enforce concurrency cap. - const availableSlots = - config.AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT - individualInFlight.size; - if (availableSlots <= 0) { - logger.warn( - { - cap: config.AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT, - inFlight: individualInFlight.size, - skipped: newMessages.length, - }, - "Individual fallback concurrency cap reached — messages will be recovered by recovery worker", - ); - return; - } - - const toProcess = newMessages.slice(0, availableSlots); - const skipped = newMessages.length - toProcess.length; - logger.info( { - count: toProcess.length, - skipped, - messageIds: toProcess.map((m) => m.id), + count: newMessages.length, + messageIds: newMessages.map((m) => m.id), }, "Enqueueing individual fallback analysis for batch-incomplete messages", ); - for (const msg of toProcess) { + for (const msg of newMessages) { individualInFlight.add(msg.id); // Fire-and-forget: processIndividualFallback handles all errors internally. processIndividualFallback(msg).catch((err) => { @@ -855,8 +836,8 @@ export function startPendingAIAnalysisWorker(client?: Client): void { setInterval(() => { // FIX #3 pattern: no async arrow — chain promises explicitly. Promise.all([ - getPendingConversationKeys(100), - getConversationKeysWithIncompleteAnalysis(50), + getPendingConversationKeys(500), + getConversationKeysWithIncompleteAnalysis(200), ]) .then(([pendingKeys, incompleteKeys]) => { const now = Date.now(); @@ -902,10 +883,7 @@ export function startPendingAIAnalysisWorker(client?: Client): void { if (isConversationProcessingLocked(key)) continue; promises.push( - getIncompleteMessagesByConversation( - key, - config.AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT, - ) + getIncompleteMessagesByConversation(key, 500) .then(async (msgs) => { const processableMessages = await skipAgeRestrictedMessages(msgs); diff --git a/src/moderation/llmModerationClient.ts b/src/moderation/llmModerationClient.ts index 4f9422a..0fb469c 100644 --- a/src/moderation/llmModerationClient.ts +++ b/src/moderation/llmModerationClient.ts @@ -769,7 +769,7 @@ export async function runModerationAnalysis( ): Promise => { try { const completion = await openai.chat.completions.create({ - model: config.AI_LLM_MODEL, + model: config.AI_LLM_VISION_MODEL ?? config.AI_LLM_MODEL, messages: [ { role: "user", diff --git a/src/moderation/messageStore.ts b/src/moderation/messageStore.ts index c48620b..5449a38 100644 --- a/src/moderation/messageStore.ts +++ b/src/moderation/messageStore.ts @@ -630,7 +630,7 @@ export async function getConversationContextBefore(input: { export async function getPendingMessagesByConversation( conversationKey: string, - limit: number = 25, + limit: number = 200, ): Promise { try { const database = db(); @@ -667,7 +667,7 @@ export async function getPendingMessagesByConversation( } export async function getPendingConversationKeys( - limit: number = 100, + limit: number = 500, ): Promise { try { const database = db(); @@ -782,7 +782,7 @@ export async function searchMessages(input: { * the individual-fallback queue. */ export async function getConversationKeysWithIncompleteAnalysis( - limit: number = 50, + limit: number = 200, ): Promise { try { const database = db(); @@ -826,7 +826,7 @@ export async function getConversationKeysWithIncompleteAnalysis( */ export async function getIncompleteMessagesByConversation( conversationKey: string, - limit: number = 20, + limit: number = 500, ): Promise { try { const database = db();