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 项目中为 Trae(以及 Claude Code、Cursor、Codex、Gemini CLI 等)生成的 Agent 技能参考文档 graphify/skills/trae/references/query.md(及其 skillgen 产物 tools/skillgen/expected/graphify__skills__trae__references__query.md)为骨架,完整讲解在图谱构建完成之后如何回答自然语言问题:先做受限于图谱词表的查询扩展,再执行 BFS/DFS 遍历、最短路径(/graphify path)与单节点解释(/graphify explain),最后通过save-result与reflect把答案沉淀成可复用的"工作记忆"。读完你将掌握一套从查询到回写、可审计且不会虚构边的完整图查询工作流,并了解其底层实现位于 graphify/serve.py、graphify/cli.py 与 graphify/reflect.py 中的哪些位置。
适用场景与文档定位
该参考文档约定在以下时机被加载:
- 用户针对一个已存在的图谱提出自然语言问题(例如"X 是怎么工作的?""谁调用了 Y?");
- 用户运行了
/graphify path或/graphify explain。
此时核心技能文件会把查询流程的完整实现委托给本文档(query stub 指向此处)。技能主入口文件 graphify/skill-trae.md 的## Usage块中给出了这些命令的用户视角:
/graphify query "<question>" # BFS traversal - broad context /graphify query "<question>" --dfs # DFS - trace a specific path /graphify query "<question>" --budget 1500 # cap answer at N tokens /graphify path "AuthModule" "Database" # shortest path between two concepts /graphify explain "SwinTransformer" # plain-language explanation of a node整个流程遵循两条统一约定:
- 优先使用
graphifyCLI(安装后执行graphify query/path/explain); - CLI 不可用时降级为内联 NetworkX 遍历:读取
graphify-out/graph.json,用networkx.readwrite.json_graph.node_link_graph重建图并就地遍历。
这保证了无论运行环境是否装有完整 CLI,Agent 都能完成同样质量的查询。
两种遍历模式:BFS 与 DFS
文档开篇就要求根据问题的形态选择遍历算法,二者在graphify query中通过--dfs开关区分:
| Mode | Flag | Best for |
|---|---|---|
| BFS (default) | (none) | "What is X connected to?" - broad context, nearest neighbors first |
| DFS | --dfs | "How does X reach Y?" - trace a specific chain or dependency path |
通俗地讲:
- BFS(默认)回答"扩散型"问题——想知道某节点周边有哪些邻居、拿到广而浅的上下文时使用;
- DFS回答"追溯型"问题——想知道一条具体调用链或依赖链上每一步时使用,一路钻到底再回溯。
前置检查:确认图谱已构建
任何遍历开始之前必须先确认graphify-out/graph.json存在,否则直接报错并提示用户先构建图谱:
$(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>构建图谱——不要凭空回答。这里读取graphify-out/.graphify_python(技能在首次运行步骤中写入的解释器路径)来保证后续所有 Python 片段使用同一解释器环境。
Step 0 —— 受限查询扩展(遍历前必做)
为什么要做扩展:字面匹配的固有缺陷
文档明确声明:graphify queryCLI 内部对节点的匹配方式是大小写折叠后的子串 + IDF(逆向文档频率)加权——没有词干还原、没有同义词、没有跨语言匹配;下方内联回退逻辑的行为与之完全一致。
这一限制在源码中可以得到印证:查询打分与种子选择实现在 graphify/serve.py,其中 serve.py#L272-L291 的_query_terms()仅做分词与停用词过滤,而 serve.py#L294-L297 定义的匹配层级是纯字符层面的加分项:
_EXACT_MATCH_BONUS = 1000.0 _PREFIX_MATCH_BONUS = 100.0 _SUBSTRING_MATCH_BONUS = 1.0 _SOURCE_MATCH_BONUS = 0.5可见它只比较字符形态(精确/前缀/子串),并不理解语义。因此当用户问题的用词与图谱节点标签不在同一语言或同一领域词汇表内时,字面匹配会返回 0 命中:
- 用户说俄语 "обработчик",而图谱标签是英文 "handler";
- 用户说 "authentication",而图谱里的概念名是 "Guardian"。
结果就是答案坍缩成噪声。文档给出的解法不是引入"猜词",而是先针对真实图谱词表做一次可审计的查询扩展——绝不发明词元(token)。
第 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') "要点解析:
- 用
re.findall(r'[^\W\d_]+', ...)按 Unicode 规则切出单词; - 再用驼峰/大写拆分正则把
camelCase、PascalCase、HTTPClient等标识符拆成基础词元(例如JSONParser→json、parser); - 仅保留长度在 3~30 之间的小写词元,长度过滤与下文遍历脚本中
len(t) >= 3的阈值保持一致; - 结果写入
graphify-out/.vocab.txt并打印词表规模。
第 2 步:从词表中挑选(最多 12 个)语义匹配词元
读取graphify-out/.vocab.txt后,针对用户问题从中选出语义上匹配查询意图、且不超过 12 个的词元。硬性约束如下(这是防幻觉的关键):
- 只能挑选词表中真实存在的词元,禁止自造;
- 若某个查询概念在词表中找不到合理词元,跳过它——绝不用训练记忆中的近义词顶替;
- 若没有任何词元匹配查询,则输出空列表,并明确告诉用户该语料对这个问题没有相关词汇,不要伪造一次搜索;
- 跨语言翻译:俄语 "аутентификация" → 仅在词表存在时才寻找
auth、credential、token、security; - 形态还原:
handlers仅当词表中存在handler时映射过去;todos仅当存在todo时映射。
第 3 步:把扩展结果显式打印给用户
在正式跑查询之前先输出审计信息,使扩展过程可追溯:
Query expanded to (from graph vocab, N tokens): [token1, token2, ...]若列表为空,就直说并停止——不要进入遍历阶段。
Step 1 —— 遍历:CLI 优先、内联回退
用空格连接已选词元构成扩展后的查询串,作为下文QUESTION的内容——注意不是用户的原始问题(原始问题仅保留到最后的save-result用于归档)。
方式 A:使用 CLI
graphify query "QUESTION" # or: graphify query "QUESTION" --dfs --budget 3000参数说明:
- 无 flag → BFS;
--dfs→ DFS 深度优先;--budget N→ 限制回答的 token 预算(默认 2000,见下文内联脚本中char_budget = token_budget * 4的实现)。
方式 B:CLI 不可用时内联遍历
文档要求按以下顺序处理:找到标签与扩展词元最匹配的 1~3 个起始节点 → 从各起始节点执行相应遍历 → 阅读子图(节点标签、边关系、置信度标签、源码位置)→只用图中存在的内容作答,引用具体事实时注明source_location→ 图谱信息不足就明说,绝不虚构边。
完整内联脚本:
$(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) "把脚本中的三个占位符替换掉后即可运行:
| 占位符 | 替换为 |
|---|---|
QUESTION | 扩展后的查询串 |
MODE | bfs或dfs |
BUDGET | token 预算(默认2000,与--budget N一致) |
实现细节值得注意:
- 起始节点选择:按"词元在标签中命中的次数"打分取前 3,命中为 0 则无起点,直接退出;
- DFS:迭代式栈实现,深度限制 6 层避免遍历整图;
- BFS:逐层扩散 3 层(frontier 推进);
- 输出预算控制:
~4 字符/token的估算,超预算截断并提示可用--budget N增加; - 多图兼容:通过
next(iter(_raw.values()), {})统一处理nx.Graph与nx.MultiGraph两种形态,保证重边场景下也能取到边的属性; - 输出行同时携带
source_file/source_location(节点)与relation/confidence(边),这些字段与图谱中 EXTRACTED/INFERRED/AMBIGUOUS 的置信标注体系对应,是作答时可引用的证据。
从源码角度看,这套"词元种子打分"逻辑与serve.py中真正被 CLI 使用的_score_query/_score_nodes(见 serve.py#L461-L472 附近的封装)属于同一设计:以词元命中为核心、子串匹配加权,tests/bench_query_scoring.py 专门对查询评分路径做基准验证,CLI 的查询端到端行为由 tests/test_query_cli.py 覆盖。
作答要求
拿到上述子图输出后作答,遵守铁律:只用图中内容;引用事实要引用source_location;信息不足要承认,绝不凭记忆补齐图中不存在的边。
回写与工作记忆:让一次查询变成下一次的起点
save-result:把问答写回图谱
答案写好之后要保存回图,让未来的查询受益。做法是把扩展词元一并放进--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:作答时引用的节点标签列表。
这闭合了反馈环:下一次--update会把该问答抽取成图谱中的一个节点,使查询历史本身成为图的一部分。
为问答标记 outcome:供 reflect 聚合
为让未来的会话能"从这次经验中学习",给save-result追加--outcome(需要纠正时再加--correction):
useful— 引用的节点很好地回答了问题(这些节点将有机会成为preferred sources首选来源);dead_end— 这个问题/路径没有通向任何有价值的结果,下次不要重复推导;corrected— 之前保存的答案是错的,--correction记录正确内容。
会话开始:reflect --if-stale 刷新经验
在开始任何图查询工作前,先刷新并阅读经验教训:
graphify reflect --if-stale然后阅读graphify-out/reflections/LESSONS.md。它会列出首选来源(从此处开始)、已知死胡同(跳过它们)与历史纠正。即使没有安装 git hook,自己运行reflect也能保持经验文件为最新;若 post-commit hook 已安装,--if-stale保证本次会话开始时的运行几乎零成本(当LESSONS.md比所有输入都新时该参数使命令成为空操作)。
reflect的底层实现在 graphify/reflect.py:模块 docstring(reflect.py#L1-L27)说明了这套"确定性工作记忆"的设计——它读取save-result写入的 Q&A 记忆文档,聚合useful / dead_end / corrected信号,输出单一经验文件。值得强调的机制:
- 节点按带符号、时间衰减的分数排序而非简单计数:
useful为正、dead_end/corrected为负,带半衰期——新近的死胡同权重高于几个月前的 useful; - 节点只有被足够多条独立结果佐证后才晋升为 preferred,一次保存不能凭空制造一条可信经验;
- 有图时经验还按社区标签分组,无图则退化为单一扁平章节;
- 确定性、无 LLM:给定相同输入与
now,输出字节级稳定; - 产物放在
graphify-out/reflections/LESSONS.md而不是 wiki 内,因为export wiki每次运行会删除所有wiki/*.md,写进 wiki 会在下次导出时被清掉。
/graphify path:两概念间的最短路径
用途:在图谱中寻找两个命名概念之间的最短路径。
CLI 方式
graphify path "NODE_A" "NODE_B"内联回退
$(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换成用户提到的真实概念名。随后用通俗语言解释路径——每一跳(hop)的含义以及为什么这些连接有意义。处理要点:
find_node同样以词元命中数为依据找最佳匹配节点,任何一端找不到匹配就明确报错;- 健壮性处理:
NetworkXNoPath(两端可达性断裂)、NodeNotFound(节点根本不存在)分别给出清晰提示; - 输出每一跳都带
relation与confidence,解释时必须讲清楚边的语义。
写解释后用以下命令保存(供未来的路径查询复用):
$(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/graphify explain:单节点的通俗解释
用途:对图谱中的单个节点给出通俗解释——它与什么相连、为什么这些连接有意义。
CLI 方式
graphify explain "NODE_NAME"内联回退
$(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 句的解释:该节点是什么、它与哪些东西相连、这些连接为什么重要,并用source_location/source_file作为引用依据。相比查询流程,explain 输出的节点元数据更完整(source、type、degree、邻居及其source_file),便于快速建立对单点角色的整体认识。
保存结果:
$(cat graphify-out/.graphify_python) -m graphify save-result --question "Explain NODE_NAME" --answer "ANSWER" --type explain --nodes NODE_NAMECLI 侧的 explain 行为由 tests/test_explain_cli.py 覆盖验证。
全套流程速查与关键纪律
把三套流程浓缩为一张可操作的清单:
| 流程 | CLI | 内联回退要点 | 回写命令 |
|---|---|---|---|
| 自然语言查询 | graphify query "Q" [--dfs] [--budget N] | 词元打分选 ≤3 起点;DFS 深限 6 层 / BFS 扩散 3 层;按 token 预算截断输出 | save-result --type query --nodes ... |
| 最短路径 | graphify path "A" "B" | nx.shortest_path+ 逐跳 relation/confidence 输出 | save-result --type path_query --nodes A B |
| 单节点解释 | graphify explain "N" | 找最佳匹配节点并列出全部邻居与连接属性 | save-result --type explain --nodes N |
贯穿始终的纪律可以归纳为四条:
- 先扩展后遍历:任何自然语言问题都必须先对照
graphify-out/.vocab.txt做受限扩展并打印审计信息;词表为空则停止。 - 先用图说话:答案只能来自子图输出中的节点、边、
relation、confidence与source_location,禁止虚构不存在的边。 - 答案必须回写:
save-result携带--outcome useful|dead_end|corrected(必要时--correction),让图谱随每次问答自我改进。 - 会话开始先读经验:
graphify reflect --if-stale后读graphify-out/reflections/LESSONS.md,从 preferred sources 起步、绕开已知 dead ends、参考历史 corrections。
在仓库中继续深入
- 参考文档本体:graphify/skills/trae/references/query.md,以及 skillgen 生成的期望产物 tools/skillgen/expected/graphify__skills__trae__references__query.md;
- 技能主文件:graphify/skill-trae.md(含全部
/graphify子命令 Usage),核心技能骨架见 graphify/skill.md; - 查询评分与词元实现:graphify/serve.py,重点看
_query_terms(L272)、停用词表_QUERY_STOPWORDS(L240,含英/德/法/西/葡/意多语言提问词过滤与中文切词处理)、匹配加分常量(L294-L297)以及_score_query/_score_nodes包装(L461-L472); - CLI 子命令注册点:graphify/cli.py——
query在 L1068、save-result在 L1312、reflect在 L1343、path在 L1400、explain在 L1567; - 工作记忆 / reflect 实现:graphify/reflect.py(时间衰减打分、preferred/dead ends/corrections 聚合、
--if-stale与确定性输出机制); - 测试与基准:tests/test_query_cli.py、tests/test_explain_cli.py、tests/bench_query_scoring.py。
【免费下载链接】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
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考