graphify 知识图谱查询实战：query / path / explain 与可审计的答案闭环
graphify 知识图谱查询实战：query / path / explain 与可审计的答案闭环
项目地址: https://gitcode.com/GitHub_Trending/graph/graphify 本指南以 graphify 的 Trae 版查询技能参考（graphify/skills/trae/references/query.md）为主体，系统讲解如何针对已构建好的知识图谱执行三类查询：语义相关节点遍历（query）、两概念间最短路径（path）、单节点关系解释（explain），并展示如何通过受限词汇扩展、NetworkX 内联兜底遍历与 save-result 回写机制形成"查询→审计→沉淀→自我改进"的完整闭环。读完你不仅能跑通三种查询流程，还能掌握如何让一次回答反向滋养后续查询（reflect 课程记忆），且全程不依赖向量库、无需编造图谱中不存在的边。
一、先决条件：先建图，再查询
这三类查询全部面向已生成的图谱文件 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 解释器路径（用于隔离虚拟环境），graphify-out/graph.json 则是默认输出，与 CLI 侧 _default_graph_path() 的实现一致（graphify/cli.py）。
两种遍历模式：按问题形态选择
| 模式 | 标志位 | 适用场景 |
|---|---|---|
| BFS（默认） | （无） | "What is X connected to?"—— 需要广泛上下文、优先展示最近邻居 |
| DFS | --dfs | "How does X reach Y?"—— 需要追踪某条具体的调用链或依赖路径 |
CLI 侧对两种模式均有落地：_query_graph_text（graphify/serve.py）根据 mode 参数分别调用 _dfs / _bfs 遍历函数，而 query 子命令读取 --dfs 标志后把模式传给该函数（graphify/cli.py）。
二、Step 0（必做）：受限查询扩展——先把问句翻译成图谱词汇
graphify 的 query CLI 是通过大小写折叠后的子串匹配 + IDF 打分来寻找起点的：二进制内部没有词干还原、没有同义词、没有跨语言匹配，内联兜底脚本的匹配方式与此一致。因此当用户问句使用的语言或领域词汇与图谱节点标签不一致时（例如用户说俄语 "обработчик"，而图谱标签是 "handler"；用户说 "authentication"，而图谱里叫 "Guardian"），字面匹配器会返回 0 个命中，答案退化成噪音。
解决办法是不凭空发明 token，而是先从图谱真实词汇中做扩展。整个过程三步：
2.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')
"
注意这段抽取同时处理了两种 token 形态：正则 [^\W\d_]+ 先按非字母数字切出整词（如 handleCallbacks），随后按驼峰边界 [A-Z]+(?=[A-Z][a-z])|[A-Z]?[a-z]+|[A-Z]+ 再拆成词素（handle、Callbacks → callbacks），并过滤掉长度小于 3 或大于 30 的 token。词表写入 graphify-out/.vocab.txt。
2.2 从词表中为问句挑选扩展 token
打开并阅读 graphify-out/.vocab.txt，针对用户问题，从这个确切列表中选出最多 12 个语义上匹配查询意图的 token。硬性约束如下：
- 只能选词表文件中真实存在的 token，禁止发明；
- 若某个查询概念在词表里没有合理 token，就跳过它——不要从训练记忆中拿近义词顶替；
- 若问句一个词表 token 都匹配不上，直接输出空列表并告知用户该语料对这个问题没有相关词汇，不要伪造一次搜索；
- 跨语言翻译：俄语 "аутентификация" → 仅在词表存在时才找
auth、credential、token、security； - 词形变化："handlers" 仅当词表存在
handler时才映射过去；"todos" 同理映射到todo。
2.3 显式打印扩展结果（可审计）
在真正执行遍历之前，必须把扩展结果展示给用户，使整个扩展过程可被审计：
Query expanded to (from graph vocab, N tokens): [token1, token2, ...]
如果列表为空，应如实说明并停止——不得进入遍历步骤。
这套"受限扩展"的价值在查询场景中尤为明显：它把 Agent 的语义理解能力与图谱的字面匹配边界做了明确分工——Agent 负责在真实词汇表内"选词对齐意图"，机器匹配器负责快速执行；两边都不会越过边界编造 token。
三、Step 1：遍历执行（CLI 优先，NetworkX 内联兜底）
用上一步选出的 token 以空格连接成扩展后的查询字符串，把它作为下面的 QUESTION 使用——不是用户的原始问句。（原始问句只保留到最后的 save-result 环节，用于回写历史。）
3.1 优先使用 CLI
graphify query "QUESTION"
# or: graphify query "QUESTION" --dfs --budget 3000
从源码看，query 子命令的实际用法为 graphify query "<question>" [--dfs] [--context C] [--budget N] [--graph path]（graphify/cli.py）。其中：
--dfs：切换到深度优先遍历；--budget N：输出 token 预算，CLI 默认传 2000（与内联脚本一致）；--context C/--graph path：分别指定上下文过滤器与图谱文件路径，不传时图谱默认落在graphify-out/graph.json；- CLI 调用
_query_graph_text时使用的遍历深度为 2（depth=2，见 graphify/cli.py）。
值得注意的工程细节：CLI 在完成查询后会调用 querylog.log_query(...) 记录一次查询日志，并通过 _touch_query_stamp 更新时间戳（graphify/cli.py），这为后续增量更新与反射判断"何时需要重跑 reflect"提供了输入信号。
3.2 CLI 不可用时的内联 NetworkX 兜底
如果 CLI 不可用，就直接加载 graphify-out/graph.json，用内联脚本执行完全相同的逻辑：
$(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 名作为起点；无命中则直接
sys.exit(0)，绝不硬遍历。 - DFS 与 BFS 的语义差异：DFS 沿单一路径深入、深度限制为 6 层，适合回答"X 如何到达 Y"；BFS 逐层扩散、深度限制为 3 层，适合回答"X 连着什么"。
- Token 预算输出：输出按
~4 chars/token折算成字符预算，超限时截断并提示use --budget N for more——把"输出规模受上下文窗口约束"显式化。 - 答案要素：节点输出带
source_file与source_location，边输出带relation与confidence标签——这就是"每条边都被解释过"的工程落地，也是最终答案可以精确引用源码位置的根基。
将脚本中的 QUESTION 替换为扩展后的查询串，MODE 替换为 bfs 或 dfs，BUDGET 替换为 token 预算（默认 2000，或与 --budget N 指定的值保持一致）。然后基于上面输出的子图回答问题，且只能依据图谱实际包含的内容：
- 先读子图：节点标签、边的 relation、confidence 标签、源码位置；
- 引用具体事实时，标注
source_location； - 若图谱信息不足，直接说明——严禁幻觉出不存在的边。
内联脚本与 CLI 在起点发现逻辑上同源：服务端 _query_graph_text 通过 _score_query 做单次全图打分得到合并排序与逐 token 种子（graphify/serve.py），并额外处理了关系意图词（calls、uses 等，见源码中 _RELATIONAL_INTENT_TERMS 相关的 #2507 逻辑），避免动词类 token 抢占起点席位——这正是图谱查询比纯全文检索更"懂结构"的地方。
3.3 回答问题后：用 save-result 回写，闭合反馈环
写出答案后，把它回写进图谱，让它改进后续查询。务必在 --answer 文本中包含扩展 token 轨迹（例如 "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 用完整答案文本（其中包含扩展 token 轨迹）；NODE1 NODE2 用你引用的节点标签列表。下一次 --update 会把这段 Q&A 抽取成图谱中的一个节点，从而形成闭环。
从源码看，save-result 子命令的参数与默认值（graphify/cli.py）比上面命令更完整，支持 --answer-file（从文件读答案）与 --memory-dir（默认 graphify-out/memory），内部调用 graphify.ingest.save_query_result(...)（graphify/ingest.py）。两个参数二选一，--answer 与 --answer-file 至少要给一个，否则报错。
工作记忆（自改进回路）：追加 --outcome 让后续会话从本次经验中学习——在 save-result 命令上追加 --outcome useful|dead_end|corrected（纠正场景还可追加 --correction "the right answer"）：
useful——被引用的节点很好地回答了问题（它们将成为 preferred sources，优先引用）；dead_end——该问题/路径走进了死胡同，下次不要再重新推导；corrected——之前保存的答案是错的，--correction记录正确内容。
3.4 会话开始时：用 reflect 刷新经验教训
在开始图谱相关工作前，刷新并阅读历史经验：
graphify reflect --if-stale
然后阅读 graphify-out/reflections/LESSONS.md，其中列出了 preferred sources（从这些节点开始找）、known dead ends（直接跳过）以及之前的 corrections（纠正记录）。reflect 是廉价、确定性、不需要 LLM 的命令；--if-stale 让它在 LESSONS.md 比所有输入都新时（例如 git hook 刚刷新过）变成空操作。即便没有安装 git hook，由你手动运行 reflect 也能让课程保持最新；如果 post-commit hook 已安装，那么会话开始时的这次运行借助 --if-stale 几乎不消耗任何成本。
源码中的 reflect 子命令还暴露了更多可调参数（graphify/cli.py），供需要精细调优时参考：
| 参数 | 默认值 | 作用 |
|---|---|---|
--memory-dir | graphify-out/memory | 读取历史问答（save-result 写入目录）的位置 |
--out | graphify-out/reflections/LESSONS.md | 生成的 LESSONS 输出路径 |
--graph | graphify-out/graph.json（存在时） | 图谱文件 |
--half-life-days | 30.0 | 信号权重每 N 天衰减一半，用于时间衰减偏好 |
--min-corroboration | 2 | 至少需要多少个不同的 useful 结果才能把一个节点提升为 preferred |
--if-stale | 关闭 | 当 LESSONS.md 已比所有输入新时跳过重算 |
四、/graphify path：两概念间的最短路径
path 的目标是找出图谱中两个具名概念之间的最短路径。同样优先使用 CLI：
graphify path "NODE_A" "NODE_B"
如果 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 为用户询问的实际概念名。然后用平实语言解释这条路径——每一跳的含义是什么、为什么这条链路有意义。路径输出的价值正在于 hop-by-hop 地呈现"依赖如何传导"：A --calls--> B --imports--> C 的每一跳都带 relation 与 confidence，解释时可以据此说明哪个环节是强证据（高置信度、有源码位置）、哪个环节偏推断。
find_node 使用的评分方式与 query 部分一致：按问句词元在标签中的出现次数给所有节点排序，只接受至少命中一次的结果。此处用 nx.shortest_path 求最短路径，nx.MultiGraph 分支说明图中同一对节点之间可能存在多条平行边（对应多关系），取边时通过 next(iter(_raw.values()), {}) 兼容单图与多重图两种表示。
写出解释后，同样回写保存：
$(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：单节点的平实解释
explain 的目标是围绕单个节点给出大白话解释——它连接了哪些东西。优先使用 CLI：
graphify explain "NODE_NAME"
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 替换为用户询问的概念。输出会给出该节点的来源文件、节点类型（file_type）、度数（degree，即连接总数）以及所有相邻节点与边。随后用 3～5 句话写一段平实解释：这个节点是什么、它连接了什么、这些连接为什么重要，并用 source_location 作为引用。
最后回写保存：
$(cat graphify-out/.graphify_python) -m graphify save-result --question "Explain NODE_NAME" --answer "ANSWER" --type explain --nodes NODE_NAME
path_query、explain、query 三种 --type 恰好对应三类查询产物；save-result 会统一落到 graphify-out/memory 目录，构成 reflect 计算偏好的输入语料。
六、三种查询的共性与最佳实践
6.1 共同的工程约定
把三份内联脚本放在一起看，能提炼出 graphify 查询侧一致的骨架：
- 统一图加载：都从
graphify-out/graph.json用json_graph.node_link_graph(data, edges='links')加载成 NetworkX 图； - 一致的字面匹配口径：query/path/explain 都以"问句词元是否出现在标签内"打分选节点，没有模糊语义层——这正是 Step 0 词汇扩展必须存在的根本原因；
- 答案即引用：节点、边、跳数、置信度都源自图谱结构与
source_location，回答必须围绕这些可验证要素组织； - 查询必回写：三类查询回答完成后都应执行对应
--type的save-result，配合会话开始时的reflect --if-stale读取 LESSONS.md，形成持续的自我改进回路。
6.2 与增量更新/重新聚类的关系
save-result 回写的 Q&A 节点之所以能进入主图谱，依赖的是下一次增量更新把 memory 内容抽取为节点——该流程在 graphify/skills/trae/references/update.md 中有完整描述（--update 仅重抽取变更文件、--cluster-only 对已有图谱重跑社区聚类）。因此一套完整的工作节奏是：reflect（读经验）→ query/path/explain（查图谱）→ save-result（写经验）→ 下次 --update（沉淀为节点）→ 再 reflect（刷新 LESSONS）。
另外，这三类查询都要求目标图谱真实存在。仓库 worked/ 目录下保留了几份真实产出的样例图谱，例如 worked/httpx/graph.json、worked/rsl-siege-manager/graph.json，可以用它们快速验证 query/path/explain 的输出形态，再回到自己的代码库上正式使用。
6.3 三条底线
- 不发明 token：扩展 token 只允许来自
graphify-out/.vocab.txt，没有匹配就直接说明，空扩展绝不进入遍历； - 不幻觉边：回答内容只能来自子图输出——节点标签、relation、confidence、source_location，图谱信息不足就明说；
- 不留无主答案：任何一次查询/路径/解释都要以对应
--type的save-result收尾，把结果与原始问题一并沉淀，供后续会话与 reflect 复用。
这套"字面匹配机器 + 词汇受限扩展 + 结构化答案回写"的组合，正是 graphify 在不依赖向量库的前提下，让 Agent 查询代码知识图谱时既保持结果可验证、又能随使用不断自我进化的核心设计。
北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台
更多推荐
9
0- 0
已为社区贡献6条内容
所有评论(0)