perf(ai-moderation): naikkan cache hit dgn guard akurasi
- Fase-1 exact-cache lookup: N query serial -> SATU query ANY($1::text[]) - Global reuse utk bare key legacy, HANYA verdict non-actionable (clean/flagless/action=none, conf>=0.85, umur<=72h) — flagged/warn tetap context-scoped - Semantic cache dua-band: clean band 0.92 default, actionable tetap 0.97; di antara band -> LLM (fail-open ke akurasi) - hit_count kini di-increment (bulk UPDATE per batch) -> hit-rate terukur - Cache hasil wikipediaSearch di Redis (6h, hanya hasil non-kosong) - Memoize fetchUrlSafely utk type=text (LRU 30m + in-flight dedupe) - makeImageCacheKey strip query CDN Discord (?ex/is/hm, format/width) -> attachment sama = satu key vision, skip re-download+re-vision Spec: .hermes/plans/2026-08-24-ai-analysis-cache-optimization.md Tests: +33 (cacheGuards, discordImageKeyNormalize, cacheBatchLookup)
This commit is contained in:
asepharyana
@@ -20,11 +20,16 @@ import { isQdrantConfigured, searchQdrantBatch } from "./qdrantClient.js";
|
||||
import { logCacheEvent } from "./responseLogger.js";
|
||||
import { runTextOnlyBatch } from "./textBatchProcessor.js";
|
||||
import {
|
||||
bumpTextModerationHitCounts,
|
||||
ERROR_ARTIFACT_FLAGS,
|
||||
findSimilarTextModeration,
|
||||
getCachedTextModeration,
|
||||
getCachedTextModerations,
|
||||
isGloballyReusableCleanVerdict,
|
||||
isSemanticBandAccepted,
|
||||
makeModerationContextKey,
|
||||
makeTextModerationCacheKey,
|
||||
parseQdrantVerdict,
|
||||
type StoredModerationVerdict,
|
||||
setCachedTextModeration,
|
||||
} from "./textCacheStore.js";
|
||||
|
||||
@@ -73,97 +78,164 @@ export async function runModerationAnalysis(
|
||||
initCacheStore(config.REDIS_URL);
|
||||
if (!targets.length) throw new Error("No targets provided for analysis");
|
||||
|
||||
// ── Phase 1: exact-hash cache (per conversation context) ────────────────
|
||||
// ── Phase 1: exact-hash cache — ONE batched DB query ────────────────────
|
||||
// Key is content + conversation context (channel/thread). On a scoped miss
|
||||
// we also probe the legacy bare key: verdicts that CANNOT trigger an action
|
||||
// (clean, flagless, action=none) may be reused across channels under strict
|
||||
// freshness + confidence guards — flagged/warn verdicts never leave their
|
||||
// conversation. This replaced the old N-sequential-query loop (60-message
|
||||
// burst = 60 PgBouncer round-trips before).
|
||||
const cacheHits: AnalysisResult[] = [];
|
||||
const uncachedTargets: MessageRecord[] = [];
|
||||
// cacheKey → result for identical-content dedupe within one batch
|
||||
// cacheKey → representative result for identical-content dedupe
|
||||
const hitByKey = new Map<string, AnalysisResult>();
|
||||
// Embedding per exact cache key — computed once during lookup, reused
|
||||
// when the fresh LLM verdict is written back to the semantic cache.
|
||||
const embeddingsByKey = new Map<string, number[]>();
|
||||
|
||||
interface ExactCandidate {
|
||||
target: MessageRecord;
|
||||
scopedKey: string;
|
||||
bareKey: string;
|
||||
}
|
||||
const candidates: ExactCandidate[] = [];
|
||||
for (const target of targets) {
|
||||
const hasMedia = hasMediaContent(target, attachments);
|
||||
if (hasMedia) {
|
||||
if (hasMediaContent(target, attachments)) {
|
||||
uncachedTargets.push(target);
|
||||
continue;
|
||||
}
|
||||
|
||||
const rawContent = target.edited_content ?? target.content;
|
||||
if (!rawContent.trim()) {
|
||||
uncachedTargets.push(target);
|
||||
continue;
|
||||
}
|
||||
candidates.push({
|
||||
target,
|
||||
scopedKey: makeTextModerationCacheKey(
|
||||
rawContent,
|
||||
makeModerationContextKey(target),
|
||||
),
|
||||
bareKey: makeTextModerationCacheKey(rawContent),
|
||||
});
|
||||
}
|
||||
|
||||
const cacheKey = makeTextModerationCacheKey(
|
||||
rawContent,
|
||||
makeModerationContextKey(target),
|
||||
);
|
||||
const seen = hitByKey.get(cacheKey);
|
||||
if (seen) {
|
||||
// Same content already resolved this batch — reuse the verdict.
|
||||
cacheHits.push({ ...seen, messageId: target.id });
|
||||
// Identical content within one batch resolves once (representative).
|
||||
const firstByScopedKey = new Map<string, ExactCandidate>();
|
||||
for (const c of candidates) {
|
||||
if (!firstByScopedKey.has(c.scopedKey))
|
||||
firstByScopedKey.set(c.scopedKey, c);
|
||||
}
|
||||
|
||||
// Single round-trip for every key we might serve from (scoped + bare).
|
||||
const storedEntries = await getCachedTextModerations([
|
||||
...firstByScopedKey.keys(),
|
||||
...Array.from(firstByScopedKey.values(), (c) => c.bareKey),
|
||||
]);
|
||||
// Keys actually served — bumped in one UPDATE at the end for metrics.
|
||||
const servedCacheKeys = new Set<string>();
|
||||
|
||||
/** Validate + admit one stored verdict for a candidate. */
|
||||
const acceptExactVerdict = (
|
||||
candidate: ExactCandidate,
|
||||
cacheKey: string,
|
||||
entry: { verdict: StoredModerationVerdict },
|
||||
policyVersion: string,
|
||||
): boolean => {
|
||||
const { verdict } = entry;
|
||||
const hasMediaInMeta =
|
||||
candidate.target.metadata &&
|
||||
(() => {
|
||||
const ev = extractMessageMediaEvidence(candidate.target.metadata);
|
||||
return (
|
||||
ev.attachments.length > 0 ||
|
||||
ev.stickers.length > 0 ||
|
||||
ev.embeds.length > 0
|
||||
);
|
||||
})();
|
||||
|
||||
if (hasMediaInMeta) {
|
||||
log.debug(
|
||||
{ messageId: candidate.target.id, cacheKey },
|
||||
"Cache entry but message has media — treating as miss",
|
||||
);
|
||||
return false;
|
||||
}
|
||||
if (
|
||||
verdict.flags.some((f) =>
|
||||
(ERROR_ARTIFACT_FLAGS as readonly string[]).includes(f),
|
||||
)
|
||||
) {
|
||||
log.warn(
|
||||
{ messageId: candidate.target.id, cacheKey },
|
||||
"Cache entry contains error artifact — treating as miss",
|
||||
);
|
||||
return false;
|
||||
}
|
||||
|
||||
hitByKey.set(candidate.scopedKey, {
|
||||
messageId: candidate.target.id,
|
||||
status: verdict.status,
|
||||
flags: verdict.flags,
|
||||
score: verdict.score,
|
||||
analysis: verdict.analysis,
|
||||
categories: verdict.categories,
|
||||
severity: verdict.severity as AnalysisResult["severity"],
|
||||
confidence: verdict.confidence,
|
||||
recommendedAction:
|
||||
verdict.recommendedAction as AnalysisResult["recommendedAction"],
|
||||
policyVersion,
|
||||
evidence: [],
|
||||
});
|
||||
servedCacheKeys.add(cacheKey);
|
||||
logCacheEvent("hit", cacheKey, "text");
|
||||
return true;
|
||||
};
|
||||
|
||||
for (const candidate of firstByScopedKey.values()) {
|
||||
const scopedEntry = storedEntries.get(candidate.scopedKey);
|
||||
if (
|
||||
scopedEntry &&
|
||||
acceptExactVerdict(
|
||||
candidate,
|
||||
candidate.scopedKey,
|
||||
scopedEntry,
|
||||
"cached-user-moderation-2026-06",
|
||||
)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
const cached = await getCachedTextModeration(cacheKey);
|
||||
if (cached) {
|
||||
const hasMediaInMeta =
|
||||
target.metadata &&
|
||||
(() => {
|
||||
const ev = extractMessageMediaEvidence(target.metadata);
|
||||
return (
|
||||
ev.attachments.length > 0 ||
|
||||
ev.stickers.length > 0 ||
|
||||
ev.embeds.length > 0
|
||||
);
|
||||
})();
|
||||
|
||||
if (hasMediaInMeta) {
|
||||
log.debug(
|
||||
{ messageId: target.id, cacheKey },
|
||||
"Cache entry but message has media — treating as miss",
|
||||
);
|
||||
} else if (
|
||||
cached.flags.some((f) =>
|
||||
[
|
||||
"analysis_api_failed",
|
||||
"analysis_parse_failed",
|
||||
"analysis_incomplete",
|
||||
].includes(f),
|
||||
)
|
||||
) {
|
||||
log.warn(
|
||||
{ messageId: target.id, cacheKey },
|
||||
"Cache entry contains error artifact — treating as miss",
|
||||
);
|
||||
} else {
|
||||
const hit: AnalysisResult = {
|
||||
messageId: target.id,
|
||||
status: cached.status,
|
||||
flags: cached.flags,
|
||||
score: cached.score,
|
||||
analysis: cached.analysis,
|
||||
categories: cached.categories,
|
||||
severity: cached.severity as AnalysisResult["severity"],
|
||||
confidence: cached.confidence,
|
||||
recommendedAction:
|
||||
cached.recommendedAction as AnalysisResult["recommendedAction"],
|
||||
policyVersion: "cached-user-moderation-2026-06",
|
||||
evidence: [],
|
||||
};
|
||||
cacheHits.push(hit);
|
||||
hitByKey.set(cacheKey, hit);
|
||||
logCacheEvent("hit", cacheKey, "text");
|
||||
continue;
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
/* proceed */
|
||||
// Context-free fallback: ONLY non-actionable clean verdicts qualify
|
||||
// (guard enforces status/flags/action/confidence/freshness). The bare
|
||||
// key equals the scoped key for context-less messages, so the guard
|
||||
// also prevents double-serving the same row.
|
||||
const bareEntry = storedEntries.get(candidate.bareKey);
|
||||
if (
|
||||
bareEntry &&
|
||||
candidate.bareKey !== candidate.scopedKey &&
|
||||
isGloballyReusableCleanVerdict(
|
||||
bareEntry.verdict,
|
||||
bareEntry.analyzedAt ?? undefined,
|
||||
)
|
||||
) {
|
||||
acceptExactVerdict(
|
||||
candidate,
|
||||
candidate.bareKey,
|
||||
bareEntry,
|
||||
"cached-global-clean-2026-08",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
uncachedTargets.push(target);
|
||||
// Fan-out: every candidate (representative + in-batch duplicates) gets its
|
||||
// own copy of the representative verdict; unresolved ones stay queued.
|
||||
for (const candidate of candidates) {
|
||||
const representative = hitByKey.get(candidate.scopedKey);
|
||||
if (representative) {
|
||||
cacheHits.push({ ...representative, messageId: candidate.target.id });
|
||||
} else {
|
||||
uncachedTargets.push(candidate.target);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Phase 2: semantic cache — batched (one embed call + one Qdrant
|
||||
@@ -199,10 +271,12 @@ export async function runModerationAnalysis(
|
||||
}
|
||||
|
||||
if (isQdrantConfigured()) {
|
||||
// ONE batch search at the LOOSER threshold; per-hit re-classification
|
||||
// enforces the strict band for actionable verdicts.
|
||||
const batchHits = await searchQdrantBatch(
|
||||
embeddings,
|
||||
config.AI_LLM_EMBEDDING_MAX_CANDIDATES,
|
||||
config.AI_LLM_EMBEDDING_MIN_SIMILARITY,
|
||||
config.AI_LLM_EMBEDDING_MIN_SIMILARITY_CLEAN,
|
||||
);
|
||||
for (let i = 0; i < semanticCandidates.length; i++) {
|
||||
const { target, cacheKey } = semanticCandidates[i];
|
||||
@@ -210,6 +284,7 @@ export async function runModerationAnalysis(
|
||||
if (hits.length === 0) continue;
|
||||
const verdict = parseQdrantVerdict(hits[0].payload, hits[0].score);
|
||||
if (!verdict) continue;
|
||||
if (!isSemanticBandAccepted(verdict, verdict.similarity)) continue;
|
||||
log.debug(
|
||||
{
|
||||
messageId: target.id,
|
||||
@@ -242,10 +317,12 @@ export async function runModerationAnalysis(
|
||||
const { target, cacheKey } = semanticCandidates[i];
|
||||
const semantic = await findSimilarTextModeration(
|
||||
embeddings[i],
|
||||
config.AI_LLM_EMBEDDING_MIN_SIMILARITY,
|
||||
config.AI_LLM_EMBEDDING_MIN_SIMILARITY_CLEAN,
|
||||
config.AI_LLM_EMBEDDING_MAX_CANDIDATES,
|
||||
);
|
||||
if (!semantic) continue;
|
||||
if (!isSemanticBandAccepted(semantic, semantic.similarity))
|
||||
continue;
|
||||
log.debug(
|
||||
{
|
||||
messageId: target.id,
|
||||
@@ -290,6 +367,8 @@ export async function runModerationAnalysis(
|
||||
}
|
||||
|
||||
if (cacheHits.length > 0) {
|
||||
// Metrics: one bulk UPDATE for every exact-cache key actually served.
|
||||
bumpTextModerationHitCounts(Array.from(servedCacheKeys));
|
||||
log.info(
|
||||
{
|
||||
cacheHits: cacheHits.length,
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { createChildLogger } from "@/shared/logger/index";
|
||||
import { config } from "../../shared/config/config.js";
|
||||
import { executeAll, executeGet } from "../../shared/database/drizzle.js";
|
||||
import { findBestEmbeddingMatch } from "./embeddingClient.js";
|
||||
import {
|
||||
@@ -84,10 +85,43 @@ export function makeImageCacheKey(imageUrl: string): string {
|
||||
// size is 8191"), which fails acquireMediaAnalysisLock and silently skips
|
||||
// every media analysis. A 32-char sha256 keeps the key well under the limit
|
||||
// and is still deterministic (same attachment → same key).
|
||||
const hash = createHash("sha256").update(imageUrl).digest("hex").slice(0, 32);
|
||||
const hash = createHash("sha256")
|
||||
.update(normalizeDiscordImageUrl(imageUrl))
|
||||
.digest("hex")
|
||||
.slice(0, 32);
|
||||
return `image:${hash}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Strip volatile query params from Discord CDN URLs so the SAME attachment
|
||||
* always maps to ONE vision-cache key regardless of how it reached us
|
||||
* (signed `?ex=&is=&hm=` tokens rotate per fetch; render variants differ by
|
||||
* `format/width/height/size`). Previously each token variant hashed to its
|
||||
* own key → the same image was re-downloaded and re-analyzed by the vision
|
||||
* model once per variant. Non-Discord URLs and data: URLs are returned
|
||||
* untouched (their query can be semantically meaningful).
|
||||
*/
|
||||
export function normalizeDiscordImageUrl(imageUrl: string): string {
|
||||
try {
|
||||
const parsed = new URL(imageUrl);
|
||||
if (parsed.protocol !== "https:" && parsed.protocol !== "http:") {
|
||||
return imageUrl;
|
||||
}
|
||||
const host = parsed.hostname;
|
||||
const isAttachmentCdn = host === "cdn.discordapp.com";
|
||||
const isRenderOrPreview =
|
||||
host === "media.discordapp.net" ||
|
||||
/^images-ext-\d+\.discordapp\.net$/.test(host);
|
||||
if (!isAttachmentCdn && !isRenderOrPreview) return imageUrl;
|
||||
if (!parsed.search) return imageUrl;
|
||||
// Path IS the stable identity of the attachment; everything after "?" is
|
||||
// signing or a render variant.
|
||||
return `${parsed.origin}${parsed.pathname}`;
|
||||
} catch {
|
||||
return imageUrl;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Lookup a cached media analysis result.
|
||||
* Returns the full cached text (the analysis summary string) or null if not found or expired.
|
||||
@@ -294,10 +328,71 @@ export interface StoredModerationVerdict {
|
||||
recommendedAction: string;
|
||||
}
|
||||
|
||||
/** Raw DB row shape needed to rebuild a StoredModerationVerdict. */
|
||||
interface VerdictRow {
|
||||
flags: string;
|
||||
analyzed_at?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse one `text_analysis_cache` row into a StoredModerationVerdict.
|
||||
* Shared by the single-key and batched getters so their semantics can never
|
||||
* drift apart (status normalization lives in exactly one place).
|
||||
*/
|
||||
export function parseStoredVerdictRow(
|
||||
row: VerdictRow,
|
||||
): StoredModerationVerdict | null {
|
||||
let parsed: Record<string, unknown>;
|
||||
try {
|
||||
parsed = JSON.parse(row.flags) as Record<string, unknown>;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
if (!parsed || typeof parsed !== "object") return null;
|
||||
|
||||
const flags = Array.isArray(parsed.flags) ? (parsed.flags as string[]) : [];
|
||||
const status = normalizeStoredStatus(
|
||||
parsed.status as string | undefined,
|
||||
flags,
|
||||
);
|
||||
return {
|
||||
status,
|
||||
flags,
|
||||
score: (parsed.score as number) ?? 0,
|
||||
analysis: (parsed.analysis as string) ?? "",
|
||||
categories: (parsed.categories as string[]) ?? [],
|
||||
severity: (parsed.severity as string) ?? "none",
|
||||
confidence: (parsed.confidence as number) ?? 0,
|
||||
recommendedAction: (parsed.recommendedAction as string) ?? "none",
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Increment the hit counter for a cache key (fire-and-forget).
|
||||
*
|
||||
* Bug history: `hit_count` was written as 0 on insert and never updated by
|
||||
* any reader, so cache effectiveness was unmeasurable. This is best-effort
|
||||
* observability — a failed bump must never affect the read path.
|
||||
*/
|
||||
function bumpHitCount(cacheKey: string): void {
|
||||
executeAll(
|
||||
`UPDATE text_analysis_cache SET hit_count = hit_count + 1 WHERE text = $1`,
|
||||
[cacheKey],
|
||||
).catch(() => {});
|
||||
}
|
||||
|
||||
/** Error-artifact flags that make a cached verdict unusable. */
|
||||
export const ERROR_ARTIFACT_FLAGS = [
|
||||
"analysis_api_failed",
|
||||
"analysis_parse_failed",
|
||||
"analysis_incomplete",
|
||||
] as const;
|
||||
|
||||
/**
|
||||
* Lookup a cached moderation result for a text content.
|
||||
* Returns the stored result fields or null.
|
||||
*/
|
||||
|
||||
export async function getCachedTextModeration(
|
||||
cacheKey: string,
|
||||
): Promise<StoredModerationVerdict | null> {
|
||||
@@ -311,25 +406,11 @@ export async function getCachedTextModeration(
|
||||
|
||||
if (!row) return null;
|
||||
|
||||
const parsed = JSON.parse(row.flags) as Record<string, unknown>;
|
||||
const flags = (parsed.flags as string[]) ?? [];
|
||||
// Use stored status if available (new entries), otherwise derive from
|
||||
// flags (legacy compatibility). "warn" must survive the round-trip.
|
||||
const status = normalizeStoredStatus(
|
||||
parsed.status as string | undefined,
|
||||
flags,
|
||||
);
|
||||
const verdict = parseStoredVerdictRow(row);
|
||||
if (!verdict) return null;
|
||||
|
||||
return {
|
||||
status,
|
||||
flags,
|
||||
score: (parsed.score as number) ?? 0,
|
||||
analysis: (parsed.analysis as string) ?? "",
|
||||
categories: (parsed.categories as string[]) ?? [],
|
||||
severity: (parsed.severity as string) ?? "none",
|
||||
confidence: (parsed.confidence as number) ?? 0,
|
||||
recommendedAction: (parsed.recommendedAction as string) ?? "none",
|
||||
};
|
||||
bumpHitCount(cacheKey);
|
||||
return verdict;
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
{ error: error instanceof Error ? error.message : String(error) },
|
||||
@@ -339,6 +420,128 @@ export async function getCachedTextModeration(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Batched exact-hash lookup: ONE query for N keys.
|
||||
*
|
||||
* Semantics are identical to calling `getCachedTextModeration` per key
|
||||
* (unexpired rows only, shared row parser). Per-key hit-count bumps are NOT
|
||||
* issued here — the orchestrator logs an aggregate "cache applied" line
|
||||
* instead, keeping a 60-message burst at exactly one round-trip.
|
||||
* `analyzedAt` is surfaced so callers can apply freshness guards.
|
||||
*/
|
||||
export interface BatchedVerdictEntry {
|
||||
verdict: StoredModerationVerdict;
|
||||
analyzedAt: number | null;
|
||||
}
|
||||
|
||||
export async function getCachedTextModerations(
|
||||
cacheKeys: string[],
|
||||
): Promise<Map<string, BatchedVerdictEntry>> {
|
||||
const results = new Map<string, BatchedVerdictEntry>();
|
||||
const uniqueKeys = Array.from(new Set(cacheKeys)).filter(Boolean);
|
||||
if (uniqueKeys.length === 0) return results;
|
||||
|
||||
const CHUNK_SIZE = 200;
|
||||
try {
|
||||
for (let i = 0; i < uniqueKeys.length; i += CHUNK_SIZE) {
|
||||
const chunk = uniqueKeys.slice(i, i + CHUNK_SIZE);
|
||||
// Postgres has a 32k bind-parameter ceiling; ANY($1) keeps it at one
|
||||
// array param per chunk regardless of chunk length.
|
||||
const rows = await executeAll(
|
||||
`SELECT text, flags, analyzed_at
|
||||
FROM text_analysis_cache
|
||||
WHERE text = ANY($1::text[]) AND expires_at > $2`,
|
||||
[chunk, Date.now()],
|
||||
);
|
||||
for (const row of rows ?? []) {
|
||||
if (results.has(row.text)) continue;
|
||||
const verdict = parseStoredVerdictRow(row);
|
||||
if (!verdict) continue;
|
||||
results.set(row.text, {
|
||||
verdict,
|
||||
analyzedAt:
|
||||
typeof row.analyzed_at === "number" ? row.analyzed_at : null,
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
{ error: error instanceof Error ? error.message : String(error) },
|
||||
"Failed batched text moderation lookup",
|
||||
);
|
||||
}
|
||||
return results;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fire-and-forget bulk hit-count bump for keys actually served as hits.
|
||||
* Companion to the batched getter (which skips per-row bumps): one UPDATE
|
||||
* per analysis batch keeps hit-rate metrics working at zero extra latency
|
||||
* cost per message.
|
||||
*/
|
||||
export function bumpTextModerationHitCounts(cacheKeys: string[]): void {
|
||||
const uniqueKeys = Array.from(new Set(cacheKeys)).filter(Boolean);
|
||||
if (uniqueKeys.length === 0) return;
|
||||
executeAll(
|
||||
`UPDATE text_analysis_cache SET hit_count = hit_count + 1 WHERE text = ANY($1::text[])`,
|
||||
[uniqueKeys],
|
||||
).catch(() => {});
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Semantic two-band acceptance
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* True when a semantic-cache hit may be reused given its verdict class.
|
||||
* Two bands (2026-08-24): non-actionable verdicts (clean / flagless /
|
||||
* action=none) are accepted from the LOOSER clean band; actionable verdicts
|
||||
* (warn/flagged or any flags/action) keep the strict historical gate.
|
||||
* Between the bands → reject → the message falls through to the LLM
|
||||
* (fail-open toward accuracy).
|
||||
*/
|
||||
export function isSemanticBandAccepted(
|
||||
verdict: StoredModerationVerdict,
|
||||
similarity: number,
|
||||
): boolean {
|
||||
const isNonActionable =
|
||||
verdict.status === "clean" &&
|
||||
verdict.flags.length === 0 &&
|
||||
(verdict.recommendedAction ?? "none") === "none";
|
||||
return isNonActionable
|
||||
? similarity >= config.AI_LLM_EMBEDDING_MIN_SIMILARITY_CLEAN
|
||||
: similarity >= config.AI_LLM_EMBEDDING_MIN_SIMILARITY;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Global exact-cache reuse guard (context-free fallback)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* True when a stored verdict is safe to reuse OUTSIDE its original channel:
|
||||
* only verdicts that cannot trigger an action and carry no flags qualify,
|
||||
* and they must be confident + fresh. Flagged/warn verdicts are NEVER
|
||||
* globally reused — enforcement is context-sensitive by design.
|
||||
*/
|
||||
export function isGloballyReusableCleanVerdict(
|
||||
verdict: StoredModerationVerdict,
|
||||
analyzedAtMs: number | undefined,
|
||||
): boolean {
|
||||
if (verdict.status !== "clean") return false;
|
||||
if (verdict.flags.length > 0) return false;
|
||||
if ((verdict.recommendedAction ?? "none") !== "none") return false;
|
||||
if (!(verdict.confidence >= config.AI_CACHE_GLOBAL_REUSE_MIN_CONFIDENCE))
|
||||
return false;
|
||||
if (
|
||||
typeof analyzedAtMs === "number" &&
|
||||
Date.now() - analyzedAtMs >
|
||||
config.AI_CACHE_GLOBAL_REUSE_MAX_AGE_H * 60 * 60 * 1000
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a Qdrant verdict payload into the result shape shared by the
|
||||
* semantic cache lookups. Returns null on malformed payloads (callers then
|
||||
@@ -353,31 +556,13 @@ export function parseQdrantVerdict(
|
||||
similarity: number;
|
||||
})
|
||||
| null {
|
||||
let parsed: Record<string, unknown>;
|
||||
try {
|
||||
parsed = JSON.parse(payload.flags) as Record<string, unknown>;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
if (!parsed || typeof parsed !== "object") return null;
|
||||
|
||||
const flags = (parsed.flags as string[]) ?? [];
|
||||
const status = normalizeStoredStatus(
|
||||
parsed.status as string | undefined,
|
||||
flags,
|
||||
);
|
||||
const parsed = parseStoredVerdictRow({ flags: payload.flags });
|
||||
if (!parsed) return null;
|
||||
|
||||
return {
|
||||
...parsed,
|
||||
text: payload.text,
|
||||
similarity,
|
||||
status,
|
||||
flags,
|
||||
score: (parsed.score as number) ?? 0,
|
||||
analysis: (parsed.analysis as string) ?? "",
|
||||
categories: (parsed.categories as string[]) ?? [],
|
||||
severity: (parsed.severity as string) ?? "none",
|
||||
confidence: (parsed.confidence as number) ?? 0,
|
||||
recommendedAction: (parsed.recommendedAction as string) ?? "none",
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { resolve } from "node:dns/promises";
|
||||
import { isIP } from "node:net";
|
||||
import { LRUCache } from "lru-cache";
|
||||
import { createChildLogger } from "@/shared/logger/index";
|
||||
import { createAbortControllerWithTimeout } from "@/shared/utils/index";
|
||||
|
||||
@@ -150,6 +151,47 @@ function truncateAndCleanHtml(html: string, maxLen = 1000): string {
|
||||
export async function fetchUrlSafely(
|
||||
url: string,
|
||||
depth = 0,
|
||||
): Promise<FetchedUrlContext> {
|
||||
// Text results are memoized (in-process, short TTL): the same link recurs
|
||||
// across batches and re-downloading + re-parsing the page each time was
|
||||
// pure latency. Images are NEVER cached here — they are vision evidence
|
||||
// and multi-MB buffers don't belong in an LRU. Errors are not cached so a
|
||||
// transient network blip retries on the next batch.
|
||||
if (depth === 0) {
|
||||
const memo = textFetchMemo.get(url);
|
||||
if (memo) return memo;
|
||||
// In-flight dedupe: concurrent callers share one live request.
|
||||
const existing = textInFlight.get(url);
|
||||
if (existing) return existing;
|
||||
const promise = fetchUrlSafelyUncached(url, depth)
|
||||
.then((fetched) => {
|
||||
if (fetched.type === "text") textFetchMemo.set(url, fetched);
|
||||
return fetched;
|
||||
})
|
||||
.finally(() => {
|
||||
textInFlight.delete(url);
|
||||
});
|
||||
textInFlight.set(url, promise);
|
||||
return promise;
|
||||
}
|
||||
return fetchUrlSafelyUncached(url, depth);
|
||||
}
|
||||
|
||||
/** In-process memo of successful TEXT fetches (30 min TTL, bounded size). */
|
||||
const textFetchMemo = new LRUCache<string, FetchedUrlContext>({
|
||||
max: 500,
|
||||
ttl: 30 * 60 * 1000,
|
||||
});
|
||||
|
||||
/** Concurrent same-URL text fetches collapse into one live request. */
|
||||
const textInFlight = new LRUCache<string, Promise<FetchedUrlContext>>({
|
||||
max: 100,
|
||||
ttl: 60_000,
|
||||
});
|
||||
|
||||
async function fetchUrlSafelyUncached(
|
||||
url: string,
|
||||
depth = 0,
|
||||
): Promise<FetchedUrlContext> {
|
||||
if (depth > 1) {
|
||||
return { url, type: "error", error: "Max redirect/meta depth reached" };
|
||||
|
||||
@@ -21,6 +21,7 @@
|
||||
import { createChildLogger } from "@/shared/logger/index";
|
||||
import { createAbortControllerWithTimeout } from "@/shared/utils/index";
|
||||
import { config } from "../../shared/config/config.js";
|
||||
import { cacheGet, cacheSet, makeCacheKey } from "./cacheStore.js";
|
||||
|
||||
const log = createChildLogger("wikipedia-client");
|
||||
|
||||
@@ -57,10 +58,19 @@ function stripHtml(snippet: string): string {
|
||||
.trim();
|
||||
}
|
||||
|
||||
/** Redis TTL for cached search results (6h — articles change slowly). */
|
||||
const SEARCH_CACHE_TTL_SECONDS = 6 * 60 * 60;
|
||||
|
||||
/**
|
||||
* Search Wikipedia for a query and return up to MAX_RESULTS structured hits.
|
||||
* Uses the Action API `list=search` (srsearch) which is stable and returns
|
||||
* title + HTML snippet. Graceful: returns [] on any failure.
|
||||
*
|
||||
* Cached in the shared Redis store: the same query recurs across batches
|
||||
* (repeat slang, recurring topics), and an uncached re-search per batch was
|
||||
* pure latency + Wikipedia rate-limit pressure. Only NON-EMPTY results are
|
||||
* cached — an empty result may be a transient limiter/network blip, so it is
|
||||
* retried on a later batch instead of being pinned for 6 hours.
|
||||
*/
|
||||
export async function wikipediaSearch(
|
||||
query: string,
|
||||
@@ -69,6 +79,32 @@ export async function wikipediaSearch(
|
||||
const q = query.trim();
|
||||
if (!q) return [];
|
||||
|
||||
const cacheKey = makeCacheKey("wikisearch", q);
|
||||
const cached = await cacheGet(cacheKey);
|
||||
if (cached) {
|
||||
try {
|
||||
const parsed = JSON.parse(cached) as SearchResult[];
|
||||
if (Array.isArray(parsed) && parsed.length > 0) {
|
||||
log.debug({ query: q }, "Wikipedia search cache HIT");
|
||||
return parsed;
|
||||
}
|
||||
} catch {
|
||||
// Malformed entry — fall through to live fetch.
|
||||
}
|
||||
}
|
||||
|
||||
const mapped = await wikipediaSearchLive(q, timeoutMs);
|
||||
if (mapped.length > 0) {
|
||||
cacheSet(cacheKey, JSON.stringify(mapped), SEARCH_CACHE_TTL_SECONDS);
|
||||
}
|
||||
return mapped;
|
||||
}
|
||||
|
||||
/** Live (uncached) Action API search. Returns [] on any failure. */
|
||||
async function wikipediaSearchLive(
|
||||
q: string,
|
||||
timeoutMs: number,
|
||||
): Promise<SearchResult[]> {
|
||||
const params = new URLSearchParams({
|
||||
action: "query",
|
||||
list: "search",
|
||||
|
||||
Reference in New Issue
Block a user