From fc475dfbb7743752185433b94164c985b5311093 Mon Sep 17 00:00:00 2001 From: Developer Date: Fri, 31 Jul 2026 21:30:43 +0700 Subject: [PATCH] feat(automod): store semantic cache embeddings in Qdrant New qdrantClient.ts (zero-dep fetch REST): ensure collection with cosine distance (auto-recreate on vector-size change), upsert point w/ verdict payload, search w/ expires_at filter + score threshold. textCacheStore: when QDRANT_URL set, embeddings are upserted to Qdrant (primary) and searched there first; Postgres embedding column remains as legacy fallback for pre-Qdrant rows. Config: QDRANT_URL/COLLECTION/API_KEY. QDRANT_URL already in repo .env; added to VPS env + GATEWAY_ENV secret. --- .env.example | 3 + .../src/modules/ai-moderation/qdrantClient.ts | 221 ++++++++++++++++++ .../modules/ai-moderation/textCacheStore.ts | 56 ++++- .../src/shared/config/index.ts | 6 + 4 files changed, 282 insertions(+), 4 deletions(-) create mode 100644 services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts diff --git a/.env.example b/.env.example index f81a827..d46cff5 100644 --- a/.env.example +++ b/.env.example @@ -90,6 +90,9 @@ AI_LLM_MODEL=text # LLM text model name (default: text) # AI_LLM_VISION_MODEL= # Vision model for image analysis (falls back to AI_LLM_MODEL) # AI_LLM_EMBEDDING_MODEL= # Embedding model for semantic moderation cache (optional; enables near-duplicate text reuse to save LLM calls) # AI_LLM_EMBEDDING_MIN_SIMILARITY=0.97 # Min cosine similarity to reuse a cached verdict (default: 0.97) +# QDRANT_URL=http://100.121.180.82:6333/ # Qdrant vector store for embeddings (semantic cache); when set, vectors are stored/searched in Qdrant instead of Postgres +# QDRANT_COLLECTION=gmw_text_moderation # Qdrant collection name (default: gmw_text_moderation) +# QDRANT_API_KEY= # Qdrant API key (optional) AI_LLM_MAX_CONCURRENT=5 # Max concurrent LLM API calls (default: 5) AI_LLM_IMAGE_MAX_DIMENSION=1024 # Max image dimension in pixels before resize (default: 1024) AI_LLM_TEXT_BATCH_SIZE=20 # Max messages per text-only moderation batch (default: 20) diff --git a/services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts b/services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts new file mode 100644 index 0000000..02559e9 --- /dev/null +++ b/services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts @@ -0,0 +1,221 @@ +/** + * qdrantClient.ts + * + * Minimal Qdrant REST client (zero dependencies, fetch-based) used by the + * semantic moderation cache. Embedding vectors + verdict payloads live in + * Qdrant instead of the Postgres `embedding` column (legacy, kept for + * backward-compatible fallback reads). + * + * All functions degrade gracefully: failures return null / empty results so + * callers fall back to the LLM — moderation quality is never reduced. + */ + +import { createHash } from "node:crypto"; + +import { createChildLogger } from "@/shared/logger/index"; +import { config } from "../../shared/config/config.js"; + +const log = createChildLogger("qdrant"); + +export interface QdrantVerdictPayload { + text: string; + flags: string; // JSON string of the full moderation result + analyzed_at: number; + expires_at: number; +} + +function baseUrl(): string { + return (config.QDRANT_URL ?? "http://100.121.180.82:6333").replace( + /\/+$/, + "", + ); +} + +function collectionName(): string { + return config.QDRANT_COLLECTION ?? "gmw_text_moderation"; +} + +function headers(): Record { + const h: Record = { + "Content-Type": "application/json", + }; + if (config.QDRANT_API_KEY) { + h["api-key"] = config.QDRANT_API_KEY; + } + return h; +} + +async function request( + method: string, + path: string, + body?: unknown, + timeoutMs = 10_000, +): Promise { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + try { + const res = await fetch(`${baseUrl()}${path}`, { + method, + headers: headers(), + body: body === undefined ? undefined : JSON.stringify(body), + signal: controller.signal, + }); + const text = await res.text(); + let json: unknown = null; + try { + json = text ? JSON.parse(text) : null; + } catch { + json = null; + } + if (!res.ok) { + throw new Error( + `Qdrant ${method} ${path} -> ${res.status}: ${text.slice(0, 200)}`, + ); + } + return json; + } finally { + clearTimeout(timer); + } +} + +/** Deterministic uint64 point id from the exact-hash cache key. */ +export function qdrantPointId(cacheKey: string): number { + const digest = createHash("sha256").update(cacheKey).digest(); + // First 8 bytes as BigInt, then clamp into Qdrant's uint64 space. + const big = digest.readBigUInt64BE(0); + return Number(big & 0x7fffffffffffffffn); +} + +/** + * Ensure the collection exists with the right vector size. If the size + * changed (embedding model swapped), recreate — stale vectors are useless + * anyway and cosine scores would be meaningless across dimensions. + */ +export async function ensureQdrantCollection( + vectorSize: number, +): Promise { + try { + // 404 = collection doesn't exist yet → create it. + let existing: { + result?: { config?: { params?: { vectors?: { size?: number } } } }; + } | null = null; + try { + existing = (await request("GET", `/collections/${collectionName()}`)) as { + result?: { config?: { params?: { vectors?: { size?: number } } } }; + }; + } catch (error) { + if (!(error instanceof Error) || !error.message.includes("-> 404")) { + throw error; + } + } + + const size = existing?.result?.config?.params?.vectors?.size; + if (size === vectorSize) return true; + + if (size !== undefined && size !== vectorSize) { + log.warn( + { collection: collectionName(), oldSize: size, newSize: vectorSize }, + "Qdrant collection vector size changed — recreating collection", + ); + await request("DELETE", `/collections/${collectionName()}`); + } + + await request("PUT", `/collections/${collectionName()}`, { + vectors: { size: vectorSize, distance: "Cosine" }, + }); + return true; + } catch (error) { + log.error( + { + error: error instanceof Error ? error.message : String(error), + collection: collectionName(), + }, + "Failed to ensure Qdrant collection", + ); + return false; + } +} + +/** Upsert one embedding + verdict payload point. Returns false on failure. */ +export async function upsertQdrantPoint( + cacheKey: string, + vector: number[], + payload: QdrantVerdictPayload, +): Promise { + try { + if (!(await ensureQdrantCollection(vector.length))) return false; + await request("PUT", `/collections/${collectionName()}/points`, { + points: [{ id: qdrantPointId(cacheKey), vector, payload }], + wait: true, + }); + return true; + } catch (error) { + log.warn( + { error: error instanceof Error ? error.message : String(error) }, + "Qdrant upsert failed — semantic entry skipped", + ); + return false; + } +} + +export interface QdrantSearchHit { + cacheKey: string; + score: number; + payload: QdrantVerdictPayload; +} + +/** + * Search the nearest stored vector. Returns hits sorted by score desc, + * filtered to unexpired payloads. Empty array on failure. + */ +export async function searchQdrant( + vector: number[], + limit: number, + scoreThreshold: number, +): Promise { + try { + const json = (await request( + "POST", + `/collections/${collectionName()}/points/search`, + { + vector, + limit, + score_threshold: scoreThreshold, + with_payload: true, + filter: { + must: [ + { + key: "expires_at", + range: { gte: Date.now() }, + }, + ], + }, + }, + )) as { + result?: Array<{ + id?: number; + score?: number; + payload?: QdrantVerdictPayload; + }>; + }; + + return (json.result ?? []) + .filter((hit) => hit.payload?.flags) + .map((hit) => ({ + cacheKey: `qdrant:${hit.id ?? "?"}`, + score: hit.score ?? 0, + payload: hit.payload as QdrantVerdictPayload, + })); + } catch (error) { + log.warn( + { error: error instanceof Error ? error.message : String(error) }, + "Qdrant search failed — semantic cache skipped", + ); + return []; + } +} + +/** True when Qdrant is configured (non-empty URL). */ +export function isQdrantConfigured(): boolean { + return Boolean(config.QDRANT_URL); +} diff --git a/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts b/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts index 504a333..c580bae 100644 --- a/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts +++ b/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts @@ -2,6 +2,11 @@ import { createHash } from "node:crypto"; import { createChildLogger } from "@/shared/logger/index"; import { executeAll, executeGet } from "../../shared/database/drizzle.js"; import { findBestEmbeddingMatch } from "./embeddingClient.js"; +import { + isQdrantConfigured, + searchQdrant, + upsertQdrantPoint, +} from "./qdrantClient.js"; const logger = createChildLogger("text-cache-store"); @@ -358,10 +363,9 @@ export async function getCachedTextModeration(cacheKey: string): Promise<{ /** * Semantic moderation cache lookup. * - * Returns the most similar stored verdict whose cosine similarity to the - * query embedding is at least `minSimilarity`. Only entries written by the - * moderation pipeline (source='user_moderation') with a stored embedding are - * considered, limited to the most recent `limit` rows to bound cost. + * Primary: Qdrant vector search (when QDRANT_URL configured) — nearest + * unexpired verdict above `minSimilarity`. Fallback: Postgres embedding + * column (legacy rows written before Qdrant was wired in). * Returns null on no match or any failure — callers then proceed to the LLM. */ export async function findSimilarTextModeration( @@ -380,6 +384,38 @@ export async function findSimilarTextModeration( confidence: number; recommendedAction: string; } | null> { + // Qdrant path (primary) + if (isQdrantConfigured()) { + const hits = await searchQdrant(embedding, limit, minSimilarity); + if (hits.length > 0) { + const hit = hits[0]; + let parsed: Record; + try { + parsed = JSON.parse(hit.payload.flags) as Record; + } catch { + return null; + } + const storedStatus = (parsed.status as string) ?? "clean"; + const status: "clean" | "warn" | "flagged" = + storedStatus === "warn" || storedStatus === "flagged" + ? storedStatus + : "clean"; + return { + text: hit.payload.text, + similarity: hit.score, + status, + flags: (parsed.flags as string[]) ?? [], + 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", + }; + } + // No Qdrant hit — fall through to Postgres legacy rows. + } + try { const rows = await executeAll( `SELECT text, flags, embedding @@ -477,6 +513,17 @@ export async function setCachedTextModeration( const USER_MOD_CACHE_TTL_MS = 24 * 60 * 60 * 1000; try { + // Qdrant is the primary vector store when configured: upsert the point + // with the verdict payload; skip the Postgres embedding column entirely. + if (isQdrantConfigured() && embedding && embedding.length > 0) { + await upsertQdrantPoint(cacheKey, embedding, { + text: cacheKey, + flags: JSON.stringify(result), + analyzed_at: now, + expires_at: now + USER_MOD_CACHE_TTL_MS, + }); + } + await executeAll( `INSERT INTO text_analysis_cache (text, flags, source, analyzed_at, expires_at, hit_count, embedding) VALUES ($1, $2, $3, $4, $5, 0, $6) @@ -492,6 +539,7 @@ export async function setCachedTextModeration( "user_moderation", now, now + USER_MOD_CACHE_TTL_MS, + // Postgres embedding stays as legacy fallback; Qdrant is primary. embedding && embedding.length > 0 ? JSON.stringify(embedding) : null, ], ); diff --git a/services/discord-gateway/src/shared/config/index.ts b/services/discord-gateway/src/shared/config/index.ts index 937cc17..07ccf12 100644 --- a/services/discord-gateway/src/shared/config/index.ts +++ b/services/discord-gateway/src/shared/config/index.ts @@ -145,6 +145,12 @@ export const configSchema = z .int() .positive() .default(30), + // Qdrant vector store for the semantic moderation cache. When + // QDRANT_URL is set, embeddings are stored/searched there (Postgres + // embedding column remains as a legacy fallback). + QDRANT_URL: z.string().optional(), + QDRANT_COLLECTION: z.string().default("gmw_text_moderation"), + QDRANT_API_KEY: z.string().optional(), AI_LLM_MAX_CONCURRENT: z.coerce.number().int().positive().default(5), AI_LLM_IMAGE_MAX_DIMENSION: z.coerce .number()