feat(automod): store semantic cache embeddings in Qdrant
Build & Deploy (Nix) / build-and-deploy (backend) (push) Successful in 3m7s
Build & Deploy (Nix) / build-and-deploy (discord-gateway) (push) Successful in 2m21s
Build & Deploy (Nix) / build-and-deploy (proxy) (push) Successful in 2m33s
Build & Deploy (Nix) / build-and-deploy (backend) (push) Successful in 3m7s
Build & Deploy (Nix) / build-and-deploy (discord-gateway) (push) Successful in 2m21s
Build & Deploy (Nix) / build-and-deploy (proxy) (push) Successful in 2m33s
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.
This commit is contained in:
@@ -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<string, string> {
|
||||
const h: Record<string, string> = {
|
||||
"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<unknown> {
|
||||
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<boolean> {
|
||||
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<boolean> {
|
||||
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<QdrantSearchHit[]> {
|
||||
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);
|
||||
}
|
||||
@@ -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<string, unknown>;
|
||||
try {
|
||||
parsed = JSON.parse(hit.payload.flags) as Record<string, unknown>;
|
||||
} 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,
|
||||
],
|
||||
);
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user