graphify query 深度解析：OpenCode Skill 的知识图谱查询工作流（query、path、explain 与受约束查询扩展）

本文基于 graphify 仓库中 OpenCode 平台 Skill 的参考文档 query.md，完整讲解在一张已构建的知识图谱上执行查询的标准工作流：query / path / explain 三个入口、遍历前的"受约束查询扩展"机制、CLI 优先 + 内联 NetworkX 兜底的双执行路径，以及 save-result / reflect 构成的反馈闭环。读完本文，你将掌握从"用户提问 → 词表扩展 → 图遍历 → 基于证据作答 → 记忆回写"的完整链路，并能结合 cli.py 与 serve.py 的源码理解每个参数的实际行为。

1. 何时加载这份参考文档

这份参考文档的触发条件在文档开头即已写明：当用户针对一张已存在的图谱提问，或运行 /graphify path、/graphify explain 时加载；核心 Skill 的 query 桩（stub）会把完整遍历流程指向这里。

执行策略上，文档给出了一条清晰的优先级：graphify query CLI 可用时优先用 CLI；不可用时回退到内联 NetworkX 遍历。两个入口消费同一份数据——graphify-out/graph.json。

从源码结构看，这个"查询优先"工作流并非仅靠文档约定：graphify/cli.py 中定义了注入给 Agent 的 PreToolUse 钩子提示（_SEARCH_NUDGE / _READ_NUDGE），要求"必须先运行 graphify query 再 grep/读取原始文件"，严格模式下甚至会 deny 首次直接读文件。本参考文档正是这一工作流在 OpenCode 平台上的具体落地。

同一份参考文档在仓库其他平台（如 claude 版）中也存在等价副本，流程完全一致，仅安装/钩子配置有平台差异。

2. 前置检查：图谱必须已存在

任何查询操作的第一步是验证图谱文件存在：

$(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) 读取的是输出目录中记录的 Python 解释器路径——整套内联脚本统一用它执行，保证运行查询的环境（含 networkx 依赖）与构建图谱时的环境一致。
• graphify-out/ 是 graphify 的默认输出目录（由 graphify/paths.py 中的 GRAPHIFY_OUT 常量定义，CLI 的 --graph 参数默认值即 <graphify-out>/graph.json，见 cli.py）。

3. 两种遍历模式的选择

文档给出了一张选择表，按问题类型决定遍历方式：

模式	标志	适用场景
BFS（默认）	(无)	"X 连接到了什么？"——需要宽泛上下文，先取最近邻
DFS	--dfs	"X 如何到达 Y？"——追踪一条具体的链路或依赖路径

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

这是整份文档最核心、也最容易被忽视的一步。

4.1 为什么需要扩展：匹配器的硬性局限

文档原文给出了理由：graphify 的 query CLI 通过小写折叠的子串匹配 + IDF 加权来匹配节点——二进制内部没有词干还原（stemming）、没有同义词、没有跨语言匹配，内联兜底脚本也是同样的匹配方式。

从源码看这一点可以印证：query 命令最终调用 _query_graph_text，其中对问题词做 _query_terms 分词后由 _score_query 完成打分，全部是字面匹配。因此，如果用户的问题与图谱标签使用了不同语言或不同领域词汇——文档举了两个例子：用户说俄语"обработчик"而图中标签是 handler；用户说 authentication 而图中是 Guardian——字面匹配器会返回 0 命中，答案就会退化成噪声。

解决方式是：不凭空发明词元（token），先针对图谱的实际词汇表扩展查询。

4.2 从节点标签抽取图谱词表

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

这段脚本的逻辑：对每个节点标签先按"非数字字母串"切词，再对驼峰/缩写作二次拆分（如 parseQuery → parse / query），只保留长度 3–30 的小写词元，去重排序后写入 graphify-out/.vocab.txt。

4.3 词元选择的硬性约束

读取 graphify-out/.vocab.txt 后，针对用户问题从该列表中挑选至多 12 个语义匹配的词元。约束是硬性的：

• 只能选词表里存在的词元，禁止发明词元；
• 若某个查询概念在词表中找不到合理对应词，跳过它——不要用训练记忆里的近义词替补；
• 若没有任何词表词元能匹配该问题，输出空列表并明确告知用户"语料中没有与该问题相关的词汇"，不要伪造检索；
• 跨语言翻译规则：如俄语"аутентификация" → 在词表中查找 auth、credential、token、security，前提是它们确实在词表中（IFF present）；
• 形态还原规则："handlers" 映射到 handler、"todos" 映射到 todo，同样以词表中存在为前提。

4.4 让扩展可审计

遍历前必须把选择显式打印给用户：

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

如果列表为空，直接说明并停止——不进入遍历。这一步的价值在于：Agent 的查询扩展是模型行为而非确定性算法，显式打印让扩展过程对用户可审查，防止"幻觉词元"悄悄进入检索。

5. Step 1：遍历

用空格连接选出的词元，构成扩展后的查询串，并在后续步骤中用这个串替代用户原始问题作为 QUESTION（原始问题只在最后 save-result 时保留）。

5.1 优先使用 CLI

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

CLI 的完整参数（见 cli.py 的参数解析）：

参数	说明
"QUESTION"	位置参数，扩展后的查询串
--dfs	切换为 DFS 遍历（默认 BFS）
--budget N	token 预算，默认 2000；支持 --budget=3000 等号写法
--context C	上下文过滤（可按模块/文件范围裁剪遍历图），支持多次传入
--graph path	指定图谱文件，默认 graphify-out/graph.json

实现细节（源码级佐证）：

• query 命令把图谱以无向图方式加载并调用 _query_graph_text，mode 取 bfs/dfs，depth=2（见 cli.py）。代码注释明确解释：刻意保持无向，是因为 BFS/DFS 需要同时探索种子节点的调用方与被调方来构建上下文；强制有向会让没有出边的种子丢掉全部 caller 侧结果。
• 注意一个与内联兜底的差异：CLI 固定传 depth=2，而下面文档中的内联兜底脚本默认 BFS 走 3 层、DFS 限深 6 层。
• 输出头部会带上 Graph: <路径> (N nodes)。从源码注释看，这是为了防"语料用错"：graphify-out/ 相对 CWD 解析，在父项目里查询 vendored 子项目时会静默地从错误语料库作答（源码引用了 #2789 的修复）。
• 匹配阶段还处理了一类"关系意图词"（如 calls、uses）：它们描述的是问题问的关系而非符号名，会被从按词种子保证中剔除，避免动词误配成一个诱饵 BFS 根节点（源码注释引用 #2507）。
• 每次查询都会写入 querylog（kind="query"，含 mode、depth、budget、耗时），并刷新查询时间戳，供钩子判断图谱相对查询的新鲜度。

5.2 内联 NetworkX 兜底（CLI 不可用时）

加载 graphify-out/graph.json 后在解释器内直接遍历。完整脚本如下（将 QUESTION 替换为扩展后的查询串，MODE 替换为 bfs 或 dfs，BUDGET 替换为 token 预算，默认 2000 或 --budget N 指定的值）：

$(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 个（start_nodes）；
• 遍历：DFS 沿一条路径尽量走深、限深 6；BFS 按层向外扩展 3 层；
• 预算感知的输出：按相关性（词元重叠数）排序节点，输出以"约 4 字符/token"估算的字符预算截断，超预算时打印截断提示。

输出格式为一行 Traversal 头 + 若干 NODE <label> [src=… loc=…] 行 + 若干 EDGE <a> --<relation> [<confidence>]--> <b> 行——relation 与 confidence 即边上的关系类型与置信度标签，source_location 是引用事实时的出处。

5.3 作答纪律

文档要求 Agent 按以下规则作答：

1. 找出标签与扩展词元最匹配的 1–3 个节点；
2. 从每个起始节点执行相应遍历；
3. 读取子图——节点标签、边关系、置信度标签、源码位置；
4. 只使用图中存在的内容作答，引用具体事实时引用 source_location；
5. 若图中信息不足，直接说明——不要虚构边。

6. 反馈闭环：save-result 与工作记忆

6.1 把答案存回图谱

写完答案后，必须回存，以改进后续查询。关键要求：把扩展词元写进 --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.py），save-result 的完整参数为：

参数	默认值 / 说明
--question	必填
--answer / --answer-file	二选一必填，长答案可走文件
--type	默认 query；本工作流还用到 path_query、explain
--nodes	引用的节点标签列表（0 或多个）
--outcome	可选，取值 useful / dead_end / corrected
--correction	可选，配合 corrected 记录正确答案
--memory-dir	默认 graphify-out/memory

落盘实现为 save_query_result（graphify/ingest.py）。

6.2 --outcome 三态工作记忆

在 save-result 命令上追加 --outcome useful|dead_end|corrected（纠正时再带 --correction "the right answer"），让后续会话从本次会话中学习：

• useful——引用的节点很好地回答了问题（它们成为 preferred sources，即优先来源）；
• dead_end——这个问题/路径走不通，下次不要重新推导；
• corrected——保存的答案是错的，--correction 记录什么才是对的。

6.3 会话开始：reflect --if-stale 与 LESSONS.md

文档要求：在开始图谱工作之前，先刷新并阅读经验教训——运行 graphify reflect --if-stale（廉价、确定性、不经过 LLM；--if-stale 在 LESSONS.md 已经比所有输入都新时使其变成 no-op，例如 git 钩子刚刷新过），然后读取 graphify-out/reflections/LESSONS.md。

该文件列出三类内容：preferred sources（从这里开始找）、known dead ends（跳过它们）、prior corrections（历史纠正）。

源码佐证（cli.py）：reflect 命令默认从 graphify-out/memory 读取记忆、输出到 graphify-out/reflections/LESSONS.md，并暴露两个调参：--half-life-days（信号权重半衰期，默认 30 天）与 --min-corroboration（将一个节点提升为 preferred 所需的独立 useful 结果数，默认 2）。--if-stale 的新鲜度判断由 lessons_fresh 完成（见 graphify/reflect.py）。

文档同时解释了两种部署形态下的成本：不装 git 钩子时，Agent 自己运行 reflect 即可保持经验最新；装了 post-commit 钩子后（graphify/hooks.py 在提交后自动触发 reflect），--if-stale 让会话开始时的这次运行几乎零成本。

7. /graphify path：两点间最短路径

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

graphify path "NODE_A" "NODE_B"

CLI 的实现细节比文档更丰富（cli.py）：

• 支持 --directed / --undirected，且默认有向（源码引用 #2487：方向信息存在于每份 graph.json 中，故默认尊重方向；无向搜索需显式 --undirected）；
• 同节点守卫：两个查询解析到同一节点时直接报错，避免"0 跳"的平凡路径（引用 #828）；
• 歧义告警：最优与次优得分差距小于 10% 时向 stderr 打印 warning；
• 确定性路径：邻接按排序物化，保证同一 graph.json 上每次进程选出的等长路径一致（引用 #2074）；
• 输出中每一跳展示实际存储的关系（同对的并行关系用 / 连接，无关系时如实输出 related），绝不伪造 calls。

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

8. /graphify explain：单个节点的白话解释

目标：对一个节点给出白话解释——它本身 + 与它相连的一切。CLI 优先：

graphify explain "NODE_NAME"

CLI 的实现（cli.py）在文档基础上还有几处值得注意的行为：

• 分层匹配：底层 _find_node 按"源文件精确 → 标签精确 → 前缀 → 子串"的层级取匹配；若多个文件存在同层级竞争节点，会打印歧义清单并提示"改用仓库相对路径或完整节点 id 重试"，而不是猜测；
• Lesson 叠加层：如果 graphify reflect 产生的经验边车文件（.graphify_learning.json）里该节点有记录，输出会附一行 Lesson: preferred source (start here) … / contested … / tentative …，代码变动后还会标注 [code changed since — re-verify]——这正是第 6 节反馈闭环在查询侧的消费点；
• 高 degree 节点友好：连接列表按邻居 degree 排序、最多展示 20 条，超出的按"方向 + 文件"分组统计展示（源码引用 #2009），避免"谁在调用它"这类问题被一个裸计数吞掉。

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

# 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 / degree，以及全部邻居的连接（关系 + 置信度 + 邻居所在文件）。跑完后写一段 3–5 句的解释：这个节点是什么、连接了谁、这些连接为什么重要，并用源码位置作为引用。然后回存：

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

9. 全流程速查：一次完整查询的调用顺序

把上述内容串起来，OpenCode Skill 在一次图谱问答中的标准动作序列是：

1. 会话开始：graphify reflect --if-stale，读取 graphify-out/reflections/LESSONS.md（preferred sources / dead ends / corrections）；
2. 前置检查：确认 graphify-out/graph.json 存在，否则提示先运行 /graphify <path>；
3. 词表扩展：抽取 .vocab.txt → 从词表选至多 12 个词元（禁止发明、空则停止）→ 向用户打印 Query expanded to …；
4. 遍历：graphify query "<扩展串>" [--dfs] [--budget N]，CLI 不可用时执行内联 NetworkX 兜底脚本；
5. 作答：只依据子图内容作答，引用 source_location；
6. 回存：save-result --question "原话" --answer "含扩展痕迹的答案" --type query --nodes …，并按情况追加 --outcome useful|dead_end|corrected 与 --correction。

path 与 explain 两条支线共用同一数据源与回存机制，仅遍历策略不同（nx.shortest_path / 单节点邻域枚举），--type 分别使用 path_query 与 explain 以便在记忆层区分。

适用前提与限制：整套流程要求 graphify-out/graph.json 已由 /graphify <path> 构建；--budget 的 token 预算按约 4 字符/token 估算；受约束扩展的上限（12 词元、词长 3–30）是文档规定的硬约束，用于在"查得到"与"不编造"之间保持平衡。