跨文档问「人」和「事」时，最贵的一步往往不是向量检索，而是让模型在当次请求里临时连读切片、现场拼关系。把实体、别名和共现边放进入库流水线离线织好，在线只用子图收窄候选，再跑 BM25 或带过滤的向量检索，路径更短，也更容易回滚。下面用 Node.js 标准库给出一套可落盘、可版本化的织图与约束召回示例，不引入虚构的图谱框架。
当场连读为什么不适合当主路径
多跳问题的典型形态是：答案不在单篇文档里。例如「谁主导了某次迁移，后来又和哪条产品线有关」——人名出现在事故复盘，产品线出现在组织公告，两边只靠同一项目、同一值班表或同一时间窗口发生共现。若把希望寄托在当次请求里检索若干切片、再让模型把它们通读一遍并现场推断关系，Demo 往往能跑通，线上却会重复付三笔账。
第一笔是费用和延迟。通读轮次一加，上下文随切片线性变长，超时和截断都会出现。第二笔是稳定性。模型对「两段文字是否构成同一条人事关系」没有一致标准，同样的两段有时看成同事，有时看成路人，评测集上一问一答很难复现。第三笔是关系不落盘。下一问还得再读一遍，缓存的是生成结果而不是可复用的约束。
更稳的做法是把约束前移到入库。切片写入检索索引的同时，抽出实体、合并别名、按共现加边。在线阶段先根据问题里的实体走有限跳数，得到允许进入检索的 chunkId 集合，再在这个子集上跑 BM25，或把同一批 ID 当作向量索引的 filter。图在这里不是知识的最终真相，只是召回阶段的硬过滤。抽错了可以改别名、回收边；文档改了可以按切片撤证据。这比给智能体再加一轮通读更容易评测，也更容易回滚。
需要克制预期：共现只说明「一起出现过」，不说明「汇报关系」或「因果关系」。把它当作候选收窄器，而不是把图谱直接当成答案。
离线织图：实体、别名与共现边
实体用稳定 ID，表面词放在别名表。共现边做成无向边，权重用共同出现过的切片数，边上保存 evidence（chunkId 列表）。同一切片内实体两两相连即可；窗口切得太大，整篇文档所有实体完全图化，约束会失效。别名表与图分离：合并「徐康」和 Xu Kang 只改别名，不必重抽全部共现。
抽取不必一上来上模型。词典最长匹配可复现、可回滚，错误形态也清楚。生产环境可以换成已有的抽取服务，但落盘结构和边回收逻辑建议保持不变：节点是 ID，边带证据，快照带版本。
将下面保存为 graph-recall.mjs，执行 node graph-recall.mjs。脚本会织第一版图、做一次约束召回、演示按切片删边，并把当前指针回滚到 v1。
import {
mkdirSync,
writeFileSync,
readFileSync,
existsSync,
} from "node:fs";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
const root = dirname(fileURLToPath(import.meta.url));
const dataDir = join(root, "data");
const documents = [
{
chunkId: "c1",
text: "徐康负责支付迁移。迁移期间与林舟共同值班，处理对账差异。",
},
{
chunkId: "c2",
text: "林舟随后接手商户平台，并在季度复盘中同步了支付迁移的残留问题。",
},
{
chunkId: "c3",
text: "商户平台的发票能力由陈楠设计，与支付迁移无直接人员重叠。",
},
];
const aliasTable = [
{ id: "e:xukang", names: ["徐康", "Xu Kang"] },
{ id: "e:linzhou", names: ["林舟"] },
{ id: "e:chennan", names: ["陈楠"] },
{ id: "e:pay-migration", names: ["支付迁移"] },
{ id: "e:merchant", names: ["商户平台"] },
];
function buildAliasIndex(table) {
const surfaceToId = new Map();
const names = [];
for (const ent of table) {
for (const name of ent.names) {
surfaceToId.set(name, ent.id);
names.push(name);
}
}
names.sort((a, b) => b.length - a.length);
return { surfaceToId, names };
}
function extractEntities(text, index) {
const found = new Map();
let i = 0;
while (i < text.length) {
let hit = null;
for (const name of index.names) {
if (text.startsWith(name, i)) {
hit = name;
break;
}
}
if (hit) {
const id = index.surfaceToId.get(hit);
if (!found.has(id)) found.set(id, { id, name: hit });
i += hit.length;
} else {
i += 1;
}
}
return [...found.values()];
}
function addUndirectedEdge(adj, a, b, chunkId) {
if (a === b) return;
const [x, y] = a < b ? [a, b] : [b, a];
const key = `${x}|${y}`;
if (!adj.has(key)) {
adj.set(key, { source: x, target: y, evidence: new Set() });
}
adj.get(key).evidence.add(chunkId);
}
function weave(docs, table) {
const index = buildAliasIndex(table);
const chunkText = {};
const chunkEntities = {};
const adj = new Map();
for (const doc of docs) {
chunkText[doc.chunkId] = doc.text;
const ents = extractEntities(doc.text, index);
chunkEntities[doc.chunkId] = ents.map((e) => e.id);
for (let i = 0; i < ents.length; i += 1) {
for (let j = i + 1; j < ents.length; j += 1) {
addUndirectedEdge(adj, ents[i].id, ents[j].id, doc.chunkId);
}
}
}
const edges = [...adj.values()].map((e) => ({
source: e.source,
target: e.target,
weight: e.evidence.size,
evidence: [...e.evidence],
}));
return {
version: 1,
parent: null,
aliasTable: table,
chunkText,
chunkEntities,
edges,
};
}
function applyDeletes(graph, deletedChunkIds) {
const deleted = new Set(deletedChunkIds);
const chunkText = { ...graph.chunkText };
const chunkEntities = { ...graph.chunkEntities };
for (const id of deleted) {
delete chunkText[id];
delete chunkEntities[id];
}
const edges = [];
for (const e of graph.edges) {
const evidence = e.evidence.filter((id) => !deleted.has(id));
if (evidence.length === 0) continue;
edges.push({ ...e, evidence, weight: evidence.length });
}
return {
...graph,
version: graph.version + 1,
parent: graph.version,
chunkText,
chunkEntities,
edges,
};
}
function saveSnapshot(graph) {
mkdirSync(dataDir, { recursive: true });
const file = `graph-v${graph.version}.json`;
writeFileSync(join(dataDir, file), JSON.stringify(graph, null, 2), "utf8");
writeFileSync(
join(dataDir, "current.json"),
JSON.stringify({ version: graph.version, file }, null, 2),
"utf8",
);
return file;
}
function loadCurrent() {
const ptr = JSON.parse(readFileSync(join(dataDir, "current.json"), "utf8"));
return JSON.parse(readFileSync(join(dataDir, ptr.file), "utf8"));
}
function rollback(version) {
const file = `graph-v${version}.json`;
if (!existsSync(join(dataDir, file))) {
throw new Error(`missing ${file}`);
}
writeFileSync(
join(dataDir, "current.json"),
JSON.stringify({ version, file }, null, 2),
"utf8",
);
}
function neighbors(graph, entityId, minWeight) {
const out = [];
for (const e of graph.edges) {
if (e.weight < minWeight) continue;
if (e.source === entityId) out.push(e.target);
else if (e.target === entityId) out.push(e.source);
}
return out;
}
function expand(graph, seedIds, hops = 1, minWeight = 1) {
let frontier = new Set(seedIds);
const seen = new Set(seedIds);
for (let h = 0; h < hops; h += 1) {
const next = new Set();
for (const id of frontier) {
for (const n of neighbors(graph, id, minWeight)) {
if (!seen.has(n)) {
seen.add(n);
next.add(n);
}
}
}
frontier = next;
}
return seen;
}
function chunksForEntities(graph, entityIds) {
const allow = [];
for (const [chunkId, ents] of Object.entries(graph.chunkEntities)) {
if (ents.some((id) => entityIds.has(id))) allow.push(chunkId);
}
return allow;
}
function tokenize(text) {
const units = [];
let buf = "";
const flush = () => {
if (buf) {
units.push(buf.toLowerCase());
buf = "";
}
};
for (const ch of text) {
if (/[a-zA-Z0-9]/.test(ch)) buf += ch;
else {
flush();
if (/[\u4e00-\u9fff]/.test(ch)) units.push(ch);
}
}
flush();
const grams = [];
for (let i = 0; i < units.length; i += 1) {
grams.push(units[i]);
if (i + 1 < units.length) grams.push(units[i] + units[i + 1]);
}
return grams;
}
function bm25(query, chunkIds, chunkText, k1 = 1.2, b = 0.75) {
const docs = chunkIds.map((id) => ({
id,
tokens: tokenize(chunkText[id] || ""),
}));
const nDocs = docs.length || 1;
const avgdl = docs.reduce((sum, d) => sum + d.tokens.length, 0) / nDocs;
const df = new Map();
for (const d of docs) {
for (const t of new Set(d.tokens)) {
df.set(t, (df.get(t) || 0) + 1);
}
}
const qTokens = tokenize(query);
const scored = docs.map((d) => {
const tfMap = new Map();
for (const t of d.tokens) tfMap.set(t, (tfMap.get(t) || 0) + 1);
let score = 0;
const dl = d.tokens.length || 1;
for (const t of qTokens) {
const tf = tfMap.get(t) || 0;
if (!tf) continue;
const n = df.get(t) || 0;
const idf = Math.log((nDocs - n + 0.5) / (n + 0.5) + 1);
score +=
idf *
((tf * (k1 + 1)) / (tf + k1 * (1 - b + b * (dl / avgdl))));
}
return { id: d.id, score: Number(score.toFixed(4)) };
});
scored.sort((a, b) => b.score - a.score);
return scored;
}
function constrainedRecall(graph, query, hops = 1) {
const index = buildAliasIndex(graph.aliasTable);
const seeds = extractEntities(query, index).map((e) => e.id);
const expanded = expand(graph, seeds, hops, 1);
const allowed = chunksForEntities(graph, expanded);
const ranked = bm25(query, allowed, graph.chunkText);
return { seeds, expanded: [...expanded], allowed, ranked };
}
const v1 = weave(documents, aliasTable);
saveSnapshot(v1);
const query = "徐康后来和哪条产品线有关";
console.log("v1 recall", constrainedRecall(v1, query, 1));
const v2 = applyDeletes(v1, ["c3"]);
saveSnapshot(v2);
rollback(1);
console.log("current pointer", loadCurrent().version);语料很小，示例里边的最低权重取 1；真实库应提高到 2 或 3，避免一次误抽把两个无关实体焊死。跑完后 ./data/graph-v1.json 与 current.json 会留下，下一次在线服务只读指针，不现场织图。
在线：子图收窄后再检索
查询侧必须用同一张别名表。问句里有多个实体时，先取并集再向外扩，避免只命中中心节点、漏掉桥接文档。跳数默认 1，最多 2：再往外走，候选会迅速接近全库，约束等于没做。收窄规则是「扩展后的实体集合」映射回切片，而不是要求查询里的所有实体出现在同一篇文档——后者会直接把多跳打回单跳。
上面的例子里，问句只抽出 徐康。一跳后进入 林舟 与 支付迁移，于是 c1、c2 进入候选，c3 被丢掉。这正是约束召回要的效果：桥接人留下，无关的发票设计离开集合。随后 BM25 只在允许的切片上打分。中文用单字加邻接二元，不依赖外部分词器；若已有词法索引，把 allowed 当成 filter 即可，不必换打分公式。
向量检索走同一条路径：子图产出 chunkId，再交给现有索引做 ID 过滤。不要在当次请求里为了「补关系」去把过滤前的长尾切片塞进模型。生成阶段需要通读时，读的应是约束之后的前几条，而不是用通读来发现该读哪篇。
评测不要看模型是否把故事讲圆。固定图版本与查询集，对比「全库 BM25」和「子图约束后 BM25」的命中切片、漏召回和空结果率。约束过死会丢桥接文档，过松则退回全库；跳数、最低权重、是否用查询实体并集，都应变成可调参数，而不是写死在提示词里。
抽错、文档变更与回滚
词典和模型都会抽错。不要让错误边永久污染图，准备三层补丁。第一，别名表可单独发版：一次错误的字符串对齐，用别名合并或拆分修复，不必重跑全部共现。第二，边必须带证据；删除或改写某篇文档时，从所有边的 evidence 里去掉对应 chunkId，权重改为剩余证据数，证据为空则删边。第三，在线扩跳忽略低权重边，把一次性共现挡在召回之外。
文档变更不要原地改文件。织图任务写成「读旧快照 + 变更集 → 写新快照」：输入是切片的 upsert/delete 列表，输出是 graph-vN.json 和指针。示例里的 applyDeletes 只覆盖删除；更新应先按旧 chunkId 删边，再把新文本当作新切片织进去，避免同一 ID 上残留过期共现。父版本写入快照，是为了出问题时知道从哪一版来。
在线服务只认 current.json。回滚就是改指针，而不是在图数据库里做一堆难以撤销的 DELETE。抽取质量突然下降时，先停织图任务、回滚指针，再修词典。图版本应和向量索引版本对齐，避免「图还认为这篇在、倒排里已经没了」的空窗。任务本身要可重入：同一变更集打两次，应得到同一版结果，而不是边权重被加两次。
若一定要给边加类型（同事、从属、前后任），把它当作可选标注，不要挡在召回主路径上。类型抽错的代价高于共现抽错：前者会让过滤条件在语义上看起来很强，实际上更脆。
总结
多跳召回的稳定性，来自把关系从「当场编」改成「入库时记账」。实体 ID、别名、带证据的共现边、版本化快照，这四件套就够支撑约束召回和回滚。在线路径保持短：抽实体、子图收窄、子集上 BM25 或向量过滤。通读留给最终生成，而不是用来发现该读哪篇文档。先把可回滚的织图任务和固定版本的召回评测跑通，再考虑要不要上更重的关系类型或有向边。
评论