graphify 查询参考实战：query、path、explain 三种图谱遍历与“工作记忆”反馈闭环

本文为 graphify 技能体系的查询参考文档（Copilot 版 query.md）的完整技术解读，覆盖三大图谱查询流程：对已有知识图谱提问的 query 遍历、两概念间最短路径的 path 查询、单节点邻域解释的 explain，以及贯穿其中的“约束式查询扩展 + 答案回写 + 经验教训沉淀”反馈闭环。读完本文，你将掌握如何在已构建的 graphify-out/graph.json 之上执行确定性遍历、选择 BFS/DFS 模式、用图谱自身词表安全扩展查询词，并通过 save-result 与 reflect 让 Agent 会话越用越聪明。

参考文档的触发时机与整体设计

该参考文档的加载条件非常明确：当用户针对已有图谱提问，或执行 /graphify path、/graphify explain 命令时加载。技能核心的 query stub 遇到完整遍历需求时会指向本文档。所有流程遵循同一优先级原则：

• 优先使用 graphify query CLI（如果已安装）；
• CLI 不可用时，回退为内联 NetworkX 遍历——直接加载 graphify-out/graph.json，用文档中自带的 Python 脚本完成同样的工作。

两种遍历模式按问题类型选择：

模式	标志	适用场景
BFS（默认）	无	“X 和什么相连？”——宽泛上下文，最近邻优先
DFS	--dfs	“X 如何到达 Y？”——追踪特定链条或依赖路径

前置检查：图谱必须存在

一切查询以图谱文件存在为前提，用输出目录里记录的 Python 解释器先做检查：

$(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> 构建图谱。这里 $(cat graphify-out/.graphify_python) 的写法是 graphify 的惯例：构建时把所用 Python 解释器路径写入该文件，保证内联脚本与构建环境一致。

Step 0：约束式查询扩展（遍历前必做）

这一节是查询流程中最容易被忽视、却直接决定成败的环节。文档给出的动机是：graphify 的 query CLI 通过大小写折叠的子串匹配 + IDF 权重来命中节点——二进制内部没有词干提取、没有同义词、没有跨语言匹配，内联回退脚本也是同样的匹配方式。于是出现了典型的失配场景：

• 用户说 “обработчик”（俄语“处理器”），图谱标签写的是 handler；
• 用户说 “authentication”，图谱写的是 Guardian。

字面匹配器会返回 0 命中，答案随即退化为噪声。解法是在不发明任何词元的前提下，先针对图谱的真实词表做查询扩展：

第 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]+）：它把 FooBarService 拆成 foo、bar、service 三个词元，长度窗口保留 3~30 字符的词，避免超短噪声与超长标签碎片进入词表。词表落盘到 graphify-out/.vocab.txt。

第 2 步：从词表中挑选扩展词元（硬性约束）

读完 graphify-out/.vocab.txt 后，针对用户问题最多选择 12 个语义匹配的词元。文档对此给出了四条不可违反的约束：

• 只能挑选词表文件中真实存在的词元，禁止发明词元；
• 若某个查询概念在词表中找不到合理对应，直接跳过该概念——不要用训练记忆里的近义替换词顶替；
• 若没有任何词表词元与查询匹配，输出空列表并如实告知用户“语料中没有与这个问题相关的词汇”，不得伪造搜索；
• 跨语言翻译：俄语 “аутентификация” → 仅当词表中存在时，才选用 auth、credential、token、security；
• 形态学归一：“handlers” 映射到 handler、"todos" 映射到 todo，前提同样是词表中存在。

第 3 步：显式输出选择结果，保证可审计

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

遍历前必须把扩展结果打印给用户；若列表为空，直说并停止——不要进入遍历。这一步让“LLM 到底用了哪些词去查图”完全可审计，是防止幻觉搜索的关键设计。

Step 1：遍历——CLI 优先，内联回退兜底

将选中的词元用空格连接构成扩展查询串，后续所有遍历使用这个串作为 QUESTION（原始问题只在结尾 save-result 时保留）。

CLI 路径

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

CLI 的实现在 cli.py 的 query 分支，可以确认文档中提及的每个参数都有对应实现：

• --dfs 切换遍历模式，缺省为 BFS；
• --budget N 指定 token 预算，默认 2000，非法整数会直接报错退出；
• --context 支持上下文过滤；--graph 可指向非默认位置的图谱；
• 每次查询还会调用 querylog 记录 kind、问题、语料、耗时，并刷新查询时间戳。

一个值得注意的实现细节：源码注释明确指出 query 刻意保持无向图（不同于强制有向的 path/explain），因为 BFS/DFS 必须同时探索种子节点的调用方与调用方两侧；边方向则通过 _src/_tgt 标记在渲染层恢复，保证遍历不被方向收窄。

CLI 内部的核心遍历入口是 _query_graph_text，其签名参数与文档语义一一对应：

def _query_graph_text(
    G: nx.Graph,
    question: str,
    *,
    mode: str = "bfs",
    depth: int = 3,
    token_budget: int = 2000,
    context_filters: list[str] | None = None,
    graph_path: str | None = None,
) -> str:

匹配管线比文档描述的“子串 + IDF”更细，从源码看可确认三层机制（serve.py）：

1. 查询词切分（_query_terms）：剔除英德法西葡意多语停用词表，支持中文分词；若整句全是停用词则回退为未过滤词元，保证 “how does it work” 这类问题仍能落到某些词上；
2. 加权评分（_score_query，serve.py）：精确命中标签加分 1000、前缀命中加 100、子串命中加 1.0、源文件命中加 0.5，每档权重再乘以该词元的 IDF 值——error、exception 这类高频词的权重被压低，FooBarService 这类稀有标识符权重被抬高；大图上还有三字符（trigram）候选预筛，只在可能命中的节点集合上评分；
3. 种子选择（_pick_seeds）：按分数差阈值（gap ratio 0.2）截断，防止 error、exception 这类高频噪声词抢占种子槽位；同时保证“每个有命中的查询词元至少占一个种子席位”，避免某一词元偶然精确撞名就把其他相关词元的子串命中全部挤出（源码中引用了 issue #1445 作为该设计的由来）。

输出侧，_query_graph_text 会先打印一行头部（遍历模式、深度、种子节点、命中节点数，以及当提供 graph_path 时的图谱来源与节点总数——防止在错误目录查询时“格式正确地答错图”，见源码引用的 issue #2789），随后按相关性排序、按 token_budget × 4 字符预算截断——这正是文档中“~4 chars/token”估算的出处。

内联回退路径

CLI 不可用时，加载 graphify-out/graph.json 并在本地执行与 CLI 等价的遍历。完整脚本（来自参考文档）：

$(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' 或 'dfs'
terms = [t.lower() for t in question.split() if len(t) >= 3]  # 与词表阈值一致；保留 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)
"

使用时替换三处占位符：QUESTION（扩展后的查询串，不是原始问题）、MODE（bfs 或 dfs）、BUDGET（token 预算，默认 2000，或由 --budget N 指定）。脚本的遍历语义与文档规定完全一致：

• 从标签与词元匹配度最高的 1~3 个节点出发；
• BFS 逐层 3 层、DFS 深度上限 6；
• 读子图时同时取节点标签、边关系（relation）、置信度标签（confidence）与源码位置；
• 只用图里有的东西作答，引用具体事实时给出 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 会把这个 Q&A 抽成图中的节点。

CLI 侧 save-result 的完整参数 与文档一致：--question（必填）、--answer 或 --answer-file、--type（默认 query）、--nodes（任意多个节点标签）、--outcome（取值 useful | dead_end | corrected）、--correction、--memory-dir（默认 graphify-out/memory）。

工作记忆：让每次会话都从上次会话中学习

参考文档把 --outcome 称为“self-improving loop（自我改进回路）”。在 save-result 命令上追加 --outcome，未来会话就能从本次会话中学习：

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

会话开始时要刷新并阅读教训：执行 graphify reflect --if-stale（廉价、确定性、无 LLM 参与；--if-stale 在 LESSONS.md 已比所有输入都新时是空操作——例如 git hook 刚刚刷新过），然后阅读 graphify-out/reflections/LESSONS.md。其中列出优先来源（从这里开始）、已知死路（跳过它们）和历史更正。

从源码看，这套机制的实现在 reflect.py：

• reflect 读取 save-result 写入 memory/ 的 Q&A 文档，聚合 useful / dead_end / corrected 三类结果信号，产出单一的教训工件 graphify-out/reflections/LESSONS.md；
• 来源节点是打分而非计数：每次引用贡献一个带符号、时间衰减的分数（useful 为正，dead_end/corrected 为负），默认半衰期 30 天（一个新鲜的死路可以压过几个月前的一次 useful）、最小佐证数 2（一次 save 不能凭空把一个节点变成可信教训）；
• 完全确定性：无 LLM、稳定排序、给定输入与 now 时字节级稳定输出；有图谱时按社区标签分组，无图谱则退化为单一平铺章节；
• 教训文件放在 reflections/ 而非 wiki 目录，是因为 graphify export wiki 每次运行会删除 wiki/*.md 下所有文件，写在那里的教训会被下一次导出抹掉（模块 docstring 原话）。

--if-stale 的判定逻辑在 lessons_fresh：比较 LESSONS.md 的 mtime 与所有输入（memory 目录下的 *.md、graph.json 及 sidecar）的最新 mtime，输出缺失或不可读一律视为“不新鲜，必须重建”。这正是文档所说“装了 post-commit hook 时，会话开始的 reflect 开销几乎为零”的底层依据。

另外，graphify explain 的 CLI 输出还会叠加一层“经验提示”：explain 命令实现 会从 graph.json 旁的 .graphify_learning.json sidecar 读取节点状态，打印形如 Lesson: preferred source (start here) — 3 useful, score=1.4142 或 Lesson: contested (useful 1 / dead-end 2) 的行，代码变化后还会追加 [code changed since — re-verify]。也就是说，工作记忆不仅影响 Agent 的宏观决策，也会直接体现在单节点解释里。

/graphify 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 替换为用户给出的实际概念名，然后用通俗语言解释这条路径：每一跳意味着什么、为什么重要。

值得对照的是 CLI 实际比内联脚本更严谨（cli.py path 分支）：

• 默认有向（--undirected 可切换）：方向真值存在每个 graph.json 的边标记里，默认尊重它；无有向路径时会提示 “Re-run with --undirected to search ignoring edge direction.”；
• 歧义防护：两端解析到同一节点时直接报错（几乎不可能是调用者想要的，源码引用 issue #828）；次选分数与前选差距不足 10% 时打印 ambiguous 警告；
• 确定性选路：通过构建排序过的物化图，保证同一 graph.json 上邻居顺序与选出的路径是规范化的，不再因哈希种子不同而每次跑出不同等长路径（issue #2074）；
• 如实报告关系：每一跳打印的是该节点对实际存储的关系（同一对节点可携带多条平行关系，全部展示），绝不虚构 calls。

写完解释后同样回写：

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

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

# 寻找最佳匹配节点
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

CLI 版的 explain（cli.py explain 分支）在节点解析上走的是分层匹配而非简单词元计数：优先级为源文件精确匹配 → 标签/ID 精确匹配 → 前缀匹配 → 子串匹配（_find_node_tiers，serve.py）。当同一获胜层级横跨多个源文件时（比如两个工作区各自定义了 MetricsPort），命令会列出全部竞争者并要求“用仓库相对路径或完整 node id 重试”，而不是任取其一自信作答。连接列表超过 20 条时不会只给一个裸计数，而是按“方向 + 源文件”分组统计，让高连接度节点的“谁在调用它、影响面多大”仍然可见（源码引用 issue #2009）。

小结：三个命令、一条反馈回路

把参考文档的三条流程合起来看，graphify 的查询体系是一套完整的闭环：

1. 查询：先用图谱自身词表做约束式扩展（最多 12 个词元、零发明），再走 BFS/DFS 遍历，只用图中事实作答并引用 source_location；
2. 回写：每个答案经 save-result 存回 graphify-out/memory/，扩展轨迹、引用节点、结果类型（useful/dead_end/corrected）都成为下次推导的素材；
3. 沉淀：graphify reflect --if-stale 以时间衰减打分聚合这些素材，产出 LESSONS.md（优先来源、已知死路、历史更正），并在 explain 的输出中以 Lesson 行直接提示——下一次会话从这里开始。

三条流程的 CLI 入口在 cli.py（query）、cli.py（path）、cli.py（explain），内联回退脚本则全部内嵌在参考文档 graphify/skills/copilot/references/query.md 中，可在无 CLI 环境下原样复制运行；对应的行为测试可参考 tests/test_query_cli.py 与 tests/test_explain_cli.py。