graphify 查询参考深度解析：query / path / explain 的受控查询扩展、遍历流程与自改进反馈闭环

graphify 查询参考深度解析：query / path / explain 的受控查询扩展、遍历流程与自改进反馈闭环

【免费下载链接】graphifyTurn any codebase, with its docs, SQL schemas, configs, and PDFs, into a queryable knowledge graph. A /graphify skill for Claude Code, Cursor, Codex, and Gemini CLI: local deterministic AST parsing, every edge explained, no vector store.项目地址: https://gitcode.com/GitHub_Trending/graph/graphify

本文基于 graphify 仓库中 Droid Agent 的技能参考文档 query.md，完整讲解对已构建知识图谱提问的三个核心流程：query（邻域遍历）、path（最短路径）、explain（单节点解释），以及"受控查询扩展（constrained query expansion）"这一防止零命中退化的前置步骤、save-result反馈回路与reflect工作记忆机制。读完后你可以掌握：如何在不发明词元的前提下把自然语言问题映射到图词汇表、如何用 CLI 或内联 NetworkX 脚本执行 BFS/DFS 遍历、如何把答案写回图谱形成自改进闭环。

1. 文档定位与触发场景

query.md是 graphify 面向 Droid Agent 的/graphify技能参考之一（同目录还有 update.md、exports.md、hooks.md 等，其他 Agent 平台如 Claude、Codex、Copilot 拥有各自独立副本）。文档开篇明确了加载条件与双轨执行策略：

• 加载时机：用户对已有图谱提问，或运行/graphify path、/graphify explain时；核心 query stub 指向这里获取完整遍历流程。
• 双轨执行：优先使用graphify queryCLI；CLI 不可用时，回退到内联 NetworkX 遍历脚本。两条路径的匹配语义保持一致（大小写折叠的子串 + IDF 式打分），保证结果可预期。

关键前提是图谱必须已存在。文档给出的检查方式（注意：Python 解释器路径来自构建时写入的标记文件graphify-out/.graphify_python，这是 graphify 保证环境一致性的做法）：

$(cat graphify-out/.graphify_python) -c " from pathlib import Path if not Path('graphify-out/graph.json').exists(): print('ERROR: No graph found. Run /graphify <path> first to build the graph.') raise SystemExit(1) "

检查失败时应停止并提示用户先运行/graphify <path>构建图谱，而不是凭空作答。

2. 两种遍历模式的选择

文档用一个决策表区分两种遍历模式：

这个选择在源码中同样成立：graphify query的 CLI 入口解析--dfs标志后，以mode="dfs"或mode="bfs"调用统一遍历入口（见 cli.py 中cmd == "query"分支）。

3. Step 0 —— 受控查询扩展（遍历前必做）

这是整份参考中最具方法论价值的部分。文档指出 graphify 的queryCLI 通过大小写折叠子串 + IDF匹配节点，二进制内部没有词干提取、没有同义词、没有跨语言匹配；内联回退脚本也是同样的匹配方式。因此当用户的提问与图谱标签在语言或领域词汇上不一致时（用户说 "обработчик"，图里是 "handler"；用户说 "authentication"，图里是 "Guardian"），字面匹配器会返回 0 命中，答案退化为噪声。

修复思路是：不发明任何词元，先把查询扩展映射到图的真实词汇表上。

3.1 从节点标签抽取词汇表

$(cat graphify-out/.graphify_python) -c " import json, re from pathlib import Path data = json.loads(Path('graphify-out/graph.json').read_text(encoding='utf-8')) vocab = set() for n in data['nodes']: for c in re.findall(r'[^\W\d_]+', n.get('label','') or '', re.UNICODE): parts = re.findall(r'[A-Z]+(?=[A-Z][a-z])|[A-Z]?[a-z]+|[A-Z]+', c) or [c] for p in parts: t = p.lower() if 3 <= len(t) <= 30: vocab.add(t) Path('graphify-out/.vocab.txt').write_text('\n'.join(sorted(vocab)), encoding='utf-8') print(f'vocab: {len(vocab)} tokens') "

注意其中的词元切分细节：正则[A-Z]+(?=[A-Z][a-z])|[A-Z]?[a-z]+|[A-Z]+同时处理了驼峰命名（parseConfig→parse、config）和全大写缩写（API），长度阈值3 <= len(t) <= 30过滤掉噪声碎片。

3.2 词汇表内的硬约束选择

读取graphify-out/.vocab.txt后，针对用户问题从这份精确列表中挑选至多 12 个与查询意图语义匹配的词元，文档给出了四条硬约束：

1. 只能选词汇表中存在的词元，绝不发明词元；
2. 若某个查询概念在词汇表中没有合理对应，跳过它，不要用训练记忆里的近义替换词；
3. 若没有任何词汇表词元匹配查询，输出空列表并明确告知用户"语料中没有相关词汇"，不得伪造搜索；
4. 跨语言翻译只在词元存在时生效：俄语 "аутентификация" → 仅当词汇表存在时选auth、credential、token、security；形态变化同理，"handlers" → 仅当存在时映射handler。

3.3 显式打印扩展结果以便审计

Query expanded to (from graph vocab, N tokens): [token1, token2, ...]

列表为空时要直说并停止，不进入遍历。这一设计让 LLM 的"改写"变得可审计——扩展完全受限于语料自身词汇，与文档"不得虚构事实"的总基调一致。

4. Step 1 —— 执行遍历

把选中的词元用空格连接组成扩展查询串，以它作为下面的QUESTION（原始问题仅保留给最后save-result使用）。

4.1 优先走 CLI

graphify query "QUESTION" # or: graphify query "QUESTION" --dfs --budget 3000

从源码看，graphify query完整支持--dfs、--budget N（默认 2000）、--context C（上下文过滤）、--graph path（指定图谱文件）四个选项（cli.py）。几个值得注意的实现细节：

• query 刻意保持无向图：path/explain会强制directed=True，而query不强制——因为 BFS/DFS 需要同时探索种子节点的调用方和被调用方，若转成 DiGraph，G.neighbors()只会返回后继，种子若无出边则静默丢失所有调用方侧结果；方向性改由每条边上的_src/_tgt标记在渲染时保留（cli.py 有专门注释）。
• 统一遍历入口：CLI 最终调用 serve.py 中的_query_graph_text。该函数用单次打分遍历同时产出综合排名与逐词元的种子候选（注释说明此前 T+1 次全图遍历在 10 万节点基准上浪费约 71% 打分时间）；关系意图动词（"calls"、"uses" 等）会从逐词元种子保证中剔除，防止动词偶然命中抢占 BFS 根（[#2507]）；输出头部会显式打印图谱路径与节点数，避免在父项目目录里误查 vendored 子项目图谱导致"格式正确、内容错误"的静默错答（[#2789]）。
• 留痕与钩子协同：每次 query 会经querylog.log_query记录到查询日志，并调用_touch_query_stamp更新graphify-out/cache/last_query_stamp时间戳——hook guard 依赖该时间戳判断"本会话最近是否查过图谱"，从而决定放行还是拦截直接 grep 源码（cli.py、graphify/security 相关守卫逻辑）。

4.2 CLI 不可用时的内联 NetworkX 回退

文档要求依次完成：找标签与扩展词元最匹配的 1–3 个节点 → 从各起点执行对应遍历 → 阅读子图（节点标签、边关系、置信度标记、源码位置）→仅用图内信息作答，引用具体事实时引用source_location→ 图内信息不足就明说，不得幻觉边。完整内联脚本如下（将QUESTION替换为扩展查询串，MODE替换为bfs/dfs，BUDGET为 token 预算，默认 2000）：

$(cat graphify-out/.graphify_python) -c " import sys, json from networkx.readwrite import json_graph import networkx as nx from pathlib import Path data = json.loads(Path('graphify-out/graph.json').read_text(encoding='utf-8')) G = json_graph.node_link_graph(data, edges='links') question = 'QUESTION' mode = 'MODE' # 'bfs' or 'dfs' terms = [t.lower() for t in question.split() if len(t) >= 3] # match the vocab threshold; keeps api/jwt/ios (#1392) # Find best-matching start nodes scored = [] for nid, ndata in G.nodes(data=True): label = ndata.get('label', '').lower() score = sum(1 for t in terms if t in label) if score > 0: scored.append((score, nid)) scored.sort(reverse=True) start_nodes = [nid for _, nid in scored[:3]] if not start_nodes: print('No matching nodes found for query terms:', terms) sys.exit(0) subgraph_nodes = set() subgraph_edges = [] if mode == 'dfs': # DFS: follow one path as deep as possible before backtracking. # Depth-limited to 6 to avoid traversing the whole graph. visited = set() stack = [(n, 0) for n in reversed(start_nodes)] while stack: node, depth = stack.pop() if node in visited or depth > 6: continue visited.add(node) subgraph_nodes.add(node) for neighbor in G.neighbors(node): if neighbor not in visited: stack.append((neighbor, depth + 1)) subgraph_edges.append((node, neighbor)) else: # BFS: explore all neighbors layer by layer up to depth 3. frontier = set(start_nodes) subgraph_nodes = set(start_nodes) for _ in range(3): next_frontier = set() for n in frontier: for neighbor in G.neighbors(n): if neighbor not in subgraph_nodes: next_frontier.add(neighbor) subgraph_edges.append((n, neighbor)) subgraph_nodes.update(next_frontier) frontier = next_frontier # Token-budget aware output: rank by relevance, cut at budget (~4 chars/token) token_budget = BUDGET # default 2000 char_budget = token_budget * 4 # Score each node by term overlap for ranked output def relevance(nid): label = G.nodes[nid].get('label', '').lower() return sum(1 for t in terms if t in label) ranked_nodes = sorted(subgraph_nodes, key=relevance, reverse=True) lines = [f'Traversal: {mode.upper()} | Start: {[G.nodes[n].get(\"label\",n) for n in start_nodes]} | {len(subgraph_nodes)} nodes'] for nid in ranked_nodes: d = G.nodes[nid] lines.append(f' NODE {d.get(\"label\", nid)} [src={d.get(\"source_file\",\"\")} loc={d.get(\"source_location\",\"\")}]') for u, v in subgraph_edges: if u in subgraph_nodes and v in subgraph_nodes: _raw = G[u][v]; d = next(iter(_raw.values()), {}) if isinstance(G, nx.MultiGraph) else _raw lines.append(f' EDGE {G.nodes[u].get(\"label\",u)} --{d.get(\"relation\",\"\")} [{d.get(\"confidence\",\"\")}]--> {G.nodes[v].get(\"label\",v)}') output = '\n'.join(lines) if len(output) > char_budget: output = output[:char_budget] + f'\n... (truncated at ~{token_budget} token budget - use --budget N for more)' print(output) "

脚本要点值得逐条理解：

• 词元长度阈值>= 3：与 Step 0 词汇表构建保持同一阈值，注释注明这样保留了api/jwt/ios这类 3 字符缩写（[#1392]）；
• 种子选择：按词元重叠计数排序取前 3 个节点，0 命中时打印明确信息后退出；
• 遍历边界：DFS 深度上限 6、BFS 展开 3 层，防止大图上全图遍历；
• Token 预算输出：按~4 chars/token换算字符预算，按相关度排序后截断，并提示可用--budget N放宽；
• MultiGraph 兼容：读取边属性时用isinstance(G, nx.MultiGraph)分支，对应 graphify 的 multigraph 兼容层（graphify/multigraph_compat.py）。

5. 反馈闭环：save-result 与工作记忆

5.1 把答案写回图谱

写完答案后，文档要求将其保存回图谱以改进后续查询，并在--answer文本中包含扩展词元痕迹（例如"Expanded from original query via vocab: [tokens]. Then traversed..."），使下次--update能把扩展历史提取为图节点：

$(cat graphify-out/.graphify_python) -m graphify save-result --question "ORIGINAL_QUESTION" --answer "ANSWER" --type query --nodes NODE1 NODE2

其中ORIGINAL_QUESTION是用户原话，ANSWER是含扩展词元痕迹的完整答案，NODE1 NODE2是所引用的节点标签。

对照源码，graphify save-result的完整参数集比文档示例更丰富（cli.py）：

底层实现在 graphify/ingest.py 的save_query_result。

5.2 工作记忆：三值结果与 reflect

文档要求给save-result追加--outcome，让未来会话从本次学习：

• useful—— 被引用节点很好地回答了问题（它们成为首选来源）；
• dead_end—— 该问题/路径走不通，下次不必重新推导；
• corrected—— 已存答案有误，--correction "the right answer"记录正确答案。

在每次图谱工作开始时刷新并阅读经验：运行graphify reflect --if-stale（廉价、确定性、无 LLM；当LESSONS.md已新于所有输入文件时--if-stale使其成为空操作，例如 git hook 刚刚刷新过），然后读取graphify-out/reflections/LESSONS.md。该文件列出首选来源（从那里开始）、已知死路（跳过）与历史纠正。自己运行reflect可在未安装 git hook 的情况下保持经验最新；若 post-commit hook 已安装，--if-stale使会话启动时的这次运行几乎零成本。

从源码看，graphify reflect还支持若干可调参数（cli.py）：--half-life-days（信号权重半衰期，默认 30 天）、--min-corroboration（将一个节点提升为 preferred 所需的独立 useful 结果数，默认 2）、--out（默认graphify-out/reflections/LESSONS.md）、--if-stale。核心逻辑在 graphify/reflect.py，测试见 tests/test_reflect.py。这条链路（save-result --outcome→ 记忆目录 →reflect→ LESSONS.md → 下次会话读取）构成了 graphify 所称的"自改进循环"。

6. /graphify path：两概念间的最短路径

在图谱中查找两个命名概念间的最短路径。CLI 可用时：

graphify path "NODE_A" "NODE_B"

源码中该命令还支持--graph指定图谱、--directed/--undirected二选一（互斥报错），且默认按有向图计算——因为graph.json中保存了方向真值（cli.py）。CLI 不可用时的内联脚本：

$(cat graphify-out/.graphify_python) -c " import json, sys import networkx as nx from networkx.readwrite import json_graph from pathlib import Path data = json.loads(Path('graphify-out/graph.json').read_text(encoding='utf-8')) G = json_graph.node_link_graph(data, edges='links') a_term = 'NODE_A' b_term = 'NODE_B' def find_node(term): term = term.lower() scored = sorted( [(sum(1 for w in term.split() if w in G.nodes[n].get('label','').lower()), n) for n in G.nodes()], reverse=True ) return scored[0][1] if scored and scored[0][0] > 0 else None src = find_node(a_term) tgt = find_node(b_term) if not src or not tgt: print(f'Could not find nodes matching: {a_term!r} or {b_term!r}') sys.exit(0) try: path = nx.shortest_path(G, src, tgt) print(f'Shortest path ({len(path)-1} hops):') for i, nid in enumerate(path): label = G.nodes[nid].get('label', nid) if i < len(path) - 1: _raw = G[nid][path[i+1]]; edge = next(iter(_raw.values()), {}) if isinstance(G, nx.MultiGraph) else _raw rel = edge.get('relation', '') conf = edge.get('confidence', '') print(f' {label} --{rel}--> [{conf}]') else: print(f' {label}') except nx.NetworkXNoPath: print(f'No path found between {a_term!r} and {b_term!r}') except nx.NodeNotFound as e: print(f'Node not found: {e}') "

将NODE_A/NODE_B替换为用户给出的实际概念名。找到路径后，文档要求用自然语言解释这条路径——每一跳意味着什么、为何重要。解释完成后同样写回：

$(cat graphify-out/.graphify_python) -m graphify save-result --question "Path from NODE_A to NODE_B" --answer "ANSWER" --type path_query --nodes NODE_A NODE_B

7. /graphify explain：单节点全景解释

对单个节点给出自然语言解释——以及与之相连的一切。CLI 可用时：

graphify explain "NODE_NAME"

CLI 实现比内联脚本多一层歧义防护：当同名节点分布在多个文件时，explain会打印所有竞争节点（source_file+ 节点 id）并以退出码 1 要求改用仓库相对路径或完整节点 id 重试（cli.py）。它还会输出community（社区归属），并在节点存在.graphify_learning.json伴随文件中的经验条目时叠加一行由reflect派生的经验提示（display-only，不改变图数据）。CLI 不可用时的内联脚本：

$(cat graphify-out/.graphify_python) -c " import json, sys import networkx as nx from networkx.readwrite import json_graph from pathlib import Path data = json.loads(Path('graphify-out/graph.json').read_text(encoding='utf-8')) G = json_graph.node_link_graph(data, edges='links') term = 'NODE_NAME' term_lower = term.lower() # Find best matching node scored = sorted( [(sum(1 for w in term_lower.split() if w in G.nodes[n].get('label','').lower()), n) for n in G.nodes()], reverse=True ) if not scored or scored[0][0] == 0: print(f'No node matching {term!r}') sys.exit(0) nid = scored[0][1] data_n = G.nodes[nid] print(f'NODE: {data_n.get(\"label\", nid)}') print(f' source: {data_n.get(\"source_file\",\"unknown\")}') print(f' type: {data_n.get(\"file_type\",\"unknown\")}') print(f' degree: {G.degree(nid)}') print() print('CONNECTIONS:') for neighbor in G.neighbors(nid): _raw = G[nid][neighbor]; edge = next(iter(_raw.values()), {}) if isinstance(G, nx.MultiGraph) else _raw nlabel = G.nodes[neighbor].get('label', neighbor) rel = edge.get('relation', '') conf = edge.get('confidence', '') src_file = G.nodes[neighbor].get('source_file', '') print(f' --{rel}--> {nlabel} [{conf}] ({src_file})') "

将NODE_NAME替换为用户询问的概念。随后写 3–5 句解释：这个节点是什么、连接了谁、这些连接为何重要，并以源码位置作为引用。最后同样写回：

$(cat graphify-out/.graphify_python) -m graphify save-result --question "Explain NODE_NAME" --answer "ANSWER" --type explain --nodes NODE_NAME

8. 测试与工程佐证

三条流程在仓库中均有专门的测试覆盖，可作为行为契约参考：

• tests/test_query_cli.py —— query 的 CLI 行为；
• tests/test_path_cli.py —— path 的最短路径与有向/无向选项；
• tests/test_explain_cli.py —— explain 的节点解析与歧义处理；
• tests/test_query_induced_edges.py、tests/test_query_names_its_graph.py —— 遍历输出对"回答来自哪个图谱"的标注行为；
• tests/test_querylog.py —— 查询日志落盘。

9. 小结：这套流程的设计哲学

把 query.md 的三个流程放在一起看，graphify 对"LLM 提问知识图谱"这一场景做了三层防御：

1. 输入侧：受控查询扩展把自然语言到图谱词汇的映射限制在语料自身词表内，零命中时明说而非伪造，扩展结果显式打印可审计；
2. 执行侧：CLI 优先、内联 NetworkX 回退双轨执行，BFS/DFS 明确分工且都有深度/预算上限，输出按相关度排序并带 token 预算截断；
3. 输出侧：答案只能来自图内容、引用必须带source_location，且每次作答都经save-result（含三值 outcome）写回，再经reflect蒸馏为 LESSONS.md 供下次会话读取——查询行为本身成为图谱的一等数据。

对于要在自己的项目中复用这套流程的开发者，建议按"先读 extract 理解节点/边结构 → 按本文流程执行查询 → 结合 update.md 维护增量"的顺序阅读 Droid 技能参考目录，即可获得 graphify 查询能力的完整操作面。

【免费下载链接】graphifyTurn any codebase, with its docs, SQL schemas, configs, and PDFs, into a queryable knowledge graph. A /graphify skill for Claude Code, Cursor, Codex, and Gemini CLI: local deterministic AST parsing, every edge explained, no vector store.项目地址: https://gitcode.com/GitHub_Trending/graph/graphify