feat(gateway): Jev (System One) as primary text moderator with LLM fallback (#80)
* feat(gateway): Jev (System One) as primary text moderator with LLM fallback Add TypeSafe Jev via @typesafe-ai/sdk v0.6.0 as the PRIMARY analyzer for text-only moderation sub-batches; the existing LLM stays as the fallback for anything Jev cannot decide confidently (per-message gate rejection or API failure) and for media batches. - jevAnalyzer.ts: TypeSafeClient wrapper, declarative state builder (System One models MUST get factual state, not chat-XML — chat framing made Jev confidently wrong on clean messages at 0.98 confidence), message-id-keyed question builder (5 typed questions per message), cross-consistency acceptance gate (noul↔status↔severity↔action↔category), answer→AnalysisResult mapper, fail-open outcome. - textBatchProcessor.ts: Jev-first per sub-batch, rejected ids + API failures fall back to callModerationLLM; no cross-batch pollution. - config: AI_LLM_JEV_ENABLED/API_KEY/BASE_URL/MODEL/TIMEOUT_MS/MIN_CONFIDENCE. - .env.example: documented all 6 Jev env vars. - tests: unit (question/state builders, gate, mapper) + live smoke (gated behind AI_LLM_JEV_SMOKE=1) verified 4/4 accepted vs real 9router. * fix: auto-fix code quality [skip ci]
This commit is contained in:
asepharyana
@@ -0,0 +1,545 @@
|
||||
/**
|
||||
* jevAnalyzer.ts
|
||||
*
|
||||
* Jev (TypeSafe System One, `oc/jev-1.13-free` via 9router `/v1/systemone`)
|
||||
* — the PRIMARY analyzer for text-only moderation sub-batches. The existing
|
||||
* LLM (`llmChat`) stays as the fallback for any message Jev cannot decide
|
||||
* confidently (see the acceptance gate) and for media batches (Jev is
|
||||
* decision-only, no image input).
|
||||
*
|
||||
* CRITICAL framing rule (verified 2026-09-23, live probes):
|
||||
* Jev is a System One model — it evaluates typed questions against a STATE.
|
||||
* Feeding it the chat-optimized `SYSTEM_RULES` verbatim INSIDE chat-style
|
||||
* XML (`<messages_to_analyze>`, `<location_context>`, …) makes it
|
||||
* pattern-match the structure and return CONFIDENTLY WRONG verdicts
|
||||
* (flagged clean messages at confidence 0.98 in a probe — would pass any
|
||||
* naive gate and could auto-delete innocent content).
|
||||
*
|
||||
* The state MUST be declarative facts:
|
||||
* OBJEK PENILAIAN / PESAN: `- Pesan "<id>" dari "<user>": "<content>"` /
|
||||
* KEBIJAKAN as statements / KONTEKS as statements
|
||||
* and the questions phrased as "is this true" / "classify this" against
|
||||
* those facts. With that framing the same 4-message probe returned 4/4
|
||||
* correct verdicts at confidence 1.0, including the SARA zero-tolerance
|
||||
* case and the technical-clean case.
|
||||
*
|
||||
* The distilled `JEV_POLICY` below is a compact declarative summary of the
|
||||
* full chat policy (`prompts/rules.ts` SYSTEM_RULES). It is deliberately
|
||||
* kept short (~300 tokens) — the LLM keeps the full 12k-char policy; Jev
|
||||
* triages on the core axes, and anything it can't decide confidently falls
|
||||
* back to the LLM. Keep this block in sync when SYSTEM_RULES changes.
|
||||
*/
|
||||
|
||||
import type { Question, Questions } from "@typesafe-ai/sdk";
|
||||
import { choice, noul, TypeSafeClient } from "@typesafe-ai/sdk";
|
||||
import { createChildLogger } from "@/shared/logger/index";
|
||||
import { config } from "../../shared/config/config.js";
|
||||
import { incrementCounterBy } from "../gateway-metrics/index.js";
|
||||
import type { AnalysisResult } from "../message-capture/types.js";
|
||||
|
||||
const log = createChildLogger("jev-analyzer");
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Policy + vocab
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Distilled declarative policy for Jev. DERIVED from `SYSTEM_RULES`
|
||||
* (prompts/rules.ts) — update this when the full policy changes. Kept as
|
||||
* factual statements, NOT instructions (System One evaluates truth).
|
||||
*/
|
||||
export const JEV_POLICY = `KEBIJAKAN SERVER (fakta yang berlaku):
|
||||
- Kata vulgar anatomi (kontol, memek, tit, dick, dll) = pelanggaran berat, tanpa kecuali.
|
||||
- SARA / penistaan agama / parodi ayat / mockery tokoh agama / provokasi antar-agama = pelanggaran berat.
|
||||
- Promosi atau diskusi LGBT = pelanggaran berat (zero-tolerance).
|
||||
- Diskusi Israel/Palestina/Yahudi = pelanggaran berat (zero-tolerance).
|
||||
- Hinaan terarah ke orang (harassment), seksisme, ageisme, diskriminasi fisik = pelanggaran.
|
||||
- Konten seksual eksplisit / ajakan seksual / fetish / lolicon-shota = pelanggaran.
|
||||
- Judi, narkoba, scam, doxxing, ancaman kekerasan, self-harm, child safety, konten ilegal = pelanggaran.
|
||||
- Teknik evasi (zalgo, leetspeak, regional indicator, simbol acak) yang menyembunyikan kata terlarang = pelanggaran.
|
||||
- Spam berulang / promosi = pelanggaran ringan.
|
||||
- Memancing konflik (conflict instigation) = pelanggaran ringan.
|
||||
- Username ofensif saja (isi pesan bersih) = peringatan ringan, BUKAN hapus pesan.
|
||||
- Percakapan teknis/normal, slang santai (anjay, wkwk, gaskeun, njir), typo, panggilan akrab (bang, kak, dek), ekspresi religius normal (astaghfirullah, alhamdulillah), istilah anime (waifu, wibu), lirik/kutipan, makian ke benda mati = BUKAN pelanggaran.
|
||||
- Teks acak (kode, log, stack trace, output API, cuplikan UI) = BUKAN pelanggaran.
|
||||
- Setiap pesan dinilai dari isinya sendiri; konteks percakapan dapat memengaruhi interpretasi, bukan menggantikan isi.`;
|
||||
|
||||
/** Choice labels must stay in sync with `AIRecommendedAction` (moderation-types). */
|
||||
export const JEV_ACTIONS = [
|
||||
"none",
|
||||
"monitor",
|
||||
"warn",
|
||||
"review",
|
||||
"delete",
|
||||
"escalate",
|
||||
] as const;
|
||||
|
||||
/** Category choices — the moderation category vocabulary (kept tight). */
|
||||
export const JEV_CATEGORIES = [
|
||||
"none",
|
||||
"harassment",
|
||||
"hate_speech",
|
||||
"sara",
|
||||
"sexual_content",
|
||||
"vulgar_language",
|
||||
"sexual_deviation",
|
||||
"self_harm",
|
||||
"violence",
|
||||
"illegal_content",
|
||||
"gambling",
|
||||
"drugs",
|
||||
"scam",
|
||||
"spam",
|
||||
"conflict_instigation",
|
||||
"offensive_username",
|
||||
"other",
|
||||
] as const;
|
||||
|
||||
export const JEV_STATUSES = ["clean", "warn", "flagged"] as const;
|
||||
export const JEV_SEVERITIES = [
|
||||
"none",
|
||||
"low",
|
||||
"medium",
|
||||
"high",
|
||||
"critical",
|
||||
] as const;
|
||||
|
||||
/** Policy version stamped on every Jev verdict (cache/DB provenance). */
|
||||
export const JEV_POLICY_VERSION = "jev-systemone-2026-09-23";
|
||||
|
||||
/** Lazily-built SDK client (config resolves at first use). */
|
||||
let client: TypeSafeClient | null = null;
|
||||
let clientKey = "";
|
||||
|
||||
function getClient(): TypeSafeClient | null {
|
||||
if (!config.AI_LLM_JEV_API_KEY) return null;
|
||||
if (!client || clientKey !== config.AI_LLM_JEV_API_KEY) {
|
||||
client = new TypeSafeClient({
|
||||
apiKey: config.AI_LLM_JEV_API_KEY,
|
||||
baseURL: config.AI_LLM_JEV_BASE_URL,
|
||||
defaultModel: config.AI_LLM_JEV_MODEL,
|
||||
timeout: config.AI_LLM_JEV_TIMEOUT_MS,
|
||||
retry: { maxRetries: 0 }, // pipeline owns retries/abort
|
||||
logLevel: "warn",
|
||||
});
|
||||
clientKey = config.AI_LLM_JEV_API_KEY;
|
||||
log.info(
|
||||
{
|
||||
baseURL: config.AI_LLM_JEV_BASE_URL,
|
||||
model: config.AI_LLM_JEV_MODEL,
|
||||
},
|
||||
"Jev analyzer client initialized",
|
||||
);
|
||||
}
|
||||
return client;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Question + state builders
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface JevTarget {
|
||||
/** Message id — echoed verbatim into analysis/result. */
|
||||
id: string;
|
||||
/** Display name shown to the model (username). */
|
||||
user: string;
|
||||
/** Content to evaluate (truncated by caller). */
|
||||
content: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build per-message questions, keyed by the message id itself so the
|
||||
* questions read naturally against the declarative state facts
|
||||
* ("pesan <id> melanggar kebijakan server"). Five questions per message.
|
||||
*/
|
||||
export function buildJevQuestions(targets: JevTarget[]): Questions {
|
||||
const questions: Record<string, Question> = {};
|
||||
for (const t of targets) {
|
||||
const k = t.id;
|
||||
questions[`${k}__v`] = noul(
|
||||
`Pesan ${JSON.stringify(t.id)} dari ${JSON.stringify(t.user)} melanggar kebijakan server`,
|
||||
);
|
||||
questions[`${k}__status`] = choice(
|
||||
`Klasifikasi pesan ${JSON.stringify(t.id)} dari ${JSON.stringify(t.user)}`,
|
||||
{
|
||||
clean: "tidak melanggar kebijakan",
|
||||
warn: "pelanggaran ringan",
|
||||
flagged: "melanggar kebijakan",
|
||||
},
|
||||
);
|
||||
questions[`${k}__severity`] = choice(
|
||||
`Tingkat keparahan pelanggaran pesan ${JSON.stringify(t.id)}`,
|
||||
{
|
||||
none: "tidak ada pelanggaran",
|
||||
low: "ringan",
|
||||
medium: "sedang",
|
||||
high: "berat",
|
||||
critical: "kritis/darurat",
|
||||
},
|
||||
);
|
||||
questions[`${k}__category`] = choice(
|
||||
`Kategori utama pelanggaran pesan ${JSON.stringify(t.id)}`,
|
||||
Object.fromEntries(JEV_CATEGORIES.map((c) => [c, null])),
|
||||
);
|
||||
questions[`${k}__action`] = choice(
|
||||
`Tindakan moderasi yang tepat untuk pesan ${JSON.stringify(t.id)}`,
|
||||
{
|
||||
none: "tidak ada tindakan",
|
||||
monitor: "pantau",
|
||||
warn: "beri peringatan",
|
||||
review: "tinjau manual",
|
||||
delete: "hapus pesan",
|
||||
escalate: "eskalasi",
|
||||
},
|
||||
);
|
||||
}
|
||||
return questions as Questions;
|
||||
}
|
||||
|
||||
export interface JevBatchContext {
|
||||
/** Context block (location/conversation) as raw XML or prose — stripped to facts. */
|
||||
contextBlock: string;
|
||||
/** `<web_searches>` XML block (may be ""). */
|
||||
webSearchBlock: string;
|
||||
/** `<term_glossary>` XML block (may be ""). */
|
||||
glossaryBlock: string;
|
||||
/** Raw channel culture summary (may be undefined). */
|
||||
channelCulture?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Strip XML/HTML tags from a raw block and collapse whitespace so it can be
|
||||
* restated as plain factual prose in the declarative state. Empty after
|
||||
* stripping → omitted from the state.
|
||||
*/
|
||||
function stripToFacts(block: string, maxLength: number): string | null {
|
||||
const cleaned = block
|
||||
.replace(/<[^>]+>/g, " ")
|
||||
.replace(/\s+/g, " ")
|
||||
.trim();
|
||||
if (!cleaned) return null;
|
||||
return JSON.stringify(cleaned.slice(0, maxLength));
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the declarative `state` payload. NO chat/XML scaffolding — plain
|
||||
* factual statements (see the framing rule above; chat-style injection
|
||||
* makes Jev confidently wrong).
|
||||
*/
|
||||
export function buildJevState(
|
||||
targets: JevTarget[],
|
||||
ctx: JevBatchContext,
|
||||
correctedExamples = "",
|
||||
): string {
|
||||
const facts = targets.map(
|
||||
(t) =>
|
||||
`- Pesan ${JSON.stringify(t.id)} dari ${JSON.stringify(t.user)}: ${JSON.stringify(t.content)}`,
|
||||
);
|
||||
const parts = [
|
||||
`OBJEK PENILAIAN: ${targets.length} pesan dari server Discord.`,
|
||||
"PESAN:",
|
||||
...facts,
|
||||
JEV_POLICY,
|
||||
];
|
||||
|
||||
// Conversation/who context as facts (declarative, not instructions).
|
||||
const extraFacts: string[] = [];
|
||||
if (ctx.channelCulture) {
|
||||
extraFacts.push(
|
||||
`KULTUR CHANNEL (fakta): ${JSON.stringify(ctx.channelCulture.slice(0, 800))}`,
|
||||
);
|
||||
}
|
||||
const contextFacts = stripToFacts(ctx.contextBlock, 1200);
|
||||
if (contextFacts) extraFacts.push(`KONTEKS: ${contextFacts}`);
|
||||
const webSearchFacts = stripToFacts(ctx.webSearchBlock, 1500);
|
||||
if (webSearchFacts) extraFacts.push(`HASIL PENCARIAN WEB: ${webSearchFacts}`);
|
||||
const glossaryFacts = stripToFacts(ctx.glossaryBlock, 800);
|
||||
if (glossaryFacts) extraFacts.push(`GLOSARIUM: ${glossaryFacts}`);
|
||||
const correctionFacts = stripToFacts(correctedExamples, 800);
|
||||
if (correctionFacts)
|
||||
extraFacts.push(`KOREKSI SEBELUMNYA: ${correctionFacts}`);
|
||||
if (extraFacts.length > 0) parts.push(...extraFacts);
|
||||
|
||||
return parts.join("\n");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Acceptance gate + mapper
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** Shape of the raw `answers` map returned by `systemOne`. */
|
||||
export type JevAnswers = Record<
|
||||
string,
|
||||
| { type: "noul"; noul: number }
|
||||
| {
|
||||
type: "choice";
|
||||
choice: string;
|
||||
confidence: number;
|
||||
probabilities?: Record<string, number>;
|
||||
}
|
||||
>;
|
||||
|
||||
/** Reads the per-message answer subset by id, missing → null. */
|
||||
function answersOf(
|
||||
answers: JevAnswers,
|
||||
id: string,
|
||||
): {
|
||||
v?: { type: "noul"; noul: number };
|
||||
status?: { type: "choice"; choice: string; confidence: number };
|
||||
severity?: { type: "choice"; choice: string };
|
||||
category?: { type: "choice"; choice: string };
|
||||
action?: { type: "choice"; choice: string };
|
||||
} {
|
||||
return {
|
||||
v: answers[`${id}__v`] as never,
|
||||
status: answers[`${id}__status`] as never,
|
||||
severity: answers[`${id}__severity`] as never,
|
||||
category: answers[`${id}__category`] as never,
|
||||
action: answers[`${id}__action`] as never,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Decide per-message Jev acceptance. Requires ALL five questions present
|
||||
* with valid labels and CROSS-CONSISTENT semantics:
|
||||
* - status choice confidence >= threshold
|
||||
* - status == clean ⟺ noul < 0.5 (flagged/warn need noul ≥ 0.5)
|
||||
* - severity == none ⟺ status == clean (flagged must have severity)
|
||||
* - action == none ⟺ status == clean; warn must not delete/escalate;
|
||||
* clean must never delete/escalate
|
||||
* - category == none ⟺ status == clean
|
||||
* Anything else → LLM fallback (fail-open).
|
||||
*/
|
||||
export function isJevAccepted(
|
||||
answers: JevAnswers,
|
||||
messageId: string,
|
||||
minConfidence: number,
|
||||
): boolean {
|
||||
const a = answersOf(answers, messageId);
|
||||
if (!a.v || a.v.type !== "noul" || typeof a.v.noul !== "number") return false;
|
||||
if (!a.status || a.status.type !== "choice" || !a.status.choice) return false;
|
||||
if (!a.severity || a.severity.type !== "choice" || !a.severity.choice)
|
||||
return false;
|
||||
if (!a.category || a.category.type !== "choice" || !a.category.choice)
|
||||
return false;
|
||||
if (!a.action || a.action.type !== "choice" || !a.action.choice) return false;
|
||||
|
||||
const { status, severity, category, action } = a;
|
||||
if (
|
||||
typeof status.confidence !== "number" ||
|
||||
status.confidence < minConfidence
|
||||
)
|
||||
return false;
|
||||
|
||||
if (!JEV_STATUSES.includes(status.choice as (typeof JEV_STATUSES)[number]))
|
||||
return false;
|
||||
if (
|
||||
!JEV_SEVERITIES.includes(severity.choice as (typeof JEV_SEVERITIES)[number])
|
||||
)
|
||||
return false;
|
||||
if (
|
||||
!JEV_CATEGORIES.includes(category.choice as (typeof JEV_CATEGORIES)[number])
|
||||
)
|
||||
return false;
|
||||
if (!JEV_ACTIONS.includes(action.choice as (typeof JEV_ACTIONS)[number]))
|
||||
return false;
|
||||
|
||||
const noulVal = a.v.noul;
|
||||
|
||||
// noul ↔ status consistency
|
||||
if (status.choice === "clean" && noulVal >= 0.5) return false;
|
||||
if (status.choice !== "clean" && noulVal < 0.5) return false;
|
||||
// severity ↔ status: clean must be none; flagged/warn must NOT be none
|
||||
if (status.choice === "clean" && severity.choice !== "none") return false;
|
||||
if (status.choice !== "clean" && severity.choice === "none") return false;
|
||||
// action ↔ status: clean must be none; flagged must NOT be none;
|
||||
// warn must not delete/escalate; clean must never delete/escalate
|
||||
if (status.choice === "clean" && action.choice !== "none") return false;
|
||||
if (status.choice === "flagged" && action.choice === "none") return false;
|
||||
if (
|
||||
status.choice === "warn" &&
|
||||
(action.choice === "delete" || action.choice === "escalate")
|
||||
)
|
||||
return false;
|
||||
// category ↔ status: clean must be none; flagged must NOT be none
|
||||
if (status.choice === "clean" && category.choice !== "none") return false;
|
||||
if (status.choice === "flagged" && category.choice === "none") return false;
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Map accepted Jev answers for one message into the pipeline's `AnalysisResult`.
|
||||
* All values are derived from the model's own typed answers — no fabrication.
|
||||
*/
|
||||
export function mapJevAnswersToResult(
|
||||
answers: JevAnswers,
|
||||
messageId: string,
|
||||
): AnalysisResult {
|
||||
const a = answersOf(answers, messageId);
|
||||
const v = a.v as { type: "noul"; noul: number };
|
||||
const st = a.status as { type: "choice"; choice: string; confidence: number };
|
||||
const sev = a.severity as { type: "choice"; choice: string };
|
||||
const cat = a.category as { type: "choice"; choice: string };
|
||||
const act = a.action as { type: "choice"; choice: string };
|
||||
|
||||
const status = st.choice as "clean" | "warn" | "flagged";
|
||||
// Calibrated score: clean → 0; warn → 0.45; flagged → P(violates) clamped.
|
||||
const rawNoul = typeof v.noul === "number" ? v.noul : 0;
|
||||
const score =
|
||||
status === "clean"
|
||||
? 0
|
||||
: status === "warn"
|
||||
? 0.45
|
||||
: Math.min(1, Math.max(0.7, rawNoul));
|
||||
const confidence =
|
||||
typeof st.confidence === "number"
|
||||
? st.confidence
|
||||
: config.AI_LLM_JEV_MIN_CONFIDENCE;
|
||||
|
||||
return {
|
||||
messageId,
|
||||
status,
|
||||
flags: cat.choice === "none" ? [] : [cat.choice],
|
||||
score,
|
||||
analysis:
|
||||
`[Jev] status=${status}, kategori=${cat.choice}, keparahan=${sev.choice}, ` +
|
||||
`keyakinan=${confidence.toFixed(2)}, tindakan=${act.choice}, p_melanggar=${rawNoul.toFixed(2)}`,
|
||||
categories: cat.choice === "none" ? [] : [cat.choice],
|
||||
severity: sev.choice as AnalysisResult["severity"],
|
||||
confidence,
|
||||
recommendedAction: act.choice as AnalysisResult["recommendedAction"],
|
||||
policyVersion: JEV_POLICY_VERSION,
|
||||
evidence: [],
|
||||
};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Batch entry point (one systemOne call per sub-batch)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface JevBatchOutcome {
|
||||
/** Accepted Jev verdicts (keyed by message id). */
|
||||
results: AnalysisResult[];
|
||||
/** Answers the gate rejected for ANY reason (keys = message ids). */
|
||||
rejectedIds: string[];
|
||||
/** Raw `SystemOneResult` (for usage logging / raw passthrough). */
|
||||
raw: unknown;
|
||||
/** Error thrown by the call, if the whole call failed (null = success). */
|
||||
error: string | null;
|
||||
}
|
||||
|
||||
/** True when Jev is configured and enabled (fail-open wrapper). */
|
||||
export function isJevEnabled(): boolean {
|
||||
return (
|
||||
config.AI_LLM_JEV_ENABLED === true && Boolean(config.AI_LLM_JEV_API_KEY)
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Analyze one text sub-batch with Jev. NEVER throws for API-level failures —
|
||||
* returns `{ error }` so the caller falls back to the LLM. Aborts (signal)
|
||||
* propagate as errors too (the caller's timeout must abort the SDK call and
|
||||
* fall back, not hang).
|
||||
*/
|
||||
export async function analyzeBatchWithJev(
|
||||
targets: JevTarget[],
|
||||
ctx: JevBatchContext,
|
||||
signal?: AbortSignal,
|
||||
correctedExamples = "",
|
||||
): Promise<JevBatchOutcome> {
|
||||
const outcome: JevBatchOutcome = {
|
||||
results: [],
|
||||
rejectedIds: [],
|
||||
raw: null,
|
||||
error: null,
|
||||
};
|
||||
if (targets.length === 0) return outcome;
|
||||
|
||||
const jevClient = getClient();
|
||||
if (!jevClient) {
|
||||
outcome.error = "Jev client unavailable (no API key)";
|
||||
return outcome;
|
||||
}
|
||||
|
||||
try {
|
||||
const questions = buildJevQuestions(targets);
|
||||
|
||||
// CONCURRENCY from the skill: one ownership layer owns retries — the SDK
|
||||
// gets retry: { maxRetries: 0 } and the pipeline's timeout/abort layer is
|
||||
// the only retry. The call is wrapped in withLlmConcurrency so Jev down
|
||||
// can't flood the router.
|
||||
const { withLlmConcurrency } = await import("./llmClient.js");
|
||||
const systemOneResult = await withLlmConcurrency(async () => {
|
||||
return await jevClient.systemOne(
|
||||
{
|
||||
model: config.AI_LLM_JEV_MODEL,
|
||||
state: buildJevState(targets, ctx, correctedExamples),
|
||||
questions,
|
||||
},
|
||||
{ signal, timeout: config.AI_LLM_JEV_TIMEOUT_MS },
|
||||
);
|
||||
});
|
||||
|
||||
outcome.raw = systemOneResult;
|
||||
const answers = systemOneResult.answers as unknown as JevAnswers;
|
||||
const minConfidence = config.AI_LLM_JEV_MIN_CONFIDENCE ?? 0.9;
|
||||
|
||||
for (const t of targets) {
|
||||
if (isJevAccepted(answers, t.id, minConfidence)) {
|
||||
outcome.results.push(mapJevAnswersToResult(answers, t.id));
|
||||
incrementCounterBy("moderation_jev_decisions", 1, { type: "jev" });
|
||||
} else {
|
||||
outcome.rejectedIds.push(t.id);
|
||||
incrementCounterBy("moderation_jev_decisions", 1, {
|
||||
type: "llm_fallback",
|
||||
});
|
||||
log.debug(
|
||||
{ messageId: t.id, minConfidence },
|
||||
"Jev decision rejected — falling back to LLM",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Usage accounting (same counters as the LLM path).
|
||||
const usage = systemOneResult.usage;
|
||||
if (usage?.input_tokens || usage?.output_tokens) {
|
||||
if (usage.input_tokens) {
|
||||
incrementCounterBy("llm_tokens_total", usage.input_tokens, {
|
||||
model: config.AI_LLM_JEV_MODEL,
|
||||
type: "prompt",
|
||||
label: "jev-batch",
|
||||
});
|
||||
}
|
||||
if (usage.output_tokens) {
|
||||
incrementCounterBy("llm_tokens_total", usage.output_tokens, {
|
||||
model: config.AI_LLM_JEV_MODEL,
|
||||
type: "completion",
|
||||
label: "jev-batch",
|
||||
});
|
||||
}
|
||||
log.info(
|
||||
{
|
||||
targetCount: targets.length,
|
||||
accepted: outcome.results.length,
|
||||
rejected: outcome.rejectedIds.length,
|
||||
model: config.AI_LLM_JEV_MODEL,
|
||||
input_tokens: usage.input_tokens,
|
||||
output_tokens: usage.output_tokens,
|
||||
},
|
||||
"Jev systemone batch usage",
|
||||
);
|
||||
}
|
||||
|
||||
return outcome;
|
||||
} catch (err) {
|
||||
if (err instanceof Error && err.name === "APIUserAbortError") throw err; // real abort — let caller decide
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
outcome.error = msg;
|
||||
log.warn(
|
||||
{ error: msg, targetCount: targets.length },
|
||||
"Jev systemone call failed — falling back to LLM for the whole batch",
|
||||
);
|
||||
return outcome;
|
||||
}
|
||||
}
|
||||
@@ -15,6 +15,12 @@ import type {
|
||||
} from "../message-capture/types.js";
|
||||
import { getChannelCulture } from "./channelCultureStore.js";
|
||||
import { estimateTokens } from "./conversationContext.js";
|
||||
import {
|
||||
analyzeBatchWithJev,
|
||||
isJevEnabled,
|
||||
JEV_POLICY_VERSION,
|
||||
type JevTarget,
|
||||
} from "./jevAnalyzer.js";
|
||||
import type { ModerationPromptContent, RetryState } from "./llmCaller.js";
|
||||
import { callModerationLLM } from "./llmCaller.js";
|
||||
import { analyzeSingleMediaImage } from "./mediaAnalysisClient.js";
|
||||
@@ -41,6 +47,46 @@ import {
|
||||
|
||||
const log = createChildLogger("textBatchProcessor");
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Shared helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** Maps of URL-fetch outcomes, keyed by the fetched URL (text, image, title). */
|
||||
interface UrlFetchResult {
|
||||
text: Map<string, string>;
|
||||
image: Map<string, { data: Buffer; mimeType: string }>;
|
||||
title: Map<string, string>;
|
||||
}
|
||||
|
||||
/** Provider-reported token usage from a raw LLM/Jev payload (may be absent). */
|
||||
interface TokenUsage {
|
||||
prompt_tokens: number;
|
||||
completion_tokens: number;
|
||||
total_tokens: number;
|
||||
}
|
||||
|
||||
/** Read provider-reported token usage from either the LLM or Jev raw payload. */
|
||||
function extractUsage(raw: unknown): TokenUsage | undefined {
|
||||
return (raw as { usage?: TokenUsage } | null)?.usage ?? undefined;
|
||||
}
|
||||
|
||||
/** Render the `<web_searches>` XML block from the query→results map. */
|
||||
function buildWebSearchBlock(webSearchResults: Map<string, string>): string {
|
||||
if (webSearchResults.size === 0) return "";
|
||||
const entries = Array.from(webSearchResults.entries())
|
||||
.map(
|
||||
([q, xml]) =>
|
||||
` <search_query query="${escapeXml(q)}">\n${xml} </search_query>`,
|
||||
)
|
||||
.join("\n");
|
||||
return `<web_searches>\n${entries}\n</web_searches>`;
|
||||
}
|
||||
|
||||
/** Raw message content as the models see it (truncated + sanitized). */
|
||||
function analysisContentOf(msg: MessageRecord): string {
|
||||
return truncateForAi(getAnalysisContent(msg));
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Few-shot correction builder
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -72,21 +118,21 @@ export async function buildCorrectedFewShotExamples(): Promise<string> {
|
||||
// ---------------------------------------------------------------------------
|
||||
// Few-shot correction cache (refreshes hourly)
|
||||
// ---------------------------------------------------------------------------
|
||||
let _correctedExamplesCache: string | null = null;
|
||||
let _correctedExamplesCacheAt = 0;
|
||||
let correctedExamplesCache: string | null = null;
|
||||
let correctedExamplesCacheAt = 0;
|
||||
const CORRECTED_CACHE_TTL_MS = 60 * 60 * 1000;
|
||||
|
||||
async function getCachedCorrectedExamples(): Promise<string> {
|
||||
const now = Date.now();
|
||||
if (
|
||||
_correctedExamplesCache !== null &&
|
||||
now - _correctedExamplesCacheAt < CORRECTED_CACHE_TTL_MS
|
||||
correctedExamplesCache !== null &&
|
||||
now - correctedExamplesCacheAt < CORRECTED_CACHE_TTL_MS
|
||||
) {
|
||||
return _correctedExamplesCache;
|
||||
return correctedExamplesCache;
|
||||
}
|
||||
_correctedExamplesCache = await buildCorrectedFewShotExamples();
|
||||
_correctedExamplesCacheAt = now;
|
||||
return _correctedExamplesCache;
|
||||
correctedExamplesCache = await buildCorrectedFewShotExamples();
|
||||
correctedExamplesCacheAt = now;
|
||||
return correctedExamplesCache;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -102,7 +148,7 @@ export async function runTextOnlyBatch(
|
||||
const timeoutMs = config.AI_LLM_TEXT_ANALYSIS_TIMEOUT_MS ?? 30000;
|
||||
|
||||
// Parallel: URL fetch + SearXNG
|
||||
const urlFetchPromise = (async () => {
|
||||
const urlFetchPromise: Promise<UrlFetchResult> = (async () => {
|
||||
const allUrls = new Set<string>();
|
||||
for (const msg of targets) {
|
||||
for (const url of extractUrlsFromText(msg.edited_content ?? msg.content))
|
||||
@@ -125,11 +171,7 @@ export async function runTextOnlyBatch(
|
||||
if (urlArr.length >= 10) break;
|
||||
}
|
||||
if (urlArr.length === 0) {
|
||||
return {
|
||||
text: new Map<string, string>(),
|
||||
image: new Map<string, { data: Buffer; mimeType: string }>(),
|
||||
title: new Map<string, string>(),
|
||||
};
|
||||
return { text: new Map(), image: new Map(), title: new Map() };
|
||||
}
|
||||
const results = await Promise.allSettled(
|
||||
urlArr.map((url) => fetchUrlSafely(url)),
|
||||
@@ -304,6 +346,7 @@ export async function runTextOnlyBatch(
|
||||
|
||||
const buildContent = async (
|
||||
state: RetryState,
|
||||
subset?: MessageRecord[],
|
||||
): Promise<ModerationPromptContent> => {
|
||||
const correction = state.lastParseError
|
||||
? {
|
||||
@@ -318,9 +361,10 @@ export async function runTextOnlyBatch(
|
||||
channelCulture,
|
||||
});
|
||||
|
||||
const workingSet = subset ?? batch;
|
||||
const messagesBlock = (
|
||||
await Promise.all(
|
||||
batch.map(async (msg) => {
|
||||
workingSet.map(async (msg) => {
|
||||
const content = truncateForAi(getAnalysisContent(msg));
|
||||
const msgUrls = extractUrlsFromText(content);
|
||||
const urlContexts = msgUrls
|
||||
@@ -346,15 +390,7 @@ export async function runTextOnlyBatch(
|
||||
)
|
||||
).join("\n");
|
||||
|
||||
const webSearchBlock =
|
||||
webSearchResults.size > 0
|
||||
? `<web_searches>\n${Array.from(webSearchResults.entries())
|
||||
.map(
|
||||
([q, xml]) =>
|
||||
` <search_query query="${escapeXml(q)}">\n${xml} </search_query>`,
|
||||
)
|
||||
.join("\n")}\n</web_searches>`
|
||||
: "";
|
||||
const webSearchBlock = buildWebSearchBlock(webSearchResults);
|
||||
// Data/instruction separation: the system prompt is stable per mode —
|
||||
// all per-batch context (conversation, web evidence) lives in the USER
|
||||
// payload, ordered oldest-first so targets come last. Personal user
|
||||
@@ -375,7 +411,12 @@ export async function runTextOnlyBatch(
|
||||
const timeoutId = setTimeout(() => abortController.abort(), timeoutMs);
|
||||
timeoutId.unref();
|
||||
|
||||
let batchResult: { results: AnalysisResult[]; raw: unknown };
|
||||
let batchResult: { results: AnalysisResult[]; raw: unknown } = {
|
||||
results: [],
|
||||
raw: null,
|
||||
};
|
||||
// Per-sub-batch verdicts before fan-out (Jev + LLM fallback merged).
|
||||
let subBatchResults: AnalysisResult[] = [];
|
||||
try {
|
||||
// Output budget scales with the prompt: the JSON verdict block is
|
||||
// roughly proportional to message count, so a small sub-batch doesn't
|
||||
@@ -394,15 +435,84 @@ export async function runTextOnlyBatch(
|
||||
16384,
|
||||
Math.max(2048, Math.ceil(subBatchPromptEstimate * 1.5)),
|
||||
);
|
||||
batchResult = await callModerationLLM(
|
||||
buildContent,
|
||||
targetIds,
|
||||
`text-batch-${i + 1}`,
|
||||
abortController.signal,
|
||||
dynamicMaxTokens,
|
||||
);
|
||||
} catch (err: any) {
|
||||
if (err.name === "AbortError" || abortController.signal.aborted) {
|
||||
|
||||
// ── Jev-first (TypeSafe System One) with LLM fallback ────────────────
|
||||
// Jev is the PRIMARY text analyzer: ONE systemOne call per sub-batch
|
||||
// (5 typed questions × N messages, evaluated in parallel by Jev).
|
||||
// Verdicts that pass the acceptance gate are used directly; anything
|
||||
// Jev rejects (low confidence / inconsistent) and any Jev API failure
|
||||
// falls back to the existing LLM call — fail-open, never dead.
|
||||
if (isJevEnabled()) {
|
||||
const jevTargets: JevTarget[] = batch.map((msg) => ({
|
||||
id: msg.id,
|
||||
user: resolveDisplayName(msg),
|
||||
content: analysisContentOf(msg),
|
||||
}));
|
||||
const jevOutcome = await analyzeBatchWithJev(
|
||||
jevTargets,
|
||||
{
|
||||
contextBlock,
|
||||
webSearchBlock: buildWebSearchBlock(webSearchResults),
|
||||
glossaryBlock,
|
||||
channelCulture: channelCultureObj?.culture_summary,
|
||||
},
|
||||
abortController.signal,
|
||||
correctedExamples,
|
||||
);
|
||||
|
||||
subBatchResults.push(...jevOutcome.results);
|
||||
if (jevOutcome.results.length > 0) {
|
||||
log.info(
|
||||
{
|
||||
subBatch: i + 1,
|
||||
accepted: jevOutcome.results.length,
|
||||
rejected: jevOutcome.rejectedIds.length,
|
||||
},
|
||||
"Jev analyzed sub-batch — accepted verdicts kept, rejected go to LLM",
|
||||
);
|
||||
}
|
||||
|
||||
// Which targets still need the LLM?
|
||||
const coveredIds = new Set(subBatchResults.map((r) => r.messageId));
|
||||
const llmTargets = batch.filter((m) => !coveredIds.has(m.id));
|
||||
|
||||
if (llmTargets.length > 0) {
|
||||
const llmResult = await callModerationLLM(
|
||||
(state) => buildContent(state, llmTargets),
|
||||
llmTargets.map((m) => m.id),
|
||||
`text-batch-${i + 1}-jev-fallback`,
|
||||
abortController.signal,
|
||||
dynamicMaxTokens,
|
||||
);
|
||||
subBatchResults.push(...llmResult.results);
|
||||
batchResult = llmResult;
|
||||
logModerationAnalysis(
|
||||
llmTargets.map((m) => m.id),
|
||||
config.AI_LLM_MODEL,
|
||||
llmResult.results,
|
||||
0,
|
||||
extractUsage(llmResult.raw),
|
||||
);
|
||||
} else {
|
||||
// Jev accepted everything — no LLM usage to attribute.
|
||||
batchResult = { results: subBatchResults, raw: null };
|
||||
}
|
||||
} else {
|
||||
// Jev disabled / unconfigured — pure LLM path (unchanged).
|
||||
batchResult = await callModerationLLM(
|
||||
buildContent,
|
||||
targetIds,
|
||||
`text-batch-${i + 1}`,
|
||||
abortController.signal,
|
||||
dynamicMaxTokens,
|
||||
);
|
||||
subBatchResults = batchResult.results;
|
||||
}
|
||||
} catch (err: unknown) {
|
||||
const isAbort =
|
||||
(err instanceof Error && err.name === "AbortError") ||
|
||||
abortController.signal.aborted;
|
||||
if (isAbort) {
|
||||
// Sub-batch timed out — log but DO NOT throw. Previous sub-batches'
|
||||
// results are already in allResults; throwing would discard them.
|
||||
log.warn(
|
||||
@@ -416,35 +526,32 @@ export async function runTextOnlyBatch(
|
||||
clearTimeout(timeoutId);
|
||||
}
|
||||
|
||||
// Fan-out results for deduplicated messages
|
||||
// Fan-out results for deduplicated messages (applies to Jev + LLM
|
||||
// verdicts alike — they only consume AnalysisResult[]).
|
||||
const fannedOutResults =
|
||||
groupMapping.size > 0
|
||||
? batchResult.results.flatMap((result) => {
|
||||
? subBatchResults.flatMap((result) => {
|
||||
const members = groupMapping.get(result.messageId);
|
||||
return members
|
||||
? members.map((memberId) => ({ ...result, messageId: memberId }))
|
||||
: [result];
|
||||
})
|
||||
: batchResult.results;
|
||||
: subBatchResults;
|
||||
|
||||
allResults.push(...fannedOutResults);
|
||||
if (batchResult.raw) lastRaw = batchResult.raw;
|
||||
|
||||
logModerationAnalysis(
|
||||
targetIds,
|
||||
config.AI_LLM_MODEL,
|
||||
batchResult.results,
|
||||
0,
|
||||
(
|
||||
batchResult.raw as {
|
||||
usage?: {
|
||||
prompt_tokens: number;
|
||||
completion_tokens: number;
|
||||
total_tokens: number;
|
||||
};
|
||||
} | null
|
||||
)?.usage ?? undefined,
|
||||
);
|
||||
if (subBatchResults.length > 0) {
|
||||
logModerationAnalysis(
|
||||
targetIds,
|
||||
subBatchResults.every((r) => r.policyVersion === JEV_POLICY_VERSION)
|
||||
? config.AI_LLM_JEV_MODEL
|
||||
: config.AI_LLM_MODEL,
|
||||
subBatchResults,
|
||||
0,
|
||||
extractUsage(batchResult.raw),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
log.debug(
|
||||
|
||||
@@ -193,6 +193,23 @@ export const configSchema = z
|
||||
// (AI_LLM_BASE_URL) but a different model alias. The dedicated NVIDIA
|
||||
// multimodal endpoint was removed.
|
||||
AI_LLM_VISION_MODEL: z.string().default("multimodal"),
|
||||
// ── Jev (TypeSafe System One) — primary text analyzer ────────────────
|
||||
// Jev evaluates typed questions against a state and returns calibrated
|
||||
// structured answers (no text generation). Runs on 9router's
|
||||
// /v1/systemone (free model oc/jev-1.13-free). The existing LLM remains
|
||||
// the fallback for anything Jev cannot decide confidently and for media.
|
||||
AI_LLM_JEV_ENABLED: z
|
||||
.string()
|
||||
.optional()
|
||||
.transform((v) => v === "true")
|
||||
.default(true),
|
||||
AI_LLM_JEV_API_KEY: z.string().optional().default(""),
|
||||
AI_LLM_JEV_BASE_URL: z.string().url().default("http://127.0.0.1:4014"),
|
||||
AI_LLM_JEV_MODEL: z.string().default("oc/jev-1.13-free"),
|
||||
// Per-message acceptance threshold on the status choice confidence.
|
||||
// Below this the message falls back to the LLM (fail-open).
|
||||
AI_LLM_JEV_MIN_CONFIDENCE: z.coerce.number().min(0).max(1).default(0.9),
|
||||
AI_LLM_JEV_TIMEOUT_MS: z.coerce.number().int().positive().default(45_000),
|
||||
AI_LLM_DISABLE_THINKING: z
|
||||
.string()
|
||||
.default("true")
|
||||
|
||||
Reference in New Issue
Block a user