feat(ai-moderation): rich context + link media vision analysis

- Conversation context recency gates (GAP_MS/MAX_AGE_MS): drop stale
  messages before silence gaps; cold_start anchor + flow descriptor
  tells LLM whether conversation is ongoing or restarted
- [location] block: channel name, thread name, nsfw/age flags from
  captured metadata (thread names instead of bare IDs)
- Link media -> multimodal: text-batch URL fetches that resolve to
  images now run vision analysis (bounded 15s) and switch prompt to
  mixed mode; <web_content> gains og:title for page context
- pnpm-workspace.yaml: approve sharp build script (unblocks install)
This commit is contained in:
asepharyana
2026-08-10 11:26:26 +07:00
parent 5d094829c4
commit 4049ab4201
8 changed files with 497 additions and 24 deletions
@@ -3,6 +3,7 @@ allowBuilds:
"@lng2004/node-datachannel": true
esbuild: true
node-av: true
sharp: true
zeromq: true
# pnpm 11 requires build-script approvals here (the legacy `pnpm` field in
# package.json is ignored). Native voice deps need their postinstall build.
@@ -21,7 +21,10 @@ import { config } from "../../shared/config/config.js";
import { initializeDatabase } from "../../shared/database/drizzle.js";
import { messageStore } from "../message-capture/messageStore.js";
import type { MessageRecord } from "../message-capture/types.js";
import { buildConversationContext } from "./conversationContext.js";
import {
buildConversationContext,
buildLocationContext,
} from "./conversationContext.js";
import { runModerationAnalysis } from "./moderationOrchestrator.js";
const logger = createChildLogger("ai-analysis-worker");
@@ -274,8 +277,16 @@ async function processBatch(job: {
contextBefore,
targets: messages,
maxTokens: config.AI_ANALYSIS_MAX_CONTEXT_TOKENS,
maxAgeMs: config.AI_ANALYSIS_CONTEXT_MAX_AGE_MS,
gapMs: config.AI_ANALYSIS_CONTEXT_GAP_MS,
});
const contextText = contextLines.join("\n");
const contextText = [
buildLocationContext(messages),
contextLines.descriptor,
...contextLines.lines,
]
.filter((l) => l.trim().length > 0)
.join("\n");
const targetIds = messages.map((m) => m.id);
const contextIds = contextBefore.map((m) => m.id);
@@ -359,8 +370,16 @@ async function processIndividual(job: {
contextBefore,
targets: [message],
maxTokens: config.AI_ANALYSIS_MAX_CONTEXT_TOKENS,
maxAgeMs: config.AI_ANALYSIS_CONTEXT_MAX_AGE_MS,
gapMs: config.AI_ANALYSIS_CONTEXT_GAP_MS,
});
const contextText = contextLines.join("\n");
const contextText = [
buildLocationContext([message]),
contextLines.descriptor,
...contextLines.lines,
]
.filter((l) => l.trim().length > 0)
.join("\n");
const contextIds = contextBefore.map((m) => m.id);
const attachments = await messageStore.getAttachmentsForMessages([
@@ -13,6 +13,26 @@ export interface ConversationContextInput {
contextBefore: MessageRecord[];
targets: MessageRecord[];
maxTokens: number;
/**
* Hard age cap for context messages (ms). Messages older than this
* relative to the target are stale conversation noise and dropped.
*/
maxAgeMs?: number;
/**
* Silence threshold (ms). A gap between consecutive context messages
* larger than this means the conversation restarted — older messages
* belong to a previous conversation and are dropped.
*/
gapMs?: number;
}
export interface ConversationContextResult {
/** Formatted context lines (oldest → newest, recency-gated). */
lines: string[];
/** One-line flow descriptor: status, span, dropped counts. */
descriptor: string;
/** Number of context messages dropped by the recency gates. */
dropped: number;
}
let _encoder: ReturnType<typeof encodingForModel> | null = null;
@@ -113,16 +133,114 @@ export function formatMessageForPrompt(
return `[${label}] id=${msg.id} time=${timestamp} user=${msg.username}: ${content}${mediaSuffix}${refInfo}`;
}
/**
* Builds a one-line `<location_context>` source line for the batch — channel
* name, thread name and age-restriction flags from captured message metadata.
* The LLM uses it to judge messages in the right channel context (e.g. a
* thread about a specific topic, or an age-restricted channel).
*/
export function buildLocationContext(targets: MessageRecord[]): string {
const target = targets[0];
if (!target?.metadata) return "";
try {
const meta = JSON.parse(target.metadata) as {
channel?: {
channelName?: string | null;
threadName?: string | null;
nsfw?: boolean;
ageRestricted?: boolean;
nsfwLevel?: string | null;
} | null;
};
const ch = meta?.channel;
if (!ch) return "";
const parts: string[] = [];
parts.push(
`id=${target.channel_id}${
ch.channelName ? ` name=${JSON.stringify(ch.channelName)}` : ""
}`,
);
if (target.thread_id || ch.threadName) {
parts.push(
`thread=${target.thread_id}${
ch.threadName ? ` thread_name=${JSON.stringify(ch.threadName)}` : ""
}`,
);
}
if (typeof ch.nsfw === "boolean") {
parts.push(`nsfw=${ch.nsfw}`);
}
if (typeof ch.ageRestricted === "boolean") {
parts.push(`age_restricted=${ch.ageRestricted}`);
}
return `[location] ${parts.join(" ")}`;
} catch {
return "";
}
}
/**
* Builds conversation historical context without including targets.
* Calculates how much token budget targets use, and fills the rest with context.
*
* Two recency gates decide whether a conversation is STILL the same one
* ("obrolan berlanjut") or already restarted:
* - `gapMs`: a silence longer than this between two context messages cuts
* the block there — earlier messages belong to a previous conversation.
* - `maxAgeMs`: anything older than this relative to the target is noise.
*
* On a cold start (no recent context), the nearest messages are kept as a
* sparse anchor and the descriptor says `cold_start` instead of `ongoing`,
* so the LLM does not mistake scattered old messages for an active chat.
*/
export function buildConversationContext(
input: ConversationContextInput,
): string[] {
): ConversationContextResult {
const { contextBefore, targets, maxTokens } = input;
const maxAgeMs = input.maxAgeMs ?? 45 * 60 * 1000;
const gapMs = input.gapMs ?? 12 * 60 * 1000;
// Calculate tokens used by targets (parallel)
const targetTime = targets.reduce(
(min, t) => Math.min(min, t.created_at),
targets[0]?.created_at ?? Date.now(),
);
// ── Recency gating (walk newest → oldest) ───────────────────────────────
const gated: MessageRecord[] = [];
let latestSelected: MessageRecord | null = null;
let gapBeforeMs: number | null = null;
let dropped = 0;
for (let i = contextBefore.length - 1; i >= 0; i--) {
const msg = contextBefore[i];
// Age gate
if (targetTime - msg.created_at > maxAgeMs) {
dropped += i + 1; // everything older also exceeds the age cap
break;
}
// Gap gate — silence between this message and the newer one already selected
if (latestSelected && latestSelected.created_at - msg.created_at > gapMs) {
gapBeforeMs = latestSelected.created_at - msg.created_at;
dropped += i + 1;
break;
}
gated.push(msg);
latestSelected = msg;
}
const gatedNewestFirst = gated.reverse();
let status: "ongoing" | "cold_start" | "sparse";
if (gatedNewestFirst.length === 0) {
// Cold start — keep a small anchor of the nearest messages so the LLM
// still senses the channel, but mark it clearly.
status = "cold_start";
gatedNewestFirst.push(...contextBefore.slice(-2)); // ± 2 nearest to target
} else if (gapBeforeMs === null) {
status = "ongoing";
} else {
status = "sparse";
}
// ── Format + token budget (most recent first, like before) ─────────────
const targetLines = targets.map((msg) =>
formatMessageForPrompt(msg, "target"),
);
@@ -131,7 +249,7 @@ export function buildConversationContext(
0,
);
const contextLines = contextBefore.map((msg) =>
const contextLines = gatedNewestFirst.map((msg) =>
formatMessageForPrompt(msg, "context"),
);
const selectedContextLines: string[] = [];
@@ -148,14 +266,26 @@ export function buildConversationContext(
}
}
const descriptorParts = [
`[conversation_flow] status=${status}`,
`context_msgs=${selectedContextLines.length}`,
`dropped=${dropped}`,
];
if (gapBeforeMs !== null) {
descriptorParts.push(`gap_before_min=${Math.round(gapBeforeMs / 60000)}`);
}
const descriptor = descriptorParts.join(" ");
logger.debug(
{
targetCount: targets.length,
contextCount: selectedContextLines.length,
status,
dropped,
usedTokens,
maxTokens,
},
"Conversation context built",
);
return selectedContextLines;
return { lines: selectedContextLines, descriptor, dropped };
}
@@ -494,8 +494,9 @@ export async function fetchUrlInline(
sourceLabel: `[gambar dari URL ${url} (inline), pesan id=${targetId}]`,
});
} else if (result.type === "text" && result.textContent) {
const titleAttr = result.title ? ` title="${escapeXml(result.title)}"` : "";
webTexts.push(
`<web_content url="${escapeXml(url)}">${escapeXml(result.textContent.slice(0, 2000))}</web_content>`,
`<web_content url="${escapeXml(url)}"${titleAttr}>${escapeXml(result.textContent.slice(0, 2000))}</web_content>`,
);
}
}
@@ -6,7 +6,9 @@
* the LLM for analysis. Extracted from moderationOrchestrator.ts.
*/
import { createChildLogger } from "@/shared/logger/index";
import { delay } from "@/shared/utils/index";
import { config } from "../../shared/config/config.js";
import { resizeImageForVision } from "../attachment-upload/imageResizer.js";
import type {
AnalysisResult,
MessageRecord,
@@ -14,6 +16,7 @@ import type {
import { getChannelCulture } from "./channelCultureStore.js";
import type { ModerationPromptContent, RetryState } from "./llmCaller.js";
import { callModerationLLM } from "./llmCaller.js";
import { analyzeSingleMediaImage } from "./mediaAnalysisClient.js";
import {
buildReferenceXml,
escapeXml,
@@ -33,6 +36,7 @@ import { getRecentCorrectedModerations } from "./textCacheStore.js";
import { extractUrlsFromText, fetchUrlSafely } from "./urlFetcher.js";
import { getUserProfile } from "./userProfileStore.js";
import { initializeUserReputation } from "./userReputationStore.js";
import type { MessageImagePart } from "./visionAnalyzer.js";
const log = createChildLogger("textBatchProcessor");
@@ -84,22 +88,33 @@ export async function runTextOnlyBatch(
allUrls.add(url);
}
const urlArr = Array.from(allUrls).slice(0, 10);
if (urlArr.length === 0) return new Map<string, string>();
if (urlArr.length === 0) {
return {
text: new Map<string, string>(),
image: new Map<string, { data: Buffer; mimeType: string }>(),
title: new Map<string, string>(),
};
}
const results = await Promise.allSettled(
urlArr.map((url) => fetchUrlSafely(url)),
);
const map = new Map<string, string>();
const textMap = new Map<string, string>();
const imageMap = new Map<string, { data: Buffer; mimeType: string }>();
const titleMap = new Map<string, string>();
for (let i = 0; i < urlArr.length; i++) {
const r = results[i];
if (
r.status === "fulfilled" &&
r.value.type === "text" &&
r.value.textContent
) {
map.set(urlArr[i], r.value.textContent);
if (r.status !== "fulfilled") continue;
const v = r.value;
if (v.type === "text" && v.textContent) {
textMap.set(urlArr[i], v.textContent);
if (v.title) titleMap.set(urlArr[i], v.title);
} else if (v.type === "image" && v.data && v.mimeType) {
// Direct image link (or og:image followed from an HTML page) —
// kept for vision analysis below.
imageMap.set(urlArr[i], { data: v.data, mimeType: v.mimeType });
}
}
return map;
return { text: textMap, image: imageMap, title: titleMap };
})();
const searxngPromise = (async () => {
@@ -122,10 +137,11 @@ export async function runTextOnlyBatch(
return map;
})();
const [urlFetchMap, searxngResults] = await Promise.all([
const [urlFetchMaps, searxngResults] = await Promise.all([
urlFetchPromise,
searxngPromise,
]);
const urlFetchMap = urlFetchMaps.text;
// Deduplicate identical short messages
const shortContentGroups = new Map<string, MessageRecord[]>();
@@ -193,6 +209,65 @@ export async function runTextOnlyBatch(
}
}
// ── URL images → multimodal vision evidence ─────────────────────────
// The text batch fetches inline URLs; whenever one resolved to an image
// (direct image link, or og:image followed from an HTML page), run the
// vision model and append its description as media evidence. If any
// message in the sub-batch produced image evidence, the prompt switches
// to "mixed" mode so media-analysis instructions/examples are injected
// — a link to media is analyzed as media, not as bare text.
const batchImageEvidence = new Map<string, string[]>();
let batchHasImageEvidence = false;
const urlImages = urlFetchMaps.image;
const urlTitles = urlFetchMaps.title;
if (urlImages.size > 0) {
const maxDim = config.AI_LLM_IMAGE_MAX_DIMENSION ?? 1024;
const evidenceSets = await Promise.all(
batch.map(async (msg) => {
const content = getAnalysisContent(msg);
const pics = extractUrlsFromText(content)
.slice(0, 3)
.filter((url) => urlImages.has(url));
if (pics.length === 0) return { id: msg.id, lines: [] as string[] };
const lines = await Promise.all(
pics.map(async (url) => {
const img = urlImages.get(url)!;
try {
const { data: resizedBuffer, mimeType: resizedMime } =
await resizeImageForVision(img.data, maxDim);
const part: MessageImagePart = {
type: "image_url",
image_url: {
url: `data:${resizedMime};base64,${resizedBuffer.toString("base64")}`,
},
sourceLabel: `[gambar dari URL ${url} (inline), pesan id=${msg.id}]`,
};
// Bound vision time so a dead vision model can't stall the
// whole text batch — a timeout just skips the evidence.
const timedOut = delay(15000).then(() => null as string | null);
return await Promise.race([
analyzeSingleMediaImage(msg.id, part),
timedOut,
]);
} catch {
return null;
}
}),
);
return {
id: msg.id,
lines: lines.filter((l): l is string => Boolean(l)),
};
}),
);
for (const set of evidenceSets) {
if (set.lines.length > 0) {
batchImageEvidence.set(set.id, set.lines);
batchHasImageEvidence = true;
}
}
}
const buildContent = async (
state: RetryState,
): Promise<ModerationPromptContent> => {
@@ -205,7 +280,7 @@ export async function runTextOnlyBatch(
const correctedExamples = await buildCorrectedFewShotExamples();
const systemText = buildSystemPromptModular({
contextText,
mode: "text",
mode: batchHasImageEvidence ? "mixed" : "text",
correction,
correctedExamples,
channelCulture,
@@ -219,17 +294,21 @@ export async function runTextOnlyBatch(
const urlContexts = msgUrls
.map((url) => {
const ft = urlFetchMap.get(url);
return ft
? `<web_content url="${escapeXml(url)}">${escapeXml(ft)}</web_content>`
: null;
if (!ft) return null;
const title = urlTitles.get(url);
const titleAttr = title ? ` title="${escapeXml(title)}"` : "";
return `<web_content url="${escapeXml(url)}"${titleAttr}>${escapeXml(ft)}</web_content>`;
})
.filter(Boolean)
.join("\n");
const webContext = urlContexts ? `\n${urlContexts}` : "";
const mediaEvidenceCtx = (batchImageEvidence.get(msg.id) ?? [])
.map((line) => `\n${line}`)
.join("");
const userCtx = userContexts.get(msg.user_id) ?? "";
const userProfileCtx = userProfiles.get(msg.user_id) ?? "";
const refXml = await buildReferenceXml(msg);
return `<message id="${msg.id}" user="${msg.username}">\n ${userCtx}${userProfileCtx ? `\n ${userProfileCtx}` : ""}${refXml ? `\n ${refXml}` : ""}\n <content>${escapeXml(content)}</content>${webContext}\n</message>`;
return `<message id="${msg.id}" user="${msg.username}">\n ${userCtx}${userProfileCtx ? `\n ${userProfileCtx}` : ""}${refXml ? `\n ${refXml}` : ""}\n <content>${escapeXml(content)}</content>${webContext}${mediaEvidenceCtx}\n</message>`;
}),
)
).join("\n");
@@ -11,6 +11,8 @@ export interface FetchedUrlContext {
data?: Buffer;
mimeType?: string;
textContent?: string;
/** Page title from og:title / <title> — strong signal for the LLM. */
title?: string;
error?: string;
}
@@ -86,6 +88,50 @@ function extractOgImage(html: string): string | null {
return null;
}
export interface OgMeta {
title: string | null;
description: string | null;
siteName: string | null;
}
/**
* Extracts OpenGraph / twitter meta + <title> from raw HTML. Both attribute
* orders are accepted (<meta property=... content=...> and reversed).
*/
export function extractOgMeta(html: string): OgMeta {
const metaValue = (name: string): string | null => {
const re = new RegExp(
`<meta[^>]*(?:property|name)=["']${name}["'][^>]*content=["']([^"']+)["']`,
"i",
);
const m = html.match(re);
if (m?.[1]) return m[1].replace(/&amp;/g, "&").replace(/&quot;/g, '"');
const reRev = new RegExp(
`<meta[^>]*content=["']([^"']+)["'][^>]*(?:property|name)=["']${name}["']`,
"i",
);
const mRev = html.match(reRev);
return mRev?.[1]
? mRev[1].replace(/&amp;/g, "&").replace(/&quot;/g, '"')
: null;
};
const title =
metaValue("og:title") ||
metaValue("twitter:title") ||
html.match(/<title[^>]*>([^<]+)<\/title>/i)?.[1]?.trim() ||
null;
const description =
metaValue("og:description") ||
metaValue("twitter:description") ||
metaValue("description") ||
null;
const siteName =
metaValue("og:site_name") || metaValue("application-name") || null;
return { title, description, siteName };
}
function truncateAndCleanHtml(html: string, maxLen = 1000): string {
// Strip <script> and <style> entirely
let text = html.replace(
@@ -176,6 +222,7 @@ export async function fetchUrlSafely(
url,
type: "text",
textContent: cleaned,
title: extractOgMeta(text).title ?? undefined,
};
}
@@ -189,6 +189,17 @@ export const configSchema = z
.int()
.positive()
.default(20),
// Recency gates for conversation context. A silence longer than GAP_MS
// between context messages = the conversation restarted (older messages
// dropped); MAX_AGE_MS caps how far back context is considered relevant.
AI_ANALYSIS_CONTEXT_GAP_MS: z.coerce
.number()
.positive()
.default(12 * 60 * 1000),
AI_ANALYSIS_CONTEXT_MAX_AGE_MS: z.coerce
.number()
.positive()
.default(45 * 60 * 1000),
AI_ANALYSIS_PROCESSING_TIMEOUT_MS: z.coerce
.number()
.positive()
@@ -0,0 +1,185 @@
// ═══════════════════════════════════════════════════════════════════════════
// Conversation context v2 — recency gating + location context (pure, no DB)
// ═══════════════════════════════════════════════════════════════════════════
import { describe, expect, it } from "vitest";
import {
buildConversationContext,
buildLocationContext,
} from "../src/modules/ai-moderation/conversationContext.js";
import { extractOgMeta } from "../src/modules/ai-moderation/urlFetcher.js";
import type { MessageRecord } from "../src/modules/message-capture/types.js";
const NOW = 1_800_000_000_000;
function msg(id: string, createdAt: number, content = "hai"): MessageRecord {
return {
id,
guild_id: "g1",
channel_id: "c1",
thread_id: null,
user_id: `u_${id}`,
username: `user_${id}`,
avatar_url: null,
content,
edited_content: null,
created_at: createdAt,
edited_at: null,
deleted_at: null,
type: "text",
is_reply: null,
is_forward: null,
is_crosspost: null,
reference_message_id: null,
reference_channel_id: null,
reference_guild_id: null,
metadata: null,
};
}
function target(id = "t1", createdAt = NOW): MessageRecord {
return {
...msg(id, createdAt),
content: "pesan yang dianalisis",
};
}
const MIN = 60_000;
describe("buildConversationContext — recency gating", () => {
it("keeps an ONGOING conversation — recent messages, small gaps", () => {
const context = [
msg("a", NOW - 8 * MIN),
msg("b", NOW - 6 * MIN),
msg("c", NOW - 4 * MIN),
msg("d", NOW - 2 * MIN),
];
const { lines, descriptor, dropped } = buildConversationContext({
contextBefore: context,
targets: [target()],
maxTokens: 8000,
gapMs: 12 * MIN,
maxAgeMs: 45 * MIN,
});
expect(lines).toHaveLength(4);
expect(dropped).toBe(0);
expect(descriptor).toContain("status=ongoing");
});
it("drops messages before a silence gap — conversation RESTARTED", () => {
const context = [
msg("old1", NOW - 40 * MIN),
msg("old2", NOW - 38 * MIN),
msg("fresh", NOW - 5 * MIN),
];
const { lines, descriptor, dropped } = buildConversationContext({
contextBefore: context,
targets: [target()],
maxTokens: 8000,
gapMs: 12 * MIN,
maxAgeMs: 45 * MIN,
});
// 40min-old messages are within maxAge but 33min before "fresh" → gap gate
expect(lines.some((l) => l.includes("old1"))).toBe(false);
expect(lines.some((l) => l.includes("fresh"))).toBe(true);
expect(dropped).toBe(2);
expect(descriptor).toContain("status=sparse");
expect(descriptor).toContain("gap_before_min=");
});
it("drops everything older than maxAge — stale noise, cold_start anchor kept", () => {
const context = [
msg("ancient", NOW - 120 * MIN),
msg("stale", NOW - 60 * MIN),
];
const { lines, descriptor, dropped } = buildConversationContext({
contextBefore: context,
targets: [target()],
maxTokens: 8000,
gapMs: 12 * MIN,
maxAgeMs: 45 * MIN,
});
// Age gate drops both from the real context block, but the cold-start
// anchor keeps the nearest 2 so the LLM still senses the channel.
expect(dropped).toBe(2);
expect(descriptor).toContain("status=cold_start");
expect(lines).toHaveLength(2);
});
it("keeps a 2-message anchor on cold start so the LLM senses the channel", () => {
const context = [
msg("far1", NOW - 100 * MIN),
msg("far2", NOW - 99 * MIN),
msg("near1", NOW - 50 * MIN),
];
const { lines, descriptor } = buildConversationContext({
contextBefore: context,
targets: [target()],
maxTokens: 8000,
gapMs: 12 * MIN,
maxAgeMs: 45 * MIN,
});
expect(lines).toHaveLength(2); // nearest 2 kept as anchor
expect(lines.some((l) => l.includes("near1"))).toBe(true);
expect(descriptor).toContain("status=cold_start");
});
it("respects the token budget (older lines dropped first)", () => {
const context = Array.from({ length: 20 }, (_, i) =>
msg(`m${i}`, NOW - (i + 1) * MIN),
);
const { lines } = buildConversationContext({
contextBefore: context,
targets: [target()],
maxTokens: 600,
gapMs: 12 * MIN,
maxAgeMs: 45 * MIN,
});
expect(lines.length).toBeLessThan(20);
expect(lines.length).toBeGreaterThan(0);
});
});
describe("buildLocationContext — channel/thread/nsfw enrichment", () => {
it("renders channel name + thread name from captured metadata", () => {
const t = target();
t.metadata = JSON.stringify({
channel: {
channelName: "general",
threadName: "tanya coding",
nsfw: false,
ageRestricted: false,
},
});
const line = buildLocationContext([t]);
expect(line).toContain("[location]");
expect(line).toContain('name="general"');
expect(line).toContain('thread_name="tanya coding"');
expect(line).toContain("nsfw=false");
});
it("returns empty when no metadata", () => {
expect(buildLocationContext([target()])).toBe("");
});
});
describe("extractOgMeta — page title/site for <web_content>", () => {
it("extracts og:title, og:description and og:site_name", () => {
const html = `
<html><head>
<title>Fallback title</title>
<meta property="og:title" content="Judul Halaman &amp; Keren" />
<meta property="og:description" content="Deskripsi halaman" />
<meta property="og:site_name" content="Contoh Site" />
<meta property="og:image" content="https://img.example.com/x.png" />
</head></html>`;
const meta = extractOgMeta(html);
expect(meta.title).toBe("Judul Halaman & Keren");
expect(meta.description).toBe("Deskripsi halaman");
expect(meta.siteName).toBe("Contoh Site");
});
it("falls back to <title> when og:title missing", () => {
const html = "<html><head><title>Plain Title</title></head></html>";
expect(extractOgMeta(html).title).toBe("Plain Title");
});
});