graphify 知识图谱查询实战：query / path / explain 三种遍历流程、Token 预算与 Work-Memory 反馈闭环

graphify 把任意代码库（含文档、SQL 配置、PDF）构建成一个本地可查询的知识图谱后，"对图谱提问"是最高频的使用场景。本文以 Codex 平台的技能参考文档 graphify/skills/codex/references/query.md 为主体，完整讲清三条查询流程——/graphify query（BFS/DFS 子图遍历）、/graphify path（两节点最短路径）、/graphify explain（单节点解释）——的完整操作步骤、约束式查询扩词规则、内联 NetworkX 回退脚本，以及 save-result + reflect 构成的自改进工作记忆闭环。读完后你既能照着文档逐步复现全部命令，也能理解这些命令在 CLI 源码 中的实际行为与取舍。

一、文档定位与适用前提

该参考文档是 /graphify 技能体系在 Codex 平台下的"查询参考"：当用户在已有图谱上提问，或执行 /graphify path、/graphify explain 时加载。技能主文件中的 query 存根（stub）指向它来获取完整的遍历流程（同一份参考在仓库中为各平台复制了一份，如 graphify/skills/claude/references/query.md、graphify/skills/codex/references/query.md 等，内容保持一致）。

流程的统一约定是：优先使用 graphify CLI（若已安装），不可用时回退为内联 NetworkX 遍历。

开始前必须先确认图谱存在：

$(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 解释器标记文件，所有后续内联脚本都用 $(cat graphify-out/.graphify_python) 取到与构建一致的解释器，避免环境问题。

二、两种遍历模式：BFS 与 DFS 的选择

原文档给出了一张选型表，这是整篇参考文档的"决策起点"：

模式	标志	适用问题
BFS（默认）	（无标志）	"X 与什么相连？"——要广域上下文，最近邻优先
DFS	--dfs	"X 如何到达 Y？"——追踪具体链条或依赖路径

在 CLI 中对应：

graphify query "QUESTION"
# 或：graphify query "QUESTION" --dfs --budget 3000

对照 CLI query 命令实现 可以看到实际支持的参数比文档示例更多：完整用法为 graphify query "<question>" [--dfs] [--context C] [--budget N] [--graph path]，其中 --budget 默认 2000、--context 用于附加上下文过滤、--graph 指定非默认的 graph.json 路径。

三、Step 0 — 约束式查询扩词（遍历前必做）

这是本文档最有工程价值的部分。文档明确指出：graphify 的 query CLI 通过小写折叠的子串匹配 + IDF 来命中节点（在源码中对应 graphify/serve.py 的 _query_terms、_compute_idf、_score_nodes 等函数，由 CLI 入口 导入调用），二进制内部没有词干还原、没有同义词、没有跨语言匹配；内联回退脚本也用同样的匹配方式。

后果是：如果用户提问的词汇与图谱标签不一致（用户说"обработчик" / 图谱叫 handler；用户说 "authentication" / 图谱叫 Guardian），字面匹配器会返回 0 命中，回答退化成噪声。

文档给出的修复方式是不凭空造词，而是先用真实图谱词汇表扩词：

第 1 步：从节点标签抽取 token 词汇表

$(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])|...）保证 AuthHandler 会拆出 auth、handler；长度过滤 3 <= len(t) <= 30 既去掉噪声又保留 api、jwt、ios 这类短但有效的 token。

第 2 步：从词汇表里选词，硬约束四条

• 只能选 graphify-out/.vocab.txt 中实际存在的 token，最多 12 个，绝不凭空造词；
• 若某个查询概念在词汇表中找不到合理对应，直接跳过，不要用训练记忆里的近义同义词顶替；
• 若完全没有词汇表 token 匹配该问题，输出空列表并明确告诉用户"语料库中没有相关词汇"，不要编造一次搜索；
• 跨语言转换只在目标 token 确实存在时进行：俄语 "аутентификация" → 仅在词汇表存在时选 auth、credential、token、security；形态转换同理，"handlers" → handler 仅在 handler 存在时成立。

第 3 步：把选择显式打印给用户，让扩词过程可审计

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

列表为空就直说并停止，不要进入遍历。

四、Step 1 — 遍历执行

把选出的 token 用空格连接，构成扩展查询串，作为下面的 QUESTION（原始问题只在最后 save-result 时保留）。

CLI 可用时直接：

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

CLI 不可用时，加载 graphify-out/graph.json 内联遍历。流程五步：找 1~3 个与扩展 token 最匹配的起始节点 → 从每个起始节点执行对应遍历 → 读子图（节点标签、边关系、置信度标签、源码位置）→ 只用图里有的内容回答，引用具体事实时引用 source_location → 图里信息不足就明说，不要幻觉边。

完整内联脚本（替换 QUESTION、MODE（bfs/dfs）、BUDGET（默认 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]  # 与 vocab 阈值一致；保留 api/jwt/ios（#1392）

# 找最佳匹配的起始节点
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：尽可能沿一条路径深走再回溯，深度限 6 避免遍历全图
    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：逐层扩展所有邻居，深度 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 预算感知输出：按相关性排序，预算处截断（约 4 字符/token）
token_budget = BUDGET  # 默认 2000
char_budget = token_budget * 4

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)
"

几个值得注意的实现细节：

• 起始节点选取：对每个节点统计扩展 token 命中标签的子串次数，取分数最高的前 3 个（scored[:3]）；
• BFS 深度 3、DFS 深度限 6：两者都是刻意限制子图规模，防止大图谱下输出爆炸；
• Token 预算截断：输出按相关性排序后按 token_budget * 4 字符截断，超预算时追加提示（用 --budget N 扩大）；
• MultiGraph 兼容：isinstance(G, nx.MultiGraph) 分支处理平行边取第一条数据，保证脚本在旧/新两种图谱存储格式下都能跑。

CLI 路径下的对应行为（源码级佐证）

对照 graphify/cli.py 的 query 分支，内联脚本的行为与 CLI 高度一致，但有几处值得了解：

• CLI 调用 _query_graph_text（定义在 graphify/serve.py），传入 mode（bfs/dfs）、depth=2、token_budget=budget、context_filters；
• 源码注释明确说明：query 刻意保持图无向（而 path/explain 强制有向）——因为 BFS/DFS 必须同时探索种子节点的调用方与 callee 才能构成有用上下文，若强转 DiGraph，G.neighbors() 只会返回后继节点，种子无出边时调用方一侧结果会被静默丢弃；方向性则靠逐边 _src/_tgt 标记在渲染时恢复；
• 每次 query 会写 querylog 并 _touch_query_stamp（记录最近查询时间戳，供 hook 守卫判断"本会话是否查询过图谱"）。

五、把答案存回图谱：save-result 反馈闭环

回答写完后必须回写图谱，让未来查询受益。文档要求把扩展 token 痕迹写进 --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 是回答中引用的节点标签列表。

从 save-result 命令实现 可以看到完整参数面：--question（必填）、--answer 或 --answer-file（二选一必填）、--type（默认 query，path 流程用 path_query，explain 流程用 explain）、--nodes（变参列表）、--memory-dir（默认 <graphify-out>/memory）。最终落到 save_query_result 写入记忆文档。

Work Memory：三档结果标签

追加 --outcome 参数让未来会话从本次会话中学习（--correction "the right answer" 用于记录更正）：

• useful — 引用的节点很好地回答了问题（它们会成为优先来源）；
• dead_end — 该问题/路径走不通，下次不要再重复推导；
• corrected — 保存的答案是错的，--correction 记录正确答案。

会话开始时的课程刷新

开始图谱工作前，刷新并阅读课程：执行 graphify reflect --if-stale（廉价、确定性、无 LLM；--if-stale 在 LESSONS.md 已比所有输入都新时直接 no-op，例如 git hook 刚刷新过的场景），然后读 graphify-out/reflections/LESSONS.md。它列出优先来源（从这里开始）、已知死路（跳过）、历史更正。

reflect 命令实现 补充了文档未展开的参数：--half-life-days（信号权重每 N 天减半，默认 30）、--min-corroboration（提升为 preferred 所需的不同 useful 结果数，默认 2）、--out（默认 graphify-out/reflections/LESSONS.md）。底层聚合逻辑在 graphify/reflect.py，实现半衰期衰减、来源佐证与 LESSONS.md 渲染；自己跑一次 reflect 能保证课程在没装 git hook 时也是最新的，而若 post-commit hook 已安装，--if-stale 使会话启动时的这次运行几乎零成本。

六、/graphify path：两概念间的最短路径

找到图中两个命名概念之间的最短路径。CLI 优先：

graphify path "NODE_A" "NODE_B"

CLI 不可用时内联执行（替换 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}')
"

拿到路径后用自然语言解释：每一跳意味着什么、为什么重要。然后回写：

$(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

CLI 实现的三个关键行为

对照 path 命令源码：

1. 有向为默认（源码标注 #2487）：graphify path 完整用法是 graphify path "<source>" "<target>" [--graph path] [--directed|--undirected]。方向真值存在于每份 graph.json（新文件的弧线顺序、旧文件的 _src/_tgt 标记），所以默认尊重方向；无有向路径时会提示加 --undirected 重试。这与内联脚本用无向 shortest_path 的语义不同——CLI 更严格；
2. 多图层加载（#2074）：加载时强制 directed=True, multigraph=True，使同一对节点间的平行边（如一条 references 和一条 calls）不会被"最后写入者覆盖"合并掉，输出的是该点对实际存储的关系；
3. 歧义防护（#828）：两端解析到同一节点时会直接报错退出，要求更具体的标签或精确节点 ID；top 分数与次名差距小于 10% 时打印 ambiguous 警告。

七、/graphify explain：单节点全景解释

对单个节点给出通俗解释——它是什么、连向谁、为什么这些连接重要。CLI 优先：

graphify explain "NODE_NAME"

CLI 不可用时内联执行（替换 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()

# 找最佳匹配节点
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})')
"

随后写 3~5 句话的解释：节点是什么、连接了什么、为什么重要，并用 source 位置作为引用出处。最后回写：

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

CLI 版本输出得更多

从 explain 命令源码 看，CLI 版除了标签、来源、类型、度数外，还输出：

• 社区（community_name）与工作记忆叠加层：读取 reflect 生成的 .graphify_learning.json 侧车文件，显示 Lesson: preferred source (start here)、contested、或 tentative 状态及得分，代码变更过还会标注 [code changed since — re-verify]；
• 歧义防护：同名节点存在于不同文件时列出候选并要求用仓库相对路径或完整节点 ID 重试；
• 连接分类：按边的真实方向（_src 标记，#2309）区分 -->（出）与 <--（入），每条连接附带关系、置信度和边位置（调用/引用发生地，而非定义行）；高连接度节点只展开前 20 条，其余按"方向 + 文件"分组计数展示（#2009），避免裸计数把答案藏起来。

八、流程速查

步骤	命令	关键点
前置检查	读取 graphify-out/graph.json 是否存在	不存在先跑 /graphify <path>
会话开始	graphify reflect --if-stale + 读 graphify-out/reflections/LESSONS.md	优先来源、已知死路、历史更正
扩词	生成 graphify-out/.vocab.txt，选 ≤12 个真实 token	禁止造词；空列表则停止
查询	graphify query "扩展串" [--dfs] [--budget N] [--context C]	默认 BFS、budget 2000
路径	graphify path "A" "B" [--directed|--undirected]	有向为默认；歧义时报错
解释	graphify explain "NODE"	含社区、lesson 叠加、入/出边分类
回写	graphify save-result --question ... --answer ... --type ... --nodes ... [--outcome useful|dead_end|corrected]	answer 中保留扩词痕迹

九、总结

graphify/skills/codex/references/query.md 定义的不只是三条命令，而是一套可审计的图谱问答协议：查询前用真实词汇表约束扩词（防幻觉）、遍历时用预算和深度控制输出规模（防噪声）、回答只引用图中存在的 source_location（防编造）、结束后用 save-result + --outcome 把经验写回（自改进）。配合 graphify/cli.py 中的 CLI 实现与 graphify/reflect.py 的课程聚合，这套流程让同一个知识图谱随使用次数增加而越答越准——这正是 graphify "no vector store、每条边都有解释" 设计哲学在查询侧的落地。