diff --git a/dist/dsh.js b/dist/dsh.js index 4b0fdad..9787b0a 100644 --- a/dist/dsh.js +++ b/dist/dsh.js @@ -9,10 +9,11 @@ import { randomUUID } from "node:crypto"; import { dshMemorySource, isDshMemorySource } from "./src/format/dsh-source.js"; import { DshExtractionUnavailableError, resolveDshExtractionReasoning, } from "./src/engine/dsh-extraction-route.js"; import { openDb } from "./src/store/db.js"; -import { allActiveNodes, getRecentTurnMemoriesBySession, getStats, getVectorStats, getNextUnextractedTurn, getUnextractedTurn, getExtractionStats, getPendingSessionIds, getExtractionCompletedTurn, getNodeSources, markMessagesExtracted, markExtractionTurnCompleted, quarantineMessages, recordExtractionFailure, requeueQuarantined, saveMessageOnce, upsertNode, upsertTurnMemory, replaceNavigationTriples, } from "./src/store/store.js"; +import { allActiveNodes, getRecentTurnMemoriesBySession, getStats, getVectorStats, getNextUnextractedTurn, getUnextractedTurn, getExtractionStats, getPendingSessionIds, getExtractionCompletedTurn, getNodeSources, markMessagesExtracted, markExtractionTurnCompleted, quarantineMessages, recordExtractionFailure, requeueQuarantined, saveMessageOnce, upsertNode, upsertTurnMemory, replaceNavigationTriples, applyConceptExtraction, } from "./src/store/store.js"; import { Extractor } from "./src/extractor/extract.js"; import { GRAPH_EXTRACTION_TOOL, GRAPH_EXTRACTION_TOOL_NAME, } from "./src/extractor/contract.js"; import { Recaller } from "./src/recaller/recall.js"; +import { createHttpReranker } from "./src/recaller/rerank.js"; import { assembleContext } from "./src/format/assemble.js"; import { replaceDshArchivedPrefix, selectDshRollingCompactionRange, } from "./src/format/dsh-compaction.js"; import { replaceDshCompletedTurnTrace, projectDshCompletedTurnMemory, selectDshCompletedTurnTraceRange, } from "./src/format/dsh-turn-projection.js"; @@ -25,6 +26,13 @@ import { messageRetentionPolicyRevision, normalizeMessageRetentionPolicy, runMes export const name = "graph-memory-dsh"; export const inject = ["tools", "llm", "systemPrompt", "agentLoop", "agents", "sessions", "credentials", "tokenMeter"]; const HOST = "dsh"; +/** + * Retrieval pool multiplier applied before the visibility filter. Five slots + * per rendered slot is enough to backfill a block that a same-session answer + * would otherwise empty, while costing nothing measurable: the vector path is + * already a full scan and the lexical path is a single indexed LIKE. + */ +const RECALL_OVER_FETCH = 5; function sessionKey(id) { return `${HOST}:${String(id)}`; } @@ -111,12 +119,20 @@ export function apply(ctx, input = {}) { compactTurnCount: maintenanceInterval, recallMaxNodes, semanticScoreThreshold: input.semanticScoreThreshold ?? DEFAULT_CONFIG.semanticScoreThreshold, + rerankTopN: input.rerankTopN ?? 6, embedding, }; const extractionEnabled = input.extractionEnabled ?? true; const recallEnabled = input.recallEnabled ?? true; const db = openDb(config.dbPath, { busyTimeoutMs: input.dbBusyTimeoutMs }); const recaller = new Recaller(db, config); + if (input.rerankEndpoint) { + recaller.setRerankFn(createHttpReranker({ + endpoint: input.rerankEndpoint, + model: input.rerankModel, + timeoutMs: input.rerankTimeoutMs ?? 1500, + })); + } const latestRoute = new Map(); const extractChain = new Map(); const turnCounts = new Map(); @@ -281,6 +297,13 @@ export function apply(ctx, input = {}) { })), }); replaceNavigationTriples(db, turnMemory, result.triples); + // The concept layer rides on the same extraction payload. Writing it here + // is what keeps the durable graph current: without this step the host only + // ever stored navigation, so every concept had to be recorded by hand. + const concept = applyConceptExtraction(db, sid, turnMemory.sources, result.entities ?? [], result.relations ?? []); + if (concept.nodes || concept.edges) { + ctx.logger.info(`[graph-memory] DSH stored ${concept.nodes} concept nodes and ${concept.edges} concept edges`); + } // The navigation graph becomes queryable only after its atomic SPO write. // This runs inside the existing background extraction worker, never in the // foreground turn, and makes the newest completed memory available to PPR. @@ -539,7 +562,13 @@ export function apply(ctx, input = {}) { // is still in flight. Historical recall must wait for that shared probe; // otherwise the very first cross-session question can miss all vectors. await embeddingReady; - const recalled = await recaller.recall(query); + // Retrieval runs on a wider pool than the rendered width. The filters + // below drop memory already visible verbatim in this session's fresh + // window, and a pool sized exactly to the render width could be emptied + // entirely by that filter — a short follow-up whose top hits all came + // from the live session then injected nothing at all. Over-fetching lets + // the block backfill from cross-session or archived evidence instead. + const recalled = await recaller.recall(query, { overFetch: RECALL_OVER_FETCH }); signal?.throwIfAborted?.(); const key = String(id); const currentSession = sessionKey(id); @@ -555,8 +584,8 @@ export function apply(ctx, input = {}) { && isDshMemorySource(event?.data?.source) && event?.surfaceOp?.op === "replace"; }); - const recalledNodes = filterDshRecallNodes(recalled.nodes, getNodeSources(db, recalled.nodes.map(node => node.id)), currentSession, visibleMessageIds, hasArchivedHistory); - const recalledMemories = filterDshRecallMemories(recalled.turnMemories, currentSession, visibleMessageIds); + const recalledNodes = filterDshRecallNodes(recalled.nodes, getNodeSources(db, recalled.nodes.map(node => node.id)), currentSession, visibleMessageIds, hasArchivedHistory).slice(0, config.recallMaxNodes); + const recalledMemories = filterDshRecallMemories(recalled.turnMemories, currentSession, visibleMessageIds).slice(0, config.recallMaxNodes); if (!recalledNodes.length && !recalledMemories.length) return decision; const recalledIds = new Set(recalledNodes.map(node => node.id)); diff --git a/dist/src/consolidate/cluster.js b/dist/src/consolidate/cluster.js new file mode 100644 index 0000000..77811a9 --- /dev/null +++ b/dist/src/consolidate/cluster.js @@ -0,0 +1,156 @@ +/** + * Clustering primitives used by memory consolidation. + * + * Adapted from slow-stack/mneme (MIT License) — `dsh-mneme/src/dream/clustering.js`, + * https://github.com/slow-stack/mneme. The algorithms are reproduced unchanged; + * only types and the grouping key are parameterized so they fit this store's + * turn-memory shape. Original copyright (c) slow-stack/mneme contributors. + * + * Kept as a standalone module: nothing here touches the database, so it can be + * exercised and tested without a store, and the consolidation layer can be + * dropped without leaving a mark on recall or extraction. + */ +/** + * Cosine similarity of two equal-length vectors. Returns 0 for empty input, + * mismatched lengths, or a zero-norm vector (numerically stable rather than + * NaN, so a degenerate embedding cannot poison a whole run). + */ +export function cosineSimilarity(a, b) { + if (!a || !b || a.length === 0 || a.length !== b.length) + return 0; + let dot = 0; + let na = 0; + let nb = 0; + for (let i = 0; i < a.length; i += 1) { + dot += a[i] * b[i]; + na += a[i] * a[i]; + nb += b[i] * b[i]; + } + if (na === 0 || nb === 0) + return 0; + return dot / (Math.sqrt(na) * Math.sqrt(nb)); +} +/** + * K-Means over numeric vectors, returning one index array per cluster. + * + * k-means++ seeding, and empty clusters are reseeded mid-iteration so a bad + * draw cannot collapse the result to fewer clusters than requested. + */ +export function kMeans(vectors, k, opts = {}) { + const n = vectors.length; + if (n === 0) + return []; + const { maxIter = 100, tol = 1e-4, random = Math.random } = opts; + if (k <= 0 || k >= n) + return [vectors.map((_, index) => index)]; + const dim = vectors[0].length; + const randomIndex = () => Math.floor(random() * n); + const squaredDist = (v, c) => { + let sum = 0; + for (let d = 0; d < dim; d += 1) + sum += (v[d] - c[d]) * (v[d] - c[d]); + return sum; + }; + // First centroid uniform, later ones sampled proportional to squared distance. + const centroids = [Float32Array.from(vectors[randomIndex()])]; + while (centroids.length < k) { + const dist = new Array(n); + let sum = 0; + for (let i = 0; i < n; i += 1) { + let best = Infinity; + for (const centroid of centroids) + best = Math.min(best, squaredDist(vectors[i], centroid)); + dist[i] = best; + sum += best; + } + if (sum === 0) { + centroids.push(Float32Array.from(vectors[randomIndex()])); + } + else { + let r = random() * sum; + let pick = 0; + for (let i = 0; i < n; i += 1) { + r -= dist[i]; + if (r <= 0) { + pick = i; + break; + } + } + centroids.push(Float32Array.from(vectors[pick])); + } + } + const assignment = new Array(n).fill(0); + for (let iter = 0; iter < maxIter; iter += 1) { + let changed = false; + for (let i = 0; i < n; i += 1) { + let best = 0; + let bestD = Infinity; + for (let j = 0; j < k; j += 1) { + const d = squaredDist(vectors[i], centroids[j]); + if (d < bestD) { + bestD = d; + best = j; + } + } + if (assignment[i] !== best) { + assignment[i] = best; + changed = true; + } + } + if (!changed) + break; + const sums = Array.from({ length: k }, () => new Float32Array(dim)); + const counts = new Array(k).fill(0); + for (let i = 0; i < n; i += 1) { + const cluster = assignment[i]; + counts[cluster] += 1; + for (let d = 0; d < dim; d += 1) + sums[cluster][d] += vectors[i][d]; + } + let maxMove = 0; + for (let j = 0; j < k; j += 1) { + if (counts[j] === 0) { + centroids[j] = Float32Array.from(vectors[randomIndex()]); + maxMove = Infinity; + continue; + } + for (let d = 0; d < dim; d += 1) { + const mean = sums[j][d] / counts[j]; + const move = mean - centroids[j][d]; + if (move * move > maxMove) + maxMove = move * move; + centroids[j][d] = mean; + } + } + if (Math.sqrt(maxMove) < tol) + break; + } + const clusters = Array.from({ length: k }, () => []); + for (let i = 0; i < n; i += 1) + clusters[assignment[i]].push(i); + return clusters.filter(cluster => cluster.length > 0); +} +/** Cluster items by their vectors, preserving the original object references. */ +export function clusterByVector(items, vectors, k) { + return kMeans(vectors, k).map(cluster => cluster.map(index => items[index])); +} +/** + * Pairs that are highly similar yet might contradict each other. + * + * Similarity is only a candidate signal: `0.9` means "read these two together", + * never "these two disagree". Only items sharing the same group key are paired, + * because a merge or conflict between unrelated kinds of memory is meaningless. + */ +export function findPotentialConflicts(items, vectors, keyOf, threshold = 0.85) { + const pairs = []; + for (let i = 0; i < items.length; i += 1) { + for (let j = i + 1; j < items.length; j += 1) { + if (keyOf(items[i]) !== keyOf(items[j])) + continue; + const similarity = cosineSimilarity(vectors[i], vectors[j]); + if (similarity > threshold) + pairs.push({ a: items[i], b: items[j], similarity }); + } + } + return pairs; +} diff --git a/dist/src/extractor/contract.js b/dist/src/extractor/contract.js index 24b9694..1df632d 100644 --- a/dist/src/extractor/contract.js +++ b/dist/src/extractor/contract.js @@ -9,6 +9,30 @@ const NavigationTripleSchema = Type.Object({ predicate: Type.String({ minLength: 1, description: "Short relation phrase." }), object: Type.String({ minLength: 1, description: "Concrete searchable object phrase." }), }, { additionalProperties: false }); +const ConceptEntitySchema = Type.Object({ + name: Type.String({ + minLength: 1, + maxLength: 24, + description: "Reusable noun phrase naming one concept; never a sentence, a status word, or a bare quantity.", + }), + type: Type.String({ + enum: ["EVENT", "SKILL", "TASK"], + description: "Concept kind: EVENT happened, SKILL reusable capability, TASK goal or pending work.", + }), + description: Type.String({ + minLength: 1, + description: "One sentence stating what this concept is, using only evidence from this turn.", + }), +}, { additionalProperties: false }); +const ConceptRelationSchema = Type.Object({ + from: Type.String({ minLength: 1, description: "Subject name; must equal one entities[].name value." }), + to: Type.String({ minLength: 1, description: "Object name; must equal one entities[].name value." }), + type: Type.String({ + enum: ["RELATES", "SUPERSEDES", "USED_SKILL", "SOLVED_BY", "REQUIRES", "PATCHES", "CONFLICTS_WITH"], + description: "Declared relation kind between the two concepts.", + }), + instruction: Type.String({ minLength: 1, description: "Short natural-language statement of the relation." }), +}, { additionalProperties: false }); /** * Provider-facing extraction contract. * @@ -27,11 +51,17 @@ export const GRAPH_EXTRACTION_SCHEMA = Type.Object({ triples: Type.Array(NavigationTripleSchema, { description: "Required SPO relations derived only from summary; use an empty array when none are explicit.", }), + entities: Type.Optional(Type.Array(ConceptEntitySchema, { + description: "Optional durable concept list for this turn; omit or use [] when the turn yields no reusable concept.", + })), + relations: Type.Optional(Type.Array(ConceptRelationSchema, { + description: "Optional typed links between entries of entities; both endpoints must be declared concepts.", + })), }, { additionalProperties: false }); export const GRAPH_EXTRACTION_TOOL_NAME = "submit_result"; export const GRAPH_EXTRACTION_TOOL = Object.freeze({ name: GRAPH_EXTRACTION_TOOL_NAME, - description: "Submit exactly the three required fields summary, outcome, and triples. Derive zero or more subject-predicate-object triples only from the summary; triples must be [] when no relation is explicit. Emit no text.", + description: "Submit exactly the three required fields summary, outcome, and triples. Derive zero or more subject-predicate-object triples only from the summary; triples must be [] when no relation is explicit. Optionally add entities and relations when the turn names reusable concepts. Emit no text.", parameters: GRAPH_EXTRACTION_SCHEMA, }); /** Fail closed before normalization or persistence when the contract is incomplete. */ diff --git a/dist/src/extractor/extract.js b/dist/src/extractor/extract.js index c327d24..3fa6baa 100644 --- a/dist/src/extractor/extract.js +++ b/dist/src/extractor/extract.js @@ -41,9 +41,11 @@ const EXTRACT_SYS = `【任务】 1. summary 用一句简短、自包含的话写清本轮对象、结论和最终回答明确报告的完成情况;不复述过程,不添加输入未表达的信息。 2. outcome 按最终回答选择 completed、partial、failed、informational、unknown 之一。 3. triples 只从 summary 拆分。subject 和 object 使用具体可检索短语,predicate 使用简短自然语言;没有明确关系时使用空数组,不补充、不猜测。 +4. entities 可选,用于记录本轮出现的、以后还会被问到的概念。每个概念给三样:name 是 2~16 字的名词短语(禁止整句,禁止“全部正常 / 已完成 / 72 个”这类状态与数量);type 取 EVENT(发生过的事)、SKILL(可复用的工具、技能、脚本、服务、配置)、TASK(目标、待办、缺口);description 用一句话说明它是什么,只用本轮信息。本轮没有可复用概念时省略该字段。 +5. relations 可选,只在确有明确关系时输出:from 与 to 必须是 entities 里出现过的 name,type 取 RELATES、SUPERSEDES、USED_SKILL、SOLVED_BY、REQUIRES、PATCHES、CONFLICTS_WITH 之一,instruction 用一句中文说明关系。没有关系时省略该字段。 【输出合同】 -只调用 submit_result 一次。参数对象必须且只能包含 summary、outcome、triples 三个顶层字段,三个字段都不能省略。不要输出解释或正文。 +只调用 submit_result 一次。参数对象必须包含 summary、outcome、triples 三个顶层字段,三个字段都不能省略;可选包含 entities 与 relations。不要输出解释或正文。 示例一: 输入:用户要求把周会改到周四;最终回答确认日程已更新。 @@ -57,7 +59,11 @@ const EXTRACT_SYS = `【任务】 输入只确认稍后继续讨论,没有新结论。 输出:{"summary":"本轮确认稍后继续讨论,未产生新结论。","outcome":"informational","triples":[]} -调用前自检:参数是否恰好包含三个顶层字段;outcome 是否属于枚举;没有关系时 triples 是否仍明确写为 []。`; +示例四(带概念): +输入:助手把抽取路由钉到本地 27B,避免打到线上账号。 +输出:{"summary":"抽取路由已改为本地 27B,不再使用线上账号。","outcome":"completed","triples":[{"subject":"抽取路由","predicate":"改为","object":"本地 27B"}],"entities":[{"name":"抽取路由","type":"SKILL","description":"抽取请求使用的模型路由配置"},{"name":"本地 27B","type":"SKILL","description":"本机 llama-swap 上的主模型"},{"name":"线上账号","type":"EVENT","description":"此前误用线上 API 的问题"}],"relations":[{"from":"抽取路由","to":"本地 27B","type":"REQUIRES","instruction":"路由改为依赖本地模型"},{"from":"抽取路由","to":"线上账号","type":"SOLVED_BY","instruction":"改路由后不再走线上账号"}]} + +调用前自检:是否包含 summary、outcome、triples 三个必需字段;outcome 是否属于枚举;没有关系时 triples 是否仍明确写为 [];若输出 entities 与 relations,relation 两端是否都能在 entities 的 name 中找到,且 type 是否在枚举内。`; // ─── 提取 User Prompt ─────────────────────────────────────────── const EXTRACT_USER = (msgs, priorTurns) => ` ${priorTurns.length ? JSON.stringify(priorTurns.map(memory => memory.summary)) : "(无)"} @@ -94,6 +100,11 @@ export class Extractor { return { turn: { summary: p.summary, outcome: p.outcome }, triples: p.triples.map(triple => ({ ...triple })), + // Optional blocks are copied verbatim and default to empty. The contract + // decides whether they are well formed; a host that does not persist + // concepts simply ignores them, so the field stays additive. + entities: (p.entities ?? []).map(entity => ({ ...entity })), + relations: (p.relations ?? []).map(relation => ({ ...relation })), }; } catch (err) { diff --git a/dist/src/recaller/recall.js b/dist/src/recaller/recall.js index 6c80799..dabe582 100644 --- a/dist/src/recaller/recall.js +++ b/dist/src/recaller/recall.js @@ -12,10 +12,28 @@ import { createHash } from "crypto"; import { searchNodes, vectorSearchWithScore, graphWalk, saveVector, getVectorHash, searchTurnMemories, turnMemoryVectorSearchWithScore, nodesForTurnMemories, saveTurnVector, getTurnVectorHash, getNavigationTriplesForMemories, findNavigationSeedTermIds, navigationCandidateTermIds, rankTurnMemoryIdsByNavigation, getTurnMemoriesByIds, hasTurnMemories, } from "../store/store.js"; import { personalizedNavigationPageRank } from "../graph/pagerank.js"; +/** + * Recency is a tie-breaker, not a relevance signal. A memory that is slightly + * less similar but four months older should not outrank the one that describes + * what the user is doing this week, yet it must not jump ahead of a clearly + * better match either. A 30-day half life with a 0.15 ceiling sits below one + * reciprocal-rank step (1/1 - 1/2 = 0.5), so it can reorder near-ties without + * overriding the similarity ranks that the embedding provider produced. + */ +const RECENCY_HALF_LIFE_MS = 30 * 24 * 60 * 60 * 1000; +const RECENCY_WEIGHT = 0.15; +/** 1 for a memory written now, 0.5 at thirty days, 0.25 at sixty. */ +function recencyOf(memory, now) { + const updatedAt = Number(memory.updatedAt); + if (!Number.isFinite(updatedAt) || updatedAt <= 0) + return 0; + return Math.pow(0.5, Math.max(0, now - updatedAt) / RECENCY_HALF_LIFE_MS); +} export class Recaller { db; cfg; embed = null; + rerank = null; embeddingFingerprint = ""; constructor(db, cfg) { this.db = db; @@ -25,8 +43,26 @@ export class Recaller { this.embed = fn; this.embeddingFingerprint = fingerprint; } - async recall(query) { + /** + * Optional cross-encoder pass over the candidate head. Absent by default: a + * host that does not configure a reranker keeps the embedding order exactly. + */ + setRerankFn(fn) { + this.rerank = fn; + } + /** + * @param options.overFetch Retrieval pool multiplier. Defaults to 1, which + * keeps the historical behaviour: `recallMaxNodes` bounds both retrieval and + * what the caller may render. A host that post-filters results (for example + * to drop memories already visible in its fresh window) passes a multiplier + * so that filtering cannot empty the whole block: a three-slot pool where all + * three hits belong to the current session used to yield nothing at all, and + * the caller then injected no memory for that turn. + */ + async recall(query, options = {}) { const limit = this.cfg.recallMaxNodes; + const overFetch = Math.max(1, Math.floor(options.overFetch ?? 1)); + const poolLimit = limit * overFetch; let queryVector; if (this.embed) { try { @@ -37,7 +73,7 @@ export class Recaller { // temporarily unavailable. } } - const directMemories = this.recallTurnMemories(query, limit, queryVector); + const directMemories = this.recallTurnMemories(query, poolLimit, queryVector); const seedIds = findNavigationSeedTermIds(this.db, query, directMemories.map(memory => memory.id)); let graphMemories = []; let navigationScores = new Map(); @@ -46,10 +82,11 @@ export class Recaller { navigationScores = personalizedNavigationPageRank(this.db, seedIds, candidateTermIds, this.cfg).scores; graphMemories = getTurnMemoriesByIds(this.db, rankTurnMemoryIdsByNavigation(this.db, navigationScores)); } - const turnMemories = this.mergeTurnMemoryRanks(directMemories, graphMemories, limit); + let turnMemories = this.mergeTurnMemoryRanks(directMemories, graphMemories, poolLimit); + turnMemories = await this.rerankMemories(query, turnMemories); if (turnMemories.length) { const memoryIds = turnMemories.map(memory => memory.id); - const nodes = nodesForTurnMemories(this.db, memoryIds, limit); + const nodes = nodesForTurnMemories(this.db, memoryIds, poolLimit); const { edges } = graphWalk(this.db, nodes.map(node => node.id), 0); return { nodes, @@ -60,7 +97,7 @@ export class Recaller { } // Databases created before the turn-memory migration remain searchable. // The same confidence rule applies; a legacy node match cannot bypass it. - return this.recallPrecise(query, limit, queryVector, hasTurnMemories(this.db)); + return this.recallPrecise(query, poolLimit, queryVector, hasTurnMemories(this.db)); } /** * Fuse independent summary and graph ranks without mixing incomparable @@ -90,6 +127,10 @@ export class Recaller { } }); return Array.from(candidates.values()) + .map(candidate => ({ + ...candidate, + score: candidate.score + RECENCY_WEIGHT * recencyOf(candidate.memory, Date.now()), + })) .sort((left, right) => right.score - left.score || left.directRank - right.directRank || right.memory.updatedAt - left.memory.updatedAt @@ -97,6 +138,37 @@ export class Recaller { .slice(0, limit) .map(candidate => candidate.memory); } + /** + * Cross-encoder pass over the head of the candidate list. + * + * Only the head is scored, not the whole pool: the point is to separate the + * few candidates a bi-encoder ranked almost equally, and cross-encoder cost + * grows linearly per document. The untouched tail keeps its embedding order, + * so a reranker that disagrees still cannot erase a candidate. + * + * Any failure — no reranker configured, timeout, malformed response — returns + * the input order unchanged. + */ + async rerankMemories(query, memories) { + if (!this.rerank || memories.length < 2) + return memories; + const headSize = Math.max(1, Math.min(memories.length, this.cfg.rerankTopN ?? 6)); + const head = memories.slice(0, headSize); + const tail = memories.slice(headSize); + try { + const scores = await this.rerank(query, head.map(memory => `${memory.outcome}: ${memory.summary}`)); + if (!scores || scores.length !== head.length) + return memories; + const ordered = head + .map((memory, index) => ({ memory, score: scores[index] })) + .sort((left, right) => right.score - left.score) + .map(entry => entry.memory); + return [...ordered, ...tail]; + } + catch { + return memories; + } + } recallTurnMemories(query, limit, queryVector) { const lexical = searchTurnMemories(this.db, query, limit); const threshold = this.cfg.semanticScoreThreshold; diff --git a/dist/src/recaller/rerank.js b/dist/src/recaller/rerank.js new file mode 100644 index 0000000..9901d45 --- /dev/null +++ b/dist/src/recaller/rerank.js @@ -0,0 +1,67 @@ +/** + * Cross-encoder reranking over recall candidates. + * + * The integration shape (best-effort, opt-in, a failed reranker may only + * disable reranking and never search itself) follows slow-stack/mneme's + * reranker wiring (MIT License), https://github.com/slow-stack/mneme. The + * transport is different: that project loads an ONNX model in-process, while + * this one talks to an HTTP endpoint so the model can stay out of the agent + * process and out of VRAM. + * + * Why reranking is worth a network hop: a bi-encoder embeds query and document + * separately, so it cannot see term-level interaction. Measured on this + * machine's corpus, a cross-encoder separated a true match at 0.9974 from the + * next candidate at 0.3435 — an ordering the embedding scores did not produce. + */ +/** + * Build a reranker backed by an HTTP endpoint that speaks the + * `{ query, documents } -> { results: [{ index, relevance_score }] }` shape + * (Jina / Cohere compatible, also what llama.cpp-style local servers expose). + * + * Every failure mode returns null rather than throwing: a reranker outage, a + * malformed payload, a mismatched result count, or a timeout all mean "use the + * embedding order", never "fail the recall". + */ +export function createHttpReranker(options) { + const { endpoint, model, timeoutMs = 1500, headers } = options; + return async (query, documents) => { + if (!documents.length) + return []; + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + try { + const body = { query, documents }; + if (model) + body.model = model; + const response = await fetch(endpoint, { + method: "POST", + headers: { "content-type": "application/json", ...headers }, + body: JSON.stringify(body), + signal: controller.signal, + }); + if (!response.ok) + return null; + const payload = await response.json(); + const results = Array.isArray(payload?.results) ? payload.results : []; + if (results.length !== documents.length) + return null; + const scores = new Array(documents.length).fill(0); + for (const item of results) { + const index = Number(item?.index); + const score = Number(item?.relevance_score ?? item?.score); + if (!Number.isInteger(index) || index < 0 || index >= scores.length) + return null; + if (!Number.isFinite(score)) + return null; + scores[index] = score; + } + return scores; + } + catch { + return null; + } + finally { + clearTimeout(timer); + } + }; +} diff --git a/dist/src/store/store.js b/dist/src/store/store.js index 9940218..dc0e476 100644 --- a/dist/src/store/store.js +++ b/dist/src/store/store.js @@ -210,6 +210,69 @@ export function upsertEdge(db, e) { VALUES (?,?,?,?,?,?,?,?)`) .run(uid("e"), e.fromId, e.toId, e.type, e.instruction, e.condition ?? null, e.sessionId, Date.now()); } +const NODE_TYPES = new Set(["TASK", "SKILL", "EVENT"]); +const EDGE_TYPES = new Set([ + "RELATES", "SUPERSEDES", "USED_SKILL", "SOLVED_BY", "REQUIRES", "PATCHES", "CONFLICTS_WITH", +]); +/** + * Persist the concept layer a completed turn named, atomically with its provenance. + * + * The concept graph is the durable half of memory: turn capsules answer "what + * happened", concepts answer "what is this thing". Extraction already emits + * both; this is the only place that turns the concept half into rows. + * + * Idempotence and provenance rules: + * - a concept that already exists is CONFIRMED (content kept, source added) so + * a recurring mention cannot silently rewrite an established description nor + * detach the evidence that supported it; + * - a relation is keyed by (from, to, type), so re-observing it updates the + * instruction instead of duplicating the edge; + * - endpoints that are not declared concepts in the same payload are dropped: + * a relation without both nodes is not evidence of anything. + */ +export function applyConceptExtraction(db, sessionId, sources, entities = [], relations = []) { + if (!entities.length) + return { nodes: 0, edges: 0 }; + const idByName = new Map(); + let nodes = 0; + for (const entity of entities) { + // TypeBox validates shape and length but not `enum` at runtime, so a value + // outside the declared kind set would otherwise reach the CHECK constraint. + if (!NODE_TYPES.has(entity.type)) + continue; + const existing = findByName(db, entity.name); + const { node, isNew } = upsertNode(db, { + name: entity.name, + type: entity.type, + description: entity.description, + content: entity.description, + // Confirmation preserves the established claim; creation publishes it. + operation: existing ? "confirm" : "create", + }, sessionId, sources); + if (isNew) + nodes++; + idByName.set(entity.name, node.id); + } + let edges = 0; + const resolve = (name) => idByName.get(name) ?? findByName(db, name)?.id ?? null; + for (const relation of relations) { + if (!EDGE_TYPES.has(relation.type)) + continue; + const fromId = resolve(relation.from); + const toId = resolve(relation.to); + if (!fromId || !toId || fromId === toId) + continue; + upsertEdge(db, { + fromId, + toId, + type: relation.type, + instruction: relation.instruction, + sessionId, + }); + edges++; + } + return { nodes, edges }; +} export function edgesFrom(db, id) { return db.prepare("SELECT * FROM gm_edges WHERE from_id=?").all(id).map(toEdge); } @@ -843,20 +906,77 @@ export function hasTurnMemories(db) { return Number(db.prepare("SELECT COUNT(*) AS count FROM gm_turn_memories").get()?.count ?? 0) > 0; } /** Lexical fallback for hosts without a working embedding provider. */ +/** + * Question shells and particles that carry no retrieval signal. Stripping them + * is what lets a natural-language Chinese question reach the stored summary at + * all: "之前修好的发布端口是多少?" never appears verbatim in any summary, but + * "发布端口" does. + */ +const LEXICAL_TAIL = /(?:好了吗|好了|是什么呢|是多少呢|是多少|是什么|怎么样|为什么|怎么|如何|吗|呢|吧|啊|呀|么|了)+[\s??!!。.,,]*$/u; +const LEXICAL_STOPWORDS = new Set([ + "什么", "怎么", "如何", "为什么", "是不是", "可以", "需要", "这个", "那个", + "现在", "然后", "以及", "但是", "因为", "所以", "如果", "就是", "我们", + "你们", "他们", "一下", "看看", "帮我", "请问", "有没有", "多少", "继续", + "the", "and", "for", "with", "what", "how", "why", "is", "are", "was", +]); +/** + * Turn a query into substring probes. Whole-phrase matching was precise but + * blind to any question phrased as a sentence; OR-ing every token would swing + * to the opposite failure and pull unrelated turns in. The middle ground is a + * bounded term list plus a majority-hit requirement, so a match still has to + * share most of the query's substance rather than one incidental word. + */ +function lexicalTerms(query) { + const seen = new Set(); + const terms = []; + for (const part of query.split(/[\s,,。.、;;::!!??()()\[\]【】"'“”‘’\/\\|]+/u)) { + const raw = part.trim(); + if (!raw) + continue; + let term = raw.replace(LEXICAL_TAIL, "").trim(); + if (term.length < 2) + term = raw; + if (term.length < 2) + continue; + const key = term.toLowerCase(); + if (seen.has(key) || LEXICAL_STOPWORDS.has(key)) + continue; + seen.add(key); + terms.push(term); + if (terms.length >= 6) + break; + } + return terms; +} export function searchTurnMemories(db, query, limit) { - const phrase = query.trim().replace(/\s+/g, " "); - if (!phrase) + const terms = lexicalTerms(query); + if (!terms.length) return []; - // Semantic search handles paraphrases. The lexical fallback deliberately - // requires the complete phrase; OR-ing common words such as "is"/"the" - // turns a fallback into another source of unrelated prompt injection. + // Semantic search handles paraphrases. The lexical arm only has to survive + // the case where the embedding provider is down or the question is a + // near-verbatim echo of something already stored. A single shared word is not + // evidence: "发布端口,配置文件" matched five summaries that only said + // "配置文件". Multi-term queries therefore require most of their terms, and + // a two-term query requires both. + const minHits = terms.length === 1 + ? 1 + : Math.max(2, Math.ceil(terms.length * 0.6)); + const hitExpr = terms.map(() => "(CASE WHEN summary LIKE ? THEN 1 ELSE 0 END)").join(" + "); + const matchExpr = terms.map(() => "summary LIKE ?").join(" OR "); + const params = [ + ...terms.map(term => `%${term}%`), + ...terms.map(term => `%${term}%`), + limit, + ]; const rows = db.prepare(` - SELECT * FROM gm_turn_memories - WHERE summary LIKE ? - ORDER BY updated_at DESC + SELECT *, ${hitExpr} AS lexical_hits FROM gm_turn_memories + WHERE ${matchExpr} + ORDER BY lexical_hits DESC, updated_at DESC LIMIT ? - `).all(`%${phrase}%`, limit); - return rows.map(row => toTurnMemory(db, row)); + `).all(...params); + return rows + .filter(row => Number(row.lexical_hits) >= minHits) + .map(row => toTurnMemory(db, row)); } export function saveTurnVector(db, memoryId, content, vec) { const hash = createHash("md5").update(content).digest("hex"); diff --git a/dsh.ts b/dsh.ts index c39afe9..ea5bf2f 100644 --- a/dsh.ts +++ b/dsh.ts @@ -35,6 +35,7 @@ import { upsertNode, upsertTurnMemory, replaceNavigationTriples, + applyConceptExtraction, } from "./src/store/store.ts"; import { Extractor } from "./src/extractor/extract.ts"; import { @@ -42,6 +43,7 @@ import { GRAPH_EXTRACTION_TOOL_NAME, } from "./src/extractor/contract.ts"; import { Recaller } from "./src/recaller/recall.ts"; +import { createHttpReranker } from "./src/recaller/rerank.ts"; import { assembleContext } from "./src/format/assemble.ts"; import { replaceDshArchivedPrefix, @@ -88,6 +90,17 @@ export interface Config { recallMaxNodes?: number; /** Optional embedding-provider-calibrated cosine floor for every recall path. */ semanticScoreThreshold?: number; + /** + * Optional cross-encoder reranker. Off unless `rerankEndpoint` is set: an + * extra HTTP hop per turn must be an explicit choice, and a reranker outage + * may only cost the rerank, never the recall. + */ + rerankEndpoint?: string; + rerankModel?: string; + /** Hard ceiling for the rerank request. Defaults to 1500 ms. */ + rerankTimeoutMs?: number; + /** How many candidates to score, counted from the head. Defaults to 6. */ + rerankTopN?: number; maintenanceInterval?: number; /** Durable raw-message retention. Defaults to keep=all (no deletion). */ messageRetention?: MessageRetentionConfig; @@ -145,6 +158,13 @@ interface DshContext { } const HOST = "dsh"; +/** + * Retrieval pool multiplier applied before the visibility filter. Five slots + * per rendered slot is enough to backfill a block that a same-session answer + * would otherwise empty, while costing nothing measurable: the vector path is + * already a full scan and the lexical path is a single indexed LIKE. + */ +const RECALL_OVER_FETCH = 5; function sessionKey(id: unknown): string { return `${HOST}:${String(id)}`; } @@ -238,12 +258,20 @@ export function apply(ctx: DshContext, input: Config = {}): void { compactTurnCount: maintenanceInterval, recallMaxNodes, semanticScoreThreshold: input.semanticScoreThreshold ?? DEFAULT_CONFIG.semanticScoreThreshold, + rerankTopN: input.rerankTopN ?? 6, embedding, }; const extractionEnabled = input.extractionEnabled ?? true; const recallEnabled = input.recallEnabled ?? true; const db = openDb(config.dbPath, { busyTimeoutMs: input.dbBusyTimeoutMs }); const recaller = new Recaller(db, config); + if (input.rerankEndpoint) { + recaller.setRerankFn(createHttpReranker({ + endpoint: input.rerankEndpoint, + model: input.rerankModel, + timeoutMs: input.rerankTimeoutMs ?? 1500, + })); + } const latestRoute = new Map(); const extractChain = new Map>(); const turnCounts = new Map(); @@ -428,6 +456,21 @@ export function apply(ctx: DshContext, input: Config = {}): void { })), }); replaceNavigationTriples(db, turnMemory, result.triples); + // The concept layer rides on the same extraction payload. Writing it here + // is what keeps the durable graph current: without this step the host only + // ever stored navigation, so every concept had to be recorded by hand. + const concept = applyConceptExtraction( + db, + sid, + turnMemory.sources, + result.entities ?? [], + result.relations ?? [], + ); + if (concept.nodes || concept.edges) { + ctx.logger.info( + `[graph-memory] DSH stored ${concept.nodes} concept nodes and ${concept.edges} concept edges`, + ); + } // The navigation graph becomes queryable only after its atomic SPO write. // This runs inside the existing background extraction worker, never in the // foreground turn, and makes the newest completed memory available to PPR. @@ -700,7 +743,13 @@ export function apply(ctx: DshContext, input: Config = {}): void { // is still in flight. Historical recall must wait for that shared probe; // otherwise the very first cross-session question can miss all vectors. await embeddingReady; - const recalled = await recaller.recall(query); + // Retrieval runs on a wider pool than the rendered width. The filters + // below drop memory already visible verbatim in this session's fresh + // window, and a pool sized exactly to the render width could be emptied + // entirely by that filter — a short follow-up whose top hits all came + // from the live session then injected nothing at all. Over-fetching lets + // the block backfill from cross-session or archived evidence instead. + const recalled = await recaller.recall(query, { overFetch: RECALL_OVER_FETCH }); signal?.throwIfAborted?.(); const key = String(id); const currentSession = sessionKey(id); @@ -722,12 +771,12 @@ export function apply(ctx: DshContext, input: Config = {}): void { currentSession, visibleMessageIds, hasArchivedHistory, - ); + ).slice(0, config.recallMaxNodes); const recalledMemories = filterDshRecallMemories( recalled.turnMemories, currentSession, visibleMessageIds, - ); + ).slice(0, config.recallMaxNodes); if (!recalledNodes.length && !recalledMemories.length) return decision; const recalledIds = new Set(recalledNodes.map(node => node.id)); const built = assembleContext(db, { diff --git a/src/consolidate/cluster.ts b/src/consolidate/cluster.ts new file mode 100644 index 0000000..74759bd --- /dev/null +++ b/src/consolidate/cluster.ts @@ -0,0 +1,160 @@ +/** + * Clustering primitives used by memory consolidation. + * + * Adapted from slow-stack/mneme (MIT License) — `dsh-mneme/src/dream/clustering.js`, + * https://github.com/slow-stack/mneme. The algorithms are reproduced unchanged; + * only types and the grouping key are parameterized so they fit this store's + * turn-memory shape. Original copyright (c) slow-stack/mneme contributors. + * + * Kept as a standalone module: nothing here touches the database, so it can be + * exercised and tested without a store, and the consolidation layer can be + * dropped without leaving a mark on recall or extraction. + */ + +/** + * Cosine similarity of two equal-length vectors. Returns 0 for empty input, + * mismatched lengths, or a zero-norm vector (numerically stable rather than + * NaN, so a degenerate embedding cannot poison a whole run). + */ +export function cosineSimilarity(a: ArrayLike, b: ArrayLike): number { + if (!a || !b || a.length === 0 || a.length !== b.length) return 0; + let dot = 0; + let na = 0; + let nb = 0; + for (let i = 0; i < a.length; i += 1) { + dot += a[i] * b[i]; + na += a[i] * a[i]; + nb += b[i] * b[i]; + } + if (na === 0 || nb === 0) return 0; + return dot / (Math.sqrt(na) * Math.sqrt(nb)); +} + +/** + * K-Means over numeric vectors, returning one index array per cluster. + * + * k-means++ seeding, and empty clusters are reseeded mid-iteration so a bad + * draw cannot collapse the result to fewer clusters than requested. + */ +export function kMeans( + vectors: Float32Array[], + k: number, + opts: { maxIter?: number; tol?: number; random?: () => number } = {}, +): number[][] { + const n = vectors.length; + if (n === 0) return []; + const { maxIter = 100, tol = 1e-4, random = Math.random } = opts; + if (k <= 0 || k >= n) return [vectors.map((_, index) => index)]; + + const dim = vectors[0].length; + const randomIndex = () => Math.floor(random() * n); + const squaredDist = (v: Float32Array, c: Float32Array): number => { + let sum = 0; + for (let d = 0; d < dim; d += 1) sum += (v[d] - c[d]) * (v[d] - c[d]); + return sum; + }; + + // First centroid uniform, later ones sampled proportional to squared distance. + const centroids: Float32Array[] = [Float32Array.from(vectors[randomIndex()])]; + while (centroids.length < k) { + const dist = new Array(n); + let sum = 0; + for (let i = 0; i < n; i += 1) { + let best = Infinity; + for (const centroid of centroids) best = Math.min(best, squaredDist(vectors[i], centroid)); + dist[i] = best; + sum += best; + } + if (sum === 0) { + centroids.push(Float32Array.from(vectors[randomIndex()])); + } else { + let r = random() * sum; + let pick = 0; + for (let i = 0; i < n; i += 1) { + r -= dist[i]; + if (r <= 0) { + pick = i; + break; + } + } + centroids.push(Float32Array.from(vectors[pick])); + } + } + + const assignment = new Array(n).fill(0); + for (let iter = 0; iter < maxIter; iter += 1) { + let changed = false; + for (let i = 0; i < n; i += 1) { + let best = 0; + let bestD = Infinity; + for (let j = 0; j < k; j += 1) { + const d = squaredDist(vectors[i], centroids[j]); + if (d < bestD) { + bestD = d; + best = j; + } + } + if (assignment[i] !== best) { + assignment[i] = best; + changed = true; + } + } + if (!changed) break; + + const sums = Array.from({ length: k }, () => new Float32Array(dim)); + const counts = new Array(k).fill(0); + for (let i = 0; i < n; i += 1) { + const cluster = assignment[i]; + counts[cluster] += 1; + for (let d = 0; d < dim; d += 1) sums[cluster][d] += vectors[i][d]; + } + let maxMove = 0; + for (let j = 0; j < k; j += 1) { + if (counts[j] === 0) { + centroids[j] = Float32Array.from(vectors[randomIndex()]); + maxMove = Infinity; + continue; + } + for (let d = 0; d < dim; d += 1) { + const mean = sums[j][d] / counts[j]; + const move = mean - centroids[j][d]; + if (move * move > maxMove) maxMove = move * move; + centroids[j][d] = mean; + } + } + if (Math.sqrt(maxMove) < tol) break; + } + + const clusters = Array.from({ length: k }, () => [] as number[]); + for (let i = 0; i < n; i += 1) clusters[assignment[i]].push(i); + return clusters.filter(cluster => cluster.length > 0); +} + +/** Cluster items by their vectors, preserving the original object references. */ +export function clusterByVector(items: T[], vectors: Float32Array[], k: number): T[][] { + return kMeans(vectors, k).map(cluster => cluster.map(index => items[index])); +} + +/** + * Pairs that are highly similar yet might contradict each other. + * + * Similarity is only a candidate signal: `0.9` means "read these two together", + * never "these two disagree". Only items sharing the same group key are paired, + * because a merge or conflict between unrelated kinds of memory is meaningless. + */ +export function findPotentialConflicts( + items: T[], + vectors: Float32Array[], + keyOf: (item: T) => string, + threshold = 0.85, +): Array<{ a: T; b: T; similarity: number }> { + const pairs: Array<{ a: T; b: T; similarity: number }> = []; + for (let i = 0; i < items.length; i += 1) { + for (let j = i + 1; j < items.length; j += 1) { + if (keyOf(items[i]) !== keyOf(items[j])) continue; + const similarity = cosineSimilarity(vectors[i], vectors[j]); + if (similarity > threshold) pairs.push({ a: items[i], b: items[j], similarity }); + } + } + return pairs; +} diff --git a/src/extractor/contract.ts b/src/extractor/contract.ts index 7af8d3f..cbf38be 100644 --- a/src/extractor/contract.ts +++ b/src/extractor/contract.ts @@ -13,6 +13,32 @@ const NavigationTripleSchema = Type.Object({ object: Type.String({ minLength: 1, description: "Concrete searchable object phrase." }), }, { additionalProperties: false }); +const ConceptEntitySchema = Type.Object({ + name: Type.String({ + minLength: 1, + maxLength: 24, + description: "Reusable noun phrase naming one concept; never a sentence, a status word, or a bare quantity.", + }), + type: Type.String({ + enum: ["EVENT", "SKILL", "TASK"], + description: "Concept kind: EVENT happened, SKILL reusable capability, TASK goal or pending work.", + }), + description: Type.String({ + minLength: 1, + description: "One sentence stating what this concept is, using only evidence from this turn.", + }), +}, { additionalProperties: false }); + +const ConceptRelationSchema = Type.Object({ + from: Type.String({ minLength: 1, description: "Subject name; must equal one entities[].name value." }), + to: Type.String({ minLength: 1, description: "Object name; must equal one entities[].name value." }), + type: Type.String({ + enum: ["RELATES", "SUPERSEDES", "USED_SKILL", "SOLVED_BY", "REQUIRES", "PATCHES", "CONFLICTS_WITH"], + description: "Declared relation kind between the two concepts.", + }), + instruction: Type.String({ minLength: 1, description: "Short natural-language statement of the relation." }), +}, { additionalProperties: false }); + /** * Provider-facing extraction contract. * @@ -31,19 +57,36 @@ export const GRAPH_EXTRACTION_SCHEMA = Type.Object({ triples: Type.Array(NavigationTripleSchema, { description: "Required SPO relations derived only from summary; use an empty array when none are explicit.", }), + entities: Type.Optional(Type.Array(ConceptEntitySchema, { + description: "Optional durable concept list for this turn; omit or use [] when the turn yields no reusable concept.", + })), + relations: Type.Optional(Type.Array(ConceptRelationSchema, { + description: "Optional typed links between entries of entities; both endpoints must be declared concepts.", + })), }, { additionalProperties: false }); export interface StructuredGraphExtraction { summary: string; outcome: TurnOutcome; triples: Array<{ subject: string; predicate: string; object: string }>; + entities?: Array<{ + name: string; + type: "EVENT" | "SKILL" | "TASK"; + description: string; + }>; + relations?: Array<{ + from: string; + to: string; + type: "RELATES" | "SUPERSEDES" | "USED_SKILL" | "SOLVED_BY" | "REQUIRES" | "PATCHES" | "CONFLICTS_WITH"; + instruction: string; + }>; } export const GRAPH_EXTRACTION_TOOL_NAME = "submit_result"; export const GRAPH_EXTRACTION_TOOL = Object.freeze({ name: GRAPH_EXTRACTION_TOOL_NAME, - description: "Submit exactly the three required fields summary, outcome, and triples. Derive zero or more subject-predicate-object triples only from the summary; triples must be [] when no relation is explicit. Emit no text.", + description: "Submit exactly the three required fields summary, outcome, and triples. Derive zero or more subject-predicate-object triples only from the summary; triples must be [] when no relation is explicit. Optionally add entities and relations when the turn names reusable concepts. Emit no text.", parameters: GRAPH_EXTRACTION_SCHEMA, }); diff --git a/src/extractor/extract.ts b/src/extractor/extract.ts index 0904112..51f8c64 100755 --- a/src/extractor/extract.ts +++ b/src/extractor/extract.ts @@ -43,9 +43,11 @@ const EXTRACT_SYS = `【任务】 1. summary 用一句简短、自包含的话写清本轮对象、结论和最终回答明确报告的完成情况;不复述过程,不添加输入未表达的信息。 2. outcome 按最终回答选择 completed、partial、failed、informational、unknown 之一。 3. triples 只从 summary 拆分。subject 和 object 使用具体可检索短语,predicate 使用简短自然语言;没有明确关系时使用空数组,不补充、不猜测。 +4. entities 可选,用于记录本轮出现的、以后还会被问到的概念。每个概念给三样:name 是 2~16 字的名词短语(禁止整句,禁止“全部正常 / 已完成 / 72 个”这类状态与数量);type 取 EVENT(发生过的事)、SKILL(可复用的工具、技能、脚本、服务、配置)、TASK(目标、待办、缺口);description 用一句话说明它是什么,只用本轮信息。本轮没有可复用概念时省略该字段。 +5. relations 可选,只在确有明确关系时输出:from 与 to 必须是 entities 里出现过的 name,type 取 RELATES、SUPERSEDES、USED_SKILL、SOLVED_BY、REQUIRES、PATCHES、CONFLICTS_WITH 之一,instruction 用一句中文说明关系。没有关系时省略该字段。 【输出合同】 -只调用 submit_result 一次。参数对象必须且只能包含 summary、outcome、triples 三个顶层字段,三个字段都不能省略。不要输出解释或正文。 +只调用 submit_result 一次。参数对象必须包含 summary、outcome、triples 三个顶层字段,三个字段都不能省略;可选包含 entities 与 relations。不要输出解释或正文。 示例一: 输入:用户要求把周会改到周四;最终回答确认日程已更新。 @@ -59,7 +61,11 @@ const EXTRACT_SYS = `【任务】 输入只确认稍后继续讨论,没有新结论。 输出:{"summary":"本轮确认稍后继续讨论,未产生新结论。","outcome":"informational","triples":[]} -调用前自检:参数是否恰好包含三个顶层字段;outcome 是否属于枚举;没有关系时 triples 是否仍明确写为 []。`; +示例四(带概念): +输入:助手把抽取路由钉到本地 27B,避免打到线上账号。 +输出:{"summary":"抽取路由已改为本地 27B,不再使用线上账号。","outcome":"completed","triples":[{"subject":"抽取路由","predicate":"改为","object":"本地 27B"}],"entities":[{"name":"抽取路由","type":"SKILL","description":"抽取请求使用的模型路由配置"},{"name":"本地 27B","type":"SKILL","description":"本机 llama-swap 上的主模型"},{"name":"线上账号","type":"EVENT","description":"此前误用线上 API 的问题"}],"relations":[{"from":"抽取路由","to":"本地 27B","type":"REQUIRES","instruction":"路由改为依赖本地模型"},{"from":"抽取路由","to":"线上账号","type":"SOLVED_BY","instruction":"改路由后不再走线上账号"}]} + +调用前自检:是否包含 summary、outcome、triples 三个必需字段;outcome 是否属于枚举;没有关系时 triples 是否仍明确写为 [];若输出 entities 与 relations,relation 两端是否都能在 entities 的 name 中找到,且 type 是否在枚举内。`; // ─── 提取 User Prompt ─────────────────────────────────────────── @@ -111,6 +117,11 @@ export class Extractor { return { turn: { summary: p.summary, outcome: p.outcome }, triples: p.triples.map(triple => ({ ...triple })), + // Optional blocks are copied verbatim and default to empty. The contract + // decides whether they are well formed; a host that does not persist + // concepts simply ignores them, so the field stays additive. + entities: (p.entities ?? []).map(entity => ({ ...entity })), + relations: (p.relations ?? []).map(relation => ({ ...relation })), }; } catch (err) { // Extraction output can contain private conversation facts. Report the diff --git a/src/recaller/recall.ts b/src/recaller/recall.ts index 4535195..5cac1d7 100755 --- a/src/recaller/recall.ts +++ b/src/recaller/recall.ts @@ -14,6 +14,7 @@ import { type DatabaseSyncInstance } from "../store/sqlite.ts"; import { createHash } from "crypto"; import type { GmConfig, RecallResult, GmNode, GmTurnMemory } from "../types.ts"; import type { EmbedFn } from "../engine/embed.ts"; +import type { RerankFn } from "./rerank.ts"; import { searchNodes, vectorSearchWithScore, graphWalk, @@ -27,8 +28,27 @@ import { } from "../store/store.ts"; import { personalizedNavigationPageRank } from "../graph/pagerank.ts"; +/** + * Recency is a tie-breaker, not a relevance signal. A memory that is slightly + * less similar but four months older should not outrank the one that describes + * what the user is doing this week, yet it must not jump ahead of a clearly + * better match either. A 30-day half life with a 0.15 ceiling sits below one + * reciprocal-rank step (1/1 - 1/2 = 0.5), so it can reorder near-ties without + * overriding the similarity ranks that the embedding provider produced. + */ +const RECENCY_HALF_LIFE_MS = 30 * 24 * 60 * 60 * 1000; +const RECENCY_WEIGHT = 0.15; + +/** 1 for a memory written now, 0.5 at thirty days, 0.25 at sixty. */ +function recencyOf(memory: GmTurnMemory, now: number): number { + const updatedAt = Number(memory.updatedAt); + if (!Number.isFinite(updatedAt) || updatedAt <= 0) return 0; + return Math.pow(0.5, Math.max(0, now - updatedAt) / RECENCY_HALF_LIFE_MS); +} + export class Recaller { private embed: EmbedFn | null = null; + private rerank: RerankFn | null = null; private embeddingFingerprint = ""; constructor(private db: DatabaseSyncInstance, private cfg: GmConfig) {} @@ -38,8 +58,27 @@ export class Recaller { this.embeddingFingerprint = fingerprint; } - async recall(query: string): Promise { + /** + * Optional cross-encoder pass over the candidate head. Absent by default: a + * host that does not configure a reranker keeps the embedding order exactly. + */ + setRerankFn(fn: RerankFn | null): void { + this.rerank = fn; + } + + /** + * @param options.overFetch Retrieval pool multiplier. Defaults to 1, which + * keeps the historical behaviour: `recallMaxNodes` bounds both retrieval and + * what the caller may render. A host that post-filters results (for example + * to drop memories already visible in its fresh window) passes a multiplier + * so that filtering cannot empty the whole block: a three-slot pool where all + * three hits belong to the current session used to yield nothing at all, and + * the caller then injected no memory for that turn. + */ + async recall(query: string, options: { overFetch?: number } = {}): Promise { const limit = this.cfg.recallMaxNodes; + const overFetch = Math.max(1, Math.floor(options.overFetch ?? 1)); + const poolLimit = limit * overFetch; let queryVector: number[] | undefined; if (this.embed) { try { @@ -50,7 +89,7 @@ export class Recaller { } } - const directMemories = this.recallTurnMemories(query, limit, queryVector); + const directMemories = this.recallTurnMemories(query, poolLimit, queryVector); const seedIds = findNavigationSeedTermIds( this.db, query, @@ -71,13 +110,14 @@ export class Recaller { rankTurnMemoryIdsByNavigation(this.db, navigationScores), ); } - const turnMemories = this.mergeTurnMemoryRanks(directMemories, graphMemories, limit); + let turnMemories = this.mergeTurnMemoryRanks(directMemories, graphMemories, poolLimit); + turnMemories = await this.rerankMemories(query, turnMemories); if (turnMemories.length) { const memoryIds = turnMemories.map(memory => memory.id); const nodes = nodesForTurnMemories( this.db, memoryIds, - limit, + poolLimit, ); const { edges } = graphWalk(this.db, nodes.map(node => node.id), 0); return { @@ -90,7 +130,7 @@ export class Recaller { // Databases created before the turn-memory migration remain searchable. // The same confidence rule applies; a legacy node match cannot bypass it. - return this.recallPrecise(query, limit, queryVector, hasTurnMemories(this.db)); + return this.recallPrecise(query, poolLimit, queryVector, hasTurnMemories(this.db)); } /** @@ -128,6 +168,10 @@ export class Recaller { } }); return Array.from(candidates.values()) + .map(candidate => ({ + ...candidate, + score: candidate.score + RECENCY_WEIGHT * recencyOf(candidate.memory, Date.now()), + })) .sort((left, right) => right.score - left.score || left.directRank - right.directRank @@ -138,6 +182,41 @@ export class Recaller { .map(candidate => candidate.memory); } + /** + * Cross-encoder pass over the head of the candidate list. + * + * Only the head is scored, not the whole pool: the point is to separate the + * few candidates a bi-encoder ranked almost equally, and cross-encoder cost + * grows linearly per document. The untouched tail keeps its embedding order, + * so a reranker that disagrees still cannot erase a candidate. + * + * Any failure — no reranker configured, timeout, malformed response — returns + * the input order unchanged. + */ + private async rerankMemories( + query: string, + memories: GmTurnMemory[], + ): Promise { + if (!this.rerank || memories.length < 2) return memories; + const headSize = Math.max(1, Math.min(memories.length, this.cfg.rerankTopN ?? 6)); + const head = memories.slice(0, headSize); + const tail = memories.slice(headSize); + try { + const scores = await this.rerank( + query, + head.map(memory => `${memory.outcome}: ${memory.summary}`), + ); + if (!scores || scores.length !== head.length) return memories; + const ordered = head + .map((memory, index) => ({ memory, score: scores[index] })) + .sort((left, right) => right.score - left.score) + .map(entry => entry.memory); + return [...ordered, ...tail]; + } catch { + return memories; + } + } + private recallTurnMemories( query: string, limit: number, diff --git a/src/recaller/rerank.ts b/src/recaller/rerank.ts new file mode 100644 index 0000000..44a3a12 --- /dev/null +++ b/src/recaller/rerank.ts @@ -0,0 +1,73 @@ +/** + * Cross-encoder reranking over recall candidates. + * + * The integration shape (best-effort, opt-in, a failed reranker may only + * disable reranking and never search itself) follows slow-stack/mneme's + * reranker wiring (MIT License), https://github.com/slow-stack/mneme. The + * transport is different: that project loads an ONNX model in-process, while + * this one talks to an HTTP endpoint so the model can stay out of the agent + * process and out of VRAM. + * + * Why reranking is worth a network hop: a bi-encoder embeds query and document + * separately, so it cannot see term-level interaction. Measured on this + * machine's corpus, a cross-encoder separated a true match at 0.9974 from the + * next candidate at 0.3435 — an ordering the embedding scores did not produce. + */ + +/** Returns one score per document, or null when the reranker is unavailable. */ +export type RerankFn = (query: string, documents: string[]) => Promise; + +export interface HttpRerankOptions { + /** Full URL of the rerank endpoint, e.g. `http://127.0.0.1:8081/rerank`. */ + endpoint: string; + /** Model name passed through to the server; omitted when empty. */ + model?: string; + /** Hard ceiling. A slow reranker must not delay the user's turn. */ + timeoutMs?: number; + headers?: Record; +} + +/** + * Build a reranker backed by an HTTP endpoint that speaks the + * `{ query, documents } -> { results: [{ index, relevance_score }] }` shape + * (Jina / Cohere compatible, also what llama.cpp-style local servers expose). + * + * Every failure mode returns null rather than throwing: a reranker outage, a + * malformed payload, a mismatched result count, or a timeout all mean "use the + * embedding order", never "fail the recall". + */ +export function createHttpReranker(options: HttpRerankOptions): RerankFn { + const { endpoint, model, timeoutMs = 1500, headers } = options; + return async (query: string, documents: string[]): Promise => { + if (!documents.length) return []; + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + try { + const body: Record = { query, documents }; + if (model) body.model = model; + const response = await fetch(endpoint, { + method: "POST", + headers: { "content-type": "application/json", ...headers }, + body: JSON.stringify(body), + signal: controller.signal, + }); + if (!response.ok) return null; + const payload: any = await response.json(); + const results = Array.isArray(payload?.results) ? payload.results : []; + if (results.length !== documents.length) return null; + const scores = new Array(documents.length).fill(0); + for (const item of results) { + const index = Number(item?.index); + const score = Number(item?.relevance_score ?? item?.score); + if (!Number.isInteger(index) || index < 0 || index >= scores.length) return null; + if (!Number.isFinite(score)) return null; + scores[index] = score; + } + return scores; + } catch { + return null; + } finally { + clearTimeout(timer); + } + }; +} diff --git a/src/store/store.ts b/src/store/store.ts index a571545..f3d34f0 100755 --- a/src/store/store.ts +++ b/src/store/store.ts @@ -288,6 +288,74 @@ export function upsertEdge( .run(uid("e"), e.fromId, e.toId, e.type, e.instruction, e.condition ?? null, e.sessionId, Date.now()); } +const NODE_TYPES: ReadonlySet = new Set(["TASK", "SKILL", "EVENT"]); +const EDGE_TYPES: ReadonlySet = new Set([ + "RELATES", "SUPERSEDES", "USED_SKILL", "SOLVED_BY", "REQUIRES", "PATCHES", "CONFLICTS_WITH", +]); + +/** + * Persist the concept layer a completed turn named, atomically with its provenance. + * + * The concept graph is the durable half of memory: turn capsules answer "what + * happened", concepts answer "what is this thing". Extraction already emits + * both; this is the only place that turns the concept half into rows. + * + * Idempotence and provenance rules: + * - a concept that already exists is CONFIRMED (content kept, source added) so + * a recurring mention cannot silently rewrite an established description nor + * detach the evidence that supported it; + * - a relation is keyed by (from, to, type), so re-observing it updates the + * instruction instead of duplicating the edge; + * - endpoints that are not declared concepts in the same payload are dropped: + * a relation without both nodes is not evidence of anything. + */ +export function applyConceptExtraction( + db: DatabaseSyncInstance, + sessionId: string, + sources: Array<{ messageId: string; turnIndex: number }>, + entities: Array<{ name: string; type: NodeType; description: string }> = [], + relations: Array<{ from: string; to: string; type: EdgeType; instruction: string }> = [], +): { nodes: number; edges: number } { + if (!entities.length) return { nodes: 0, edges: 0 }; + const idByName = new Map(); + let nodes = 0; + + for (const entity of entities) { + // TypeBox validates shape and length but not `enum` at runtime, so a value + // outside the declared kind set would otherwise reach the CHECK constraint. + if (!NODE_TYPES.has(entity.type)) continue; + const existing = findByName(db, entity.name); + const { node, isNew } = upsertNode(db, { + name: entity.name, + type: entity.type, + description: entity.description, + content: entity.description, + // Confirmation preserves the established claim; creation publishes it. + operation: existing ? "confirm" : "create", + }, sessionId, sources); + if (isNew) nodes++; + idByName.set(entity.name, node.id); + } + + let edges = 0; + const resolve = (name: string): string | null => idByName.get(name) ?? findByName(db, name)?.id ?? null; + for (const relation of relations) { + if (!EDGE_TYPES.has(relation.type)) continue; + const fromId = resolve(relation.from); + const toId = resolve(relation.to); + if (!fromId || !toId || fromId === toId) continue; + upsertEdge(db, { + fromId, + toId, + type: relation.type, + instruction: relation.instruction, + sessionId, + }); + edges++; + } + return { nodes, edges }; +} + export function edgesFrom(db: DatabaseSyncInstance, id: string): GmEdge[] { return (db.prepare("SELECT * FROM gm_edges WHERE from_id=?").all(id) as any[]).map(toEdge); } @@ -1048,23 +1116,77 @@ export function hasTurnMemories(db: DatabaseSyncInstance): boolean { } /** Lexical fallback for hosts without a working embedding provider. */ +/** + * Question shells and particles that carry no retrieval signal. Stripping them + * is what lets a natural-language Chinese question reach the stored summary at + * all: "之前修好的发布端口是多少?" never appears verbatim in any summary, but + * "发布端口" does. + */ +const LEXICAL_TAIL = /(?:好了吗|好了|是什么呢|是多少呢|是多少|是什么|怎么样|为什么|怎么|如何|吗|呢|吧|啊|呀|么|了)+[\s??!!。.,,]*$/u; +const LEXICAL_STOPWORDS = new Set([ + "什么", "怎么", "如何", "为什么", "是不是", "可以", "需要", "这个", "那个", + "现在", "然后", "以及", "但是", "因为", "所以", "如果", "就是", "我们", + "你们", "他们", "一下", "看看", "帮我", "请问", "有没有", "多少", "继续", + "the", "and", "for", "with", "what", "how", "why", "is", "are", "was", +]); + +/** + * Turn a query into substring probes. Whole-phrase matching was precise but + * blind to any question phrased as a sentence; OR-ing every token would swing + * to the opposite failure and pull unrelated turns in. The middle ground is a + * bounded term list plus a majority-hit requirement, so a match still has to + * share most of the query's substance rather than one incidental word. + */ +function lexicalTerms(query: string): string[] { + const seen = new Set(); + const terms: string[] = []; + for (const part of query.split(/[\s,,。.、;;::!!??()()\[\]【】"'“”‘’\/\\|]+/u)) { + const raw = part.trim(); + if (!raw) continue; + let term = raw.replace(LEXICAL_TAIL, "").trim(); + if (term.length < 2) term = raw; + if (term.length < 2) continue; + const key = term.toLowerCase(); + if (seen.has(key) || LEXICAL_STOPWORDS.has(key)) continue; + seen.add(key); + terms.push(term); + if (terms.length >= 6) break; + } + return terms; +} + export function searchTurnMemories( db: DatabaseSyncInstance, query: string, limit: number, ): GmTurnMemory[] { - const phrase = query.trim().replace(/\s+/g, " "); - if (!phrase) return []; - // Semantic search handles paraphrases. The lexical fallback deliberately - // requires the complete phrase; OR-ing common words such as "is"/"the" - // turns a fallback into another source of unrelated prompt injection. + const terms = lexicalTerms(query); + if (!terms.length) return []; + // Semantic search handles paraphrases. The lexical arm only has to survive + // the case where the embedding provider is down or the question is a + // near-verbatim echo of something already stored. A single shared word is not + // evidence: "发布端口,配置文件" matched five summaries that only said + // "配置文件". Multi-term queries therefore require most of their terms, and + // a two-term query requires both. + const minHits = terms.length === 1 + ? 1 + : Math.max(2, Math.ceil(terms.length * 0.6)); + const hitExpr = terms.map(() => "(CASE WHEN summary LIKE ? THEN 1 ELSE 0 END)").join(" + "); + const matchExpr = terms.map(() => "summary LIKE ?").join(" OR "); + const params = [ + ...terms.map(term => `%${term}%`), + ...terms.map(term => `%${term}%`), + limit, + ]; const rows = db.prepare(` - SELECT * FROM gm_turn_memories - WHERE summary LIKE ? - ORDER BY updated_at DESC + SELECT *, ${hitExpr} AS lexical_hits FROM gm_turn_memories + WHERE ${matchExpr} + ORDER BY lexical_hits DESC, updated_at DESC LIMIT ? - `).all(`%${phrase}%`, limit) as any[]; - return rows.map(row => toTurnMemory(db, row)); + `).all(...params) as any[]; + return rows + .filter(row => Number(row.lexical_hits) >= minHits) + .map(row => toTurnMemory(db, row)); } export function saveTurnVector( diff --git a/src/types.ts b/src/types.ts index c14a5a0..3a17b17 100755 --- a/src/types.ts +++ b/src/types.ts @@ -115,6 +115,22 @@ export interface ExtractionResult { predicate: string; object: string; }>; + /** + * Durable concepts named by this turn. Optional: a host may omit them, and a + * turn whose summary yields no reusable concept legitimately has none. + */ + entities?: Array<{ + name: string; + type: "EVENT" | "SKILL" | "TASK"; + description: string; + }>; + /** Typed links between `entities`; both endpoints must name a declared concept. */ + relations?: Array<{ + from: string; + to: string; + type: "RELATES" | "SUPERSEDES" | "USED_SKILL" | "SOLVED_BY" | "REQUIRES" | "PATCHES" | "CONFLICTS_WITH"; + instruction: string; + }>; } // ─── 召回结果 ───────────────────────────────────────────────── @@ -156,6 +172,13 @@ export interface GmConfig { * a "nearest" memory even when no memory is actually relevant. */ semanticScoreThreshold?: number; + /** + * How many candidates the optional cross-encoder reranker scores. Only the + * head is worth scoring: it separates candidates the bi-encoder ranked nearly + * equally, while cost grows linearly per document. Ignored when no reranker + * is configured. + */ + rerankTopN?: number; /** Number of recent user turns kept as native question/final-answer endpoints on the host context surface. */ freshTurnCount: number; embedding?: EmbeddingConfig; diff --git a/test/concept-extraction.test.ts b/test/concept-extraction.test.ts new file mode 100644 index 0000000..cfcc238 --- /dev/null +++ b/test/concept-extraction.test.ts @@ -0,0 +1,132 @@ +/** + * graph-memory — concept layer persistence + * + * The extraction contract lets a turn declare durable concepts; these cases + * pin down how that declaration becomes rows, and specifically the two rules a + * recurring mention must not break: an established description is not silently + * rewritten, and its provenance is only ever added to. + */ + +import { describe, it, expect, beforeEach } from "vitest"; +import type { DatabaseSyncInstance } from "../src/store/sqlite.ts"; +import { createTestDb } from "./helpers.ts"; +import { + applyConceptExtraction, + findByName, + getNodeSources, + saveMessageOnce, +} from "../src/store/store.ts"; + +const SID = "dsh:session-concept"; + +let db: DatabaseSyncInstance; + +/** Persist one durable Q/A pair and return the provenance refs for it. */ +function seedTurn(turn: number): Array<{ messageId: string; turnIndex: number }> { + saveMessageOnce(db, `${SID}:${turn}:u`, SID, turn, "user", "问题"); + saveMessageOnce(db, `${SID}:${turn}:a`, SID, turn, "assistant", "回答"); + return [`${SID}:${turn}:u`, `${SID}:${turn}:a`].map(messageId => ({ messageId, turnIndex: turn })); +} + +beforeEach(() => { + db = createTestDb(); +}); + +describe("concept extraction persistence", () => { + it("stores declared concepts and links them to the turn evidence", () => { + const sources = seedTurn(1); + const result = applyConceptExtraction(db, SID, sources, + [ + { name: "抽取路由", type: "SKILL", description: "抽取请求使用的模型路由配置" }, + { name: "本地 27B", type: "SKILL", description: "本机 llama-swap 上的主模型" }, + ], + [{ from: "抽取路由", to: "本地 27B", type: "REQUIRES", instruction: "路由改为依赖本地模型" }], + ); + + expect(result).toEqual({ nodes: 2, edges: 1 }); + const route = findByName(db, "抽取路由"); + expect(route).not.toBeNull(); + expect(route!.type).toBe("SKILL"); + // Provenance is what makes a concept reachable from recall. + expect(getNodeSources(db, [route!.id]).length).toBe(2); + }); + + it("confirms a recurring concept instead of rewriting its established claim", () => { + const first = seedTurn(1); + applyConceptExtraction(db, SID, first, + [{ name: "抽取路由", type: "SKILL", description: "初版描述" }], []); + const original = findByName(db, "抽取路由")!; + + const second = seedTurn(2); + const result = applyConceptExtraction(db, SID, second, + [{ name: "抽取路由", type: "SKILL", description: "后来的一句话" }], []); + + const after = findByName(db, "抽取路由")!; + expect(result.nodes).toBe(0); + expect(after.id).toBe(original.id); + expect(after.content).toBe(original.content); + expect(after.validatedCount).toBe(original.validatedCount + 1); + // Confirmation adds evidence; it never detaches the earlier turn. + expect(getNodeSources(db, [after.id]).length).toBe(4); + }); + + it("drops a relation whose endpoints were not declared as concepts", () => { + const sources = seedTurn(1); + const result = applyConceptExtraction(db, SID, sources, + [{ name: "抽取路由", type: "SKILL", description: "路由配置" }], + [{ from: "抽取路由", to: "未声明的东西", type: "RELATES", instruction: "指向未声明概念" }], + ); + + expect(result).toEqual({ nodes: 1, edges: 0 }); + }); + + it("re-observing the same relation does not duplicate the edge", () => { + const entities = [ + { name: "抽取路由", type: "SKILL" as const, description: "路由配置" }, + { name: "本地 27B", type: "SKILL" as const, description: "本地模型" }, + ]; + const relations = [ + { from: "抽取路由", to: "本地 27B", type: "REQUIRES" as const, instruction: "依赖本地模型" }, + ]; + const first = applyConceptExtraction(db, SID, seedTurn(1), entities, relations); + const second = applyConceptExtraction(db, SID, seedTurn(2), entities, relations); + + expect(first).toEqual({ nodes: 2, edges: 1 }); + // The second pass confirms the concepts and updates the edge in place. + expect(second).toEqual({ nodes: 0, edges: 1 }); + const rows = db.prepare("SELECT * FROM gm_edges").all() as any[]; + expect(rows.length).toBe(1); + expect(rows[0].instruction).toBe("依赖本地模型"); + }); + + it("ignores a self-referencing relation", () => { + const result = applyConceptExtraction(db, SID, seedTurn(1), + [{ name: "抽取路由", type: "SKILL", description: "路由配置" }], + [{ from: "抽取路由", to: "抽取路由", type: "RELATES", instruction: "自指" }], + ); + expect(result).toEqual({ nodes: 1, edges: 0 }); + }); + + it("is a no-op when the turn declared no concepts", () => { + expect(applyConceptExtraction(db, SID, seedTurn(1), [], [])).toEqual({ nodes: 0, edges: 0 }); + }); + + // TypeBox's runtime check enforces shape and length but not `enum`, so the + // kind sets are re-checked at persistence time rather than trusted. + it("skips a concept whose kind is outside the declared set", () => { + const result = applyConceptExtraction(db, SID, seedTurn(1), + [{ name: "抽取路由", type: "CONCEPT" as any, description: "未声明类型" }], []); + expect(result).toEqual({ nodes: 0, edges: 0 }); + expect(findByName(db, "抽取路由")).toBeNull(); + }); + + it("skips a relation whose kind is outside the declared set", () => { + const result = applyConceptExtraction(db, SID, seedTurn(1), + [ + { name: "甲", type: "SKILL", description: "甲" }, + { name: "乙", type: "SKILL", description: "乙" }, + ], + [{ from: "甲", to: "乙", type: "LIKES" as any, instruction: "未声明关系" }]); + expect(result).toEqual({ nodes: 2, edges: 0 }); + }); +}); diff --git a/test/consolidate-cluster.test.ts b/test/consolidate-cluster.test.ts new file mode 100644 index 0000000..639c384 --- /dev/null +++ b/test/consolidate-cluster.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, it } from "vitest"; + +import { + clusterByVector, + cosineSimilarity, + findPotentialConflicts, + kMeans, +} from "../src/consolidate/cluster.ts"; + +const vec = (...values: number[]) => Float32Array.from(values); + +describe("cosine similarity", () => { + it("returns 1 for identical direction and 0 for orthogonal input", () => { + expect(cosineSimilarity(vec(1, 0), vec(1, 0))).toBeCloseTo(1, 6); + expect(cosineSimilarity(vec(1, 0), vec(0, 1))).toBeCloseTo(0, 6); + }); + + it("returns 0 instead of NaN for degenerate input", () => { + // A zero-norm vector would divide by zero and poison every downstream score. + expect(cosineSimilarity(vec(0, 0), vec(1, 1))).toBe(0); + expect(cosineSimilarity(vec(1, 1), vec(1, 1, 1))).toBe(0); + }); +}); + +describe("k-means", () => { + it("separates two clearly disjoint groups", () => { + const vectors = [vec(0, 0), vec(0, 0.1), vec(9, 9), vec(9.1, 9)]; + const clusters = kMeans(vectors, 2, { random: () => 0.5 }); + expect(clusters).toHaveLength(2); + const sizes = clusters.map(cluster => cluster.length).sort(); + expect(sizes).toEqual([2, 2]); + // Every cluster must be internally tight: no group may mix the two blobs. + for (const cluster of clusters) { + const values = cluster.map(index => vectors[index][0]); + expect(Math.max(...values) - Math.min(...values)).toBeLessThan(1); + } + }); + + it("collapses to a single cluster when k exceeds the input size", () => { + const clusters = kMeans([vec(1, 1), vec(2, 2)], 5); + expect(clusters).toEqual([[0, 1]]); + }); + + it("returns an empty result for no input", () => { + expect(kMeans([], 3)).toEqual([]); + }); + + it("preserves item identity when clustering objects", () => { + const items = [{ id: "a" }, { id: "b" }]; + const grouped = clusterByVector(items, [vec(0, 0), vec(0, 0.1)], 1); + expect(grouped).toHaveLength(1); + expect(grouped[0].map(item => item.id).sort()).toEqual(["a", "b"]); + }); +}); + +describe("potential conflicts", () => { + it("reports a near-duplicate pair once, lower index first", () => { + const items = [{ key: "x", id: 1 }, { key: "x", id: 2 }]; + const pairs = findPotentialConflicts(items, [vec(1, 0), vec(1, 0.01)], item => item.key); + expect(pairs).toHaveLength(1); + expect(pairs[0].a.id).toBe(1); + expect(pairs[0].b.id).toBe(2); + expect(pairs[0].similarity).toBeGreaterThan(0.99); + }); + + it("never pairs items that belong to different groups", () => { + const items = [{ key: "x", id: 1 }, { key: "y", id: 2 }]; + // Identical vectors, different groups: a merge or conflict across kinds of + // memory would be meaningless, so the group key gates the pair. + expect(findPotentialConflicts(items, [vec(1, 0), vec(1, 0)], item => item.key)).toEqual([]); + }); + + it("respects the similarity threshold", () => { + const items = [{ key: "x", id: 1 }, { key: "x", id: 2 }]; + expect(findPotentialConflicts(items, [vec(1, 0), vec(0.7, 0.7)], item => item.key, 0.95)).toEqual([]); + }); +}); diff --git a/test/extraction-contract.test.ts b/test/extraction-contract.test.ts index 5394455..e95e431 100644 --- a/test/extraction-contract.test.ts +++ b/test/extraction-contract.test.ts @@ -39,4 +39,20 @@ describe("graph extraction data contract", () => { it("rejects undeclared fields instead of silently rewriting them", () => { expect(() => assertGraphExtractionContract({ ...validPayload, extra: true })).toThrow("contract violation"); }); + + it("accepts the optional concept blocks alongside the required fields", () => { + const payload = { + ...validPayload, + entities: [{ name: "抽取路由", type: "SKILL", description: "抽取请求使用的模型路由配置" }], + relations: [{ from: "抽取路由", to: "抽取路由", type: "RELATES", instruction: "自指示例" }], + }; + expect(() => assertGraphExtractionContract(payload)).not.toThrow(); + }); + + it("rejects an entity carrying an undeclared field", () => { + expect(() => assertGraphExtractionContract({ + ...validPayload, + entities: [{ name: "抽取路由", type: "SKILL", description: "路由配置", weight: 1 }], + })).toThrow("contract violation"); + }); }); diff --git a/test/recall-pool.test.ts b/test/recall-pool.test.ts new file mode 100644 index 0000000..94c8220 --- /dev/null +++ b/test/recall-pool.test.ts @@ -0,0 +1,100 @@ +import { beforeEach, describe, expect, it } from "vitest"; +import type { DatabaseSyncInstance } from "../src/store/sqlite.ts"; + +import { Recaller } from "../src/recaller/recall.ts"; +import { + saveMessageOnce, + saveTurnVector, + searchTurnMemories, + upsertTurnMemory, +} from "../src/store/store.ts"; +import { DEFAULT_CONFIG } from "../src/types.ts"; +import { createTestDb } from "./helpers.ts"; + +let db: DatabaseSyncInstance; + +beforeEach(() => { + db = createTestDb(); +}); + +function seedMemory(sessionId: string, key: string, summary: string, vector: number[]) { + saveMessageOnce(db, `user-${key}`, sessionId, 1, "user", `问题 ${key}`); + saveMessageOnce(db, `assistant-${key}`, sessionId, 1, "assistant", `答案 ${key}`); + const memory = upsertTurnMemory(db, { + sessionId, + summary, + outcome: "completed", + sources: [ + { messageId: `user-${key}`, turnIndex: 1 }, + { messageId: `assistant-${key}`, turnIndex: 1 }, + ], + }); + saveTurnVector(db, memory.id, summary, vector); + return memory; +} + +function recaller(maxNodes: number) { + const instance = new Recaller(db, { + ...DEFAULT_CONFIG, + recallMaxNodes: maxNodes, + semanticScoreThreshold: 0.8, + }); + instance.setEmbedFn(async () => [1, 0]); + return instance; +} + +describe("recall pool sizing", () => { + it("bounds retrieval by recallMaxNodes when no over-fetch is requested", async () => { + for (let index = 1; index <= 5; index += 1) { + seedMemory("dsh:session-a", `m${index}`, `第 ${index} 条记忆的内容`, [1, 0]); + } + const recalled = await recaller(2).recall("任意问题"); + expect(recalled.turnMemories).toHaveLength(2); + }); + + it("returns the wider pool the host needs to survive its visibility filter", async () => { + for (let index = 1; index <= 5; index += 1) { + seedMemory("dsh:session-a", `m${index}`, `第 ${index} 条记忆的内容`, [1, 0]); + } + // A three-slot pool whose hits all belong to the live session used to be + // emptied by filterDshRecallMemories, injecting nothing for that turn. + const narrow = await recaller(2).recall("任意问题"); + const wide = await recaller(2).recall("任意问题", { overFetch: 3 }); + expect(wide.turnMemories.length).toBeGreaterThan(narrow.turnMemories.length); + expect(wide.turnMemories).toHaveLength(5); + }); + + it("treats a multiplier below one as the historical behaviour", async () => { + for (let index = 1; index <= 4; index += 1) { + seedMemory("dsh:session-a", `m${index}`, `第 ${index} 条记忆的内容`, [1, 0]); + } + const recalled = await recaller(2).recall("任意问题", { overFetch: 0 }); + expect(recalled.turnMemories).toHaveLength(2); + }); +}); + +describe("lexical turn-memory matching", () => { + it("finds a summary from a multi-term query instead of requiring the sentence", () => { + const target = seedMemory("dsh:session-a", "a", "客必得账号关联任务已全部完成", [1, 0]); + seedMemory("dsh:session-a", "b", "配置文件已更新", [0, 1]); + + const hits = searchTurnMemories(db, "客必得 账号 关联", 5); + expect(hits.map(hit => hit.id)).toEqual([target.id]); + }); + + it("strips a question tail so a natural question still matches", () => { + const target = seedMemory("dsh:session-a", "a", "体系7审计已通过", [1, 0]); + const hits = searchTurnMemories(db, "体系7审计好了吗?", 5); + expect(hits.map(hit => hit.id)).toEqual([target.id]); + }); + + it("rejects a match that shares only one term of a two-term query", () => { + seedMemory("dsh:session-a", "a", "发布端口已修复为9090", [1, 0]); + seedMemory("dsh:session-a", "b", "配置文件已更新", [0, 1]); + + // Sharing "配置文件" alone is not evidence for "发布端口,配置文件": + // this exact query used to return five summaries that never mentioned the + // port at all. + expect(searchTurnMemories(db, "发布端口,配置文件", 5)).toEqual([]); + }); +}); diff --git a/test/rerank.test.ts b/test/rerank.test.ts new file mode 100644 index 0000000..bec3186 --- /dev/null +++ b/test/rerank.test.ts @@ -0,0 +1,150 @@ +import { beforeEach, describe, expect, it } from "vitest"; +import type { DatabaseSyncInstance } from "../src/store/sqlite.ts"; + +import { Recaller } from "../src/recaller/recall.ts"; +import { createHttpReranker } from "../src/recaller/rerank.ts"; +import { + saveMessageOnce, + saveTurnVector, + upsertTurnMemory, +} from "../src/store/store.ts"; +import { DEFAULT_CONFIG } from "../src/types.ts"; +import { createTestDb } from "./helpers.ts"; + +let db: DatabaseSyncInstance; + +beforeEach(() => { + db = createTestDb(); +}); + +function seed(sessionId: string, key: string, summary: string) { + saveMessageOnce(db, `user-${key}`, sessionId, 1, "user", `问题 ${key}`); + saveMessageOnce(db, `assistant-${key}`, sessionId, 1, "assistant", `答案 ${key}`); + const memory = upsertTurnMemory(db, { + sessionId, + summary, + outcome: "completed", + sources: [ + { messageId: `user-${key}`, turnIndex: 1 }, + { messageId: `assistant-${key}`, turnIndex: 1 }, + ], + }); + saveTurnVector(db, memory.id, summary, [1, 0]); + return memory; +} + +function recaller(maxNodes = 4) { + const instance = new Recaller(db, { + ...DEFAULT_CONFIG, + recallMaxNodes: maxNodes, + semanticScoreThreshold: 0.8, + }); + instance.setEmbedFn(async () => [1, 0]); + return instance; +} + +describe("cross-encoder reranking", () => { + it("leaves the embedding order untouched when no reranker is configured", async () => { + seed("dsh:session-a", "a", "第一条记忆"); + seed("dsh:session-a", "b", "第二条记忆"); + const recalled = await recaller().recall("任意问题"); + const plain = await recaller().recall("任意问题"); + expect(recalled.turnMemories.map(m => m.id)).toEqual(plain.turnMemories.map(m => m.id)); + }); + + it("reorders the candidate head by the scores it is given", async () => { + const first = seed("dsh:session-a", "a", "第一条记忆"); + const second = seed("dsh:session-a", "b", "第二条记忆"); + const instance = recaller(); + // Score by document text: whichever summary is "第二条" must come first. + instance.setRerankFn(async (_query, documents) => + documents.map(text => (text.includes("第二条") ? 0.9 : 0.1))); + const recalled = await instance.recall("任意问题"); + expect(recalled.turnMemories[0].id).toBe(second.id); + expect(recalled.turnMemories.map(m => m.id)).toContain(first.id); + }); + + it("falls back to the embedding order when the reranker returns null", async () => { + seed("dsh:session-a", "a", "第一条记忆"); + seed("dsh:session-a", "b", "第二条记忆"); + const baseline = (await recaller().recall("任意问题")).turnMemories.map(m => m.id); + const instance = recaller(); + instance.setRerankFn(async () => null); + const recalled = await instance.recall("任意问题"); + expect(recalled.turnMemories.map(m => m.id)).toEqual(baseline); + }); + + it("survives a throwing reranker without losing recall", async () => { + seed("dsh:session-a", "a", "第一条记忆"); + seed("dsh:session-a", "b", "第二条记忆"); + const baseline = (await recaller().recall("任意问题")).turnMemories.map(m => m.id); + const instance = recaller(); + instance.setRerankFn(async () => { + throw new Error("reranker down"); + }); + const recalled = await instance.recall("任意问题"); + expect(recalled.turnMemories.map(m => m.id)).toEqual(baseline); + }); + + it("keeps candidates beyond the scored head in their original order", async () => { + seed("dsh:session-a", "a", "记忆一"); + seed("dsh:session-a", "b", "记忆二"); + seed("dsh:session-a", "c", "记忆三"); + seed("dsh:session-a", "d", "记忆四"); + const instance = new Recaller(db, { + ...DEFAULT_CONFIG, + recallMaxNodes: 4, + semanticScoreThreshold: 0.8, + rerankTopN: 2, + }); + instance.setEmbedFn(async () => [1, 0]); + let scored = 0; + instance.setRerankFn(async (_query, documents) => { + scored = documents.length; + return documents.map(() => 0.5); + }); + await instance.recall("任意问题"); + // Only the head is worth a cross-encoder pass; the tail must stay unpriced. + expect(scored).toBe(2); + }); +}); + +describe("http reranker client", () => { + it("maps a Jina/Cohere-shaped response back onto the document order", async () => { + const rerank = createHttpReranker({ + endpoint: "http://stub.invalid/rerank", + timeoutMs: 50, + }); + // The client must not depend on results arriving in input order. + const original = globalThis.fetch; + globalThis.fetch = (async () => ({ + ok: true, + json: async () => ({ + results: [ + { index: 2, relevance_score: 0.9 }, + { index: 0, relevance_score: 0.5 }, + { index: 1, relevance_score: 0.1 }, + ], + }), + })) as any; + try { + expect(await rerank("q", ["a", "b", "c"])).toEqual([0.5, 0.1, 0.9]); + } finally { + globalThis.fetch = original; + } + }); + + it("returns null rather than a partial ordering when the payload is short", async () => { + const rerank = createHttpReranker({ endpoint: "http://stub.invalid/rerank", timeoutMs: 50 }); + const original = globalThis.fetch; + globalThis.fetch = (async () => ({ + ok: true, + json: async () => ({ results: [{ index: 0, relevance_score: 0.9 }] }), + })) as any; + try { + expect(await rerank("q", ["a", "b"])).toBeNull(); + } finally { + globalThis.fetch = original; + } + }); +});