graphify 知识图谱查询实战：query / path / explain 全流程、受约束查询扩展与自改进工作记忆

本文基于 graphify 的 Claude 技能参考文档 query.md，完整拆解对已构建知识图谱的三种查询流程——query（子图遍历）、path（两节点最短路径）与 explain（单节点解释）。读完后你能掌握：BFS/DFS 两种遍历模式的选择依据、强制前置的“受约束查询扩展”如何对抗词汇失配、graphify query CLI 的底层匹配与 token 预算机制，以及用 save-result + reflect 构建“越查越聪明”的反馈闭环。

一、这份参考文档的定位

graphify/skills/claude/references/query.md 是 graphify 技能体系中专管“查询侧”的参考文档，其加载时机与职责在文档开头即有明确定义：

• 加载时机：当用户对已存在的图谱提问，或运行 /graphify path、/graphify explain 时加载；
• 职责：核心技能中的 query stub 会跳转到这里，获取完整的遍历（traversal）流程；
• 执行策略：优先使用 graphify query CLI（如已安装），CLI 不可用时退化为内联的 NetworkX 遍历脚本（inline fallback）。

需要强调的适用前提：所有查询都发生在已构建好的图谱上，输出目录约定为 graphify-out/，其中 graph.json 是图谱本体，.graphify_python 记录了构建时使用的解释器路径（后续所有内联脚本都用 $(cat graphify-out/.graphify_python) 前缀执行，保证解释器环境一致）。

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

文档给出了一张模式选择表，这是查询流程的第一个决策点：

Mode	Flag	Best for
BFS（默认）	(无)	“X 连接了什么？”——广域上下文，先取最近邻
DFS	--dfs	“X 如何到达 Y？”——追踪一条具体的链条或依赖路径

• BFS 逐层向外扩展邻居，适合回答“围绕 X 的上下文有哪些”这类发散性问题；
• DFS 沿一条路径尽可能走深、再回溯，适合回答“调用链/依赖链”这类收敛性问题。

三、前置检查：确认图谱存在

任何查询动作之前，先验证图谱文件存在，避免在空目录上静默失败：

$(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> 构建图谱。CLI 侧同样有对应防护：graphify query 在图文件不存在或非 .json 时会直接报错退出（见 graphify/cli.py）。

四、Step 0——受约束查询扩展（遍历前必做）

4.1 为什么必须先做查询扩展

文档明确指出：graphify 的 query CLI 通过大小写折叠子串 + IDF 匹配节点——二进制内部没有词干提取（stemming）、没有同义词、没有跨语言匹配，内联回退脚本也是同样的匹配方式。这一点可以在源码中得到印证：

• 查询词切分与 IDF 权重计算位于 graphify/serve.py 的 _query_terms 与 _compute_idf，后者为查询词计算 IDF 权重并缓存在图中；
• 文档中举例说明失配后果：用户说“обработчик”（俄语“处理器”）而图谱标签是 handler；用户说 “authentication” 而图谱标签是 Guardian——字面匹配器返回 0 命中，答案就会塌缩成噪声。

解法是不凭空造词，而是先把用户的查询对齐到图谱自身的真实词汇表。

4.2 从节点标签提取词汇表

第一步从所有节点标签中提取 token 词汇，写入 graphify-out/.vocab.txt：

$(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(vocab), encoding='utf-8')
print(f'vocab: {len(vocab)} tokens')
"

其中正则 [A-Z]+(?=[A-Z][a-z])|[A-Z]?[a-z]+|[A-Z]+ 负责把驼峰/大写标签（如 processOrder、IDToken）拆成独立词元；长度约束 3 <= len(t) <= 30 过滤噪声，这也与内联回退脚本中 len(t) >= 3 的匹配阈值保持一致（文档注释说明该阈值用于保留 api/jwt/ios 这类短标识符，对应上游 issue #1392）。

4.3 硬性选择约束

读取 .vocab.txt 后，针对用户问题最多选出 12 个语义匹配的 token，且必须遵守文档给出的四条硬约束：

1. 只能选词汇文件中真实存在的 token，禁止发明 token；
2. 若某查询概念在词汇表中找不到合理对应 token，直接跳过——不要用训练记忆里的近似同义词替代；
3. 若完全没有词汇 token 与查询相关，输出空列表并明确告知用户“该语料对此问题没有相关词汇”，不要伪造搜索；
4. 允许跨语言翻译与形态归约，但都以“词汇表中存在为前提（IFF present）”：
   • 跨语言：俄语 “аутентификация” → 在词汇表中存在时才找 auth、credential、token、security；
   • 形态归约：“handlers” → handler（若在词汇表中）；“todos” → todo（若在词汇表中）。

选定后必须在运行查询前把扩展结果显式打印给用户，使整个扩展过程可审计：

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

若列表为空，照实说明并停止——不进入遍历步骤。

五、Step 1——执行遍历

把选中的 token 用空格拼成扩展查询串，之后统一以该串（而非用户原话）作为 QUESTION 执行遍历。原始用户问题仅在最后 save-result 时保留。

5.1 优先走 CLI

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

结合 graphify/cli.py 的参数解析实现，query 子命令支持：

参数	说明
"QUESTION"	必填，位置参数；应传入 Step 0 的扩展查询串
--dfs	切换为 DFS 遍历（默认 BFS）
--budget N	token 预算，默认 2000；输出超长时按预算截断（支持 --budget=3000 形式）
--context C	上下文过滤（可多次出现，见 CLI 中的 context_filters）
--graph path	指定图谱文件，默认取 _default_graph_path()，必须是 .json

5.2 内联回退：CLI 不可用时的 NetworkX 遍历

文档给出了完整的内联回退脚本，核心逻辑分四段：

1. 选种子节点：遍历全部节点，对标签做大小写折叠后的 term-overlap 计分，取得分最高的 1–3 个节点作为起点；
2. 执行遍历：BFS 逐层扩展 3 层；DFS 用显式栈、深度上限 6（避免遍历整图）；
3. 读取子图：节点标签、边关系（relation）、置信度标签（confidence）、来源位置（source_location）；
4. 回答纪律：只使用图中存在的信息作答，引用具体事实时引用 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)
"

注意两个实现细节：

• 输出按相关度排序：种子节点（查询词命中的符号）排在前面，保证预算截断时最相关的内容存活；
• token 预算：内联脚本按约 4 字符/token 折算字符预算；而 CLI 路径的截断器 _cut_lines_to_budget（graphify/serve.py）采用约 3 字符/token 的更保守规则，并在输出顶部同时标注截断信息，防止“底部截断标记被误读为内容缺失”。

5.3 CLI 内部：匹配与遍历是如何实现的

从 graphify/cli.py 的 cmd == "query" 分支可以看到完整调用链：

1. 加载 graph.json；若图里只有 edges 键则自动映射为 links；
2. 刻意保持图无向：源码注释（cli.py 中 L1125-L1132）解释了原因——query 需要同时探索种子节点的调用方与被调用方；若强转有向图，G.neighbors() 只返回后继，会让没有出边的种子静默丢失所有“调用方侧”结果。方向性改为逐边保留（_src/_tgt 标记），遍历不收窄、渲染仍正确；
3. 调用 graphify/serve.py 的 _query_graph_text(G, question, mode=..., depth=2, token_budget=budget, ...)：内部完成查询词切分（_query_terms）、IDF 加权的单次评分（_score_query，缓存于 _compute_idf）、种子挑选（_pick_seeds）与 _bfs/_dfs 遍历；
4. 每次查询经 graphify/querylog.py 记录（问题、模式、预算、耗时），并更新查询时间戳；
5. 输出头还会标注所用图谱的路径与节点数——源码注释（serve.py L1238-L1245）说明这是为了防“在父项目目录下误查 vendored 子项目图谱”的静默错误（issue #2789）。

此外 CLI 对旧版节点 ID 方案（pre-#1504）会打印提示，建议用 graphify extract --force 重建以获得路径限定 ID，修复同名文件冲突。

六、把答案写回图谱：save-result 与反馈闭环

回答完成后，文档要求把问答存回，让下一次 --update 能把该 Q&A 提取为图谱节点，形成反馈闭环：

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

参数替换规则：

• ORIGINAL_QUESTION：用户原话（不是扩展串）；
• ANSWER：完整答案文本，且必须内嵌扩展 token 轨迹（例如 "Expanded from original query via vocab: [tokens]. Then traversed..."），以便下一次 --update 把扩展历史提取成图谱节点；
• NODE1 NODE2：答案中实际引用过的节点标签列表。

graphify/cli.py 中的 save-result 实现还暴露了文档流程之外的两个可选参数：--answer-file（从文件读答案，适合长答案）与 --memory-dir（默认 graphify-out/memory）。最终由 graphify/ingest.py 的 save_query_result 落盘。

工作记忆（work memory，自改进循环）：在 save-result 命令上追加 --outcome，让后续会话从本次查询中学习：

outcome	含义	效果
useful	引用的节点很好地回答了问题	这些节点成为 preferred sources（优先来源）
dead_end	该问题/路径走不通	下次不再重新推导
corrected	保存的答案有误	配合 --correction "the right answer" 记录正确答案

CLI 侧这三个取值是硬校验的（--outcome 使用 choices=("useful", "dead_end", "corrected")，见 graphify/cli.py），--correction 为自由文本。

会话开始时刷新教训：reflect

文档要求：在图谱工作开始时运行 graphify reflect --if-stale（廉价、确定性、无 LLM），然后读取 graphify-out/reflections/LESSONS.md。该文件列出三类信息：

• preferred sources：优先从这里开始查；
• known dead ends：跳过；
• prior corrections：历史更正。

从 graphify/cli.py 的 reflect 子命令可以看到支撑这一机制的参数：--half-life-days（信号权重半衰期，默认 30 天）、--min-corroboration（把一个节点提升为 preferred 所需的不同 useful 结果数，默认 2）、--if-stale（当 LESSONS.md 已比所有输入都新时跳过——例如 git hook 刚刷新过时，会话启动的刷新几乎零成本）。自己运行 reflect 即使没装 git hook 也能保持教训新鲜；实现位于 graphify/reflect.py。

七、/graphify path：两个概念之间的最短路径

优先 CLI：

graphify path "NODE_A" "NODE_B"

内联回退（替换 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}')
"

要点：find_node 用多词标签重叠计分选出两端节点；nx.shortest_path 输出每一跳的 relation 与 confidence；无路径或节点缺失都有明确的分支输出。得到路径后，文档要求用自然语言解释每一跳的含义与重要性，然后以 --type path_query 写回：

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

内联回退（替换 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})')
"

该脚本输出节点的身份信息（label、source_file、file_type、度数）与全部一跳连接（关系、置信度、邻居来源文件）。文档要求基于输出写一段 3–5 句的解释：这个节点是什么、连接了谁、这些连接为何重要，并以来源位置作为引用。最后以 --type explain 写回：

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

九、三条流程的写回类型与文件地图

流程	推荐命令	save-result 的 --type
query（子图遍历）	graphify query "QUESTION" [--dfs] [--budget N]	query
path（最短路径）	graphify path "NODE_A" "NODE_B"	path_query
explain（单节点解释）	graphify explain "NODE_NAME"	explain

围绕本文主题，可在当前仓库中继续深入的关键文件：

文件	作用
graphify/skills/claude/references/query.md	本文依据的技能参考文档（query/path/explain 完整流程）
graphify/cli.py	query 子命令：参数解析、无向图加载、querylog 记录
graphify/cli.py	save-result 与 reflect 子命令的参数定义
graphify/serve.py	_query_graph_text：IDF 评分、种子选择、BFS/DFS、预算截断与输出头
graphify/ingest.py	save_query_result：问答落盘的底层实现
graphify/reflect.py	从记忆目录生成 LESSONS.md 的确定性反思逻辑
graphify/querylog.py	查询日志，支撑 reflect 的信号来源

最后强调文档贯穿三条流程的回答纪律：只用图中存在的信息作答；引用具体事实时给出 source_location；信息不足时直说而不臆造边。再叠加 save-result（含 --outcome 工作记忆）与 reflect 的教训刷新，graphify 的查询侧才构成一个“查询 → 回答 → 写回 → 反思 → 下次查询更准”的自改进闭环。