Graphify 知识图谱查询全流程指南：query / path / explain 与自学习反馈闭环

2026-09-06 19:23:52作者：郜逊炳

Turn any codebase, with its docs, SQL schemas, configs, and PDFs, into a queryable knowledge graph. A /graphify skill for Claude Code, Cursor, Codex, and Gemini CLI: local deterministic AST parsing, every edge explained, no vector store.

项目地址：https://gitcode.com/GitHub_Trending/graph/graphify

导读

本文是 graphify 的查询参考（graphify/skills/amp/references/query.md）的完整展开，面向在已构建好的代码知识图谱上提问、追溯调用链、解释单个节点语义的场景。你将掌握三件事：如何在「图谱存在」的前提下用 graphify query / path / explain 三个子命令做结构化检索；当用户措辞与图谱标签词汇不一致时，如何通过**受约束的词汇表扩展（constrained query expansion）**避免查询退化为噪声；以及如何用 save-result 与 reflect 把每次问答沉淀为可复用的"经验"，让图谱越用越准。

适用前提：本流程建立在「已有 graph.json」之上。若尚未建图，请先运行 /graphify <path>（或 graphify extract）完成抽取，再回到本文。

一、先决条件：确认图谱已构建

所有查询类命令都依赖构建产物 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-out/.graphify_python 是 graphify 写入的运行时指示文件，内容指向本次构建实际使用的 Python 解释器，因此无论代码库环境如何，都能用同一解释器拉起 NetworkX 与 graphify 模块。
若检查失败，应停下并提示用户先运行 /graphify <path>，而不是在空图谱上继续遍历。

两种遍历模式：BFS 还是 DFS？

模式	标志	适用场景
BFS（默认）	（无）	"X 与什么相连？" —— 需要广谱上下文，按最近邻居优先展开
DFS	`--dfs`	"X 是如何到达 Y 的？" —— 需要沿着某条具体调用链 / 依赖路径深挖

在 CLI 层，这个选择直接映射到 cli.py 中 graphify query 的参数解析（use_dfs = "--dfs" in sys.argv），并最终把 mode 传给服务端的遍历实现 _query_graph_text（serve.py 中的 _query_graph_text）。

二、Step 0（必需）：受约束的查询词扩展

这是本流程中最容易被忽略、却也最关键的环节。graphify 的 query CLI 与内联回退逻辑都使用大小写折叠的子串匹配 + IDF 加权来命中节点——二进制内部没有词干化（stemming）、没有同义词、没有跨语言匹配。当用户问句用词与图谱标签词汇不一致时（比如用户说俄语 "обработчик"、图谱里叫 handler；用户说 "authentication"、图谱里叫 Guardian），字面匹配器会返回 0 个命中，答案随之坍缩成噪声。

解决方式不是"凭训练记忆发明 token"，而是先从图谱真实词汇中提取词表，再在词表范围内做语义扩展：

1. 从节点标签抽取 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(sorted(vocab)), encoding='utf-8')
print(f'vocab: {len(vocab)} tokens')
"

这段脚本的细节值得注意：正则 [A-Z]+(?=[A-Z][a-z])|[A-Z]?[a-z]+|[A-Z]+ 负责把 camelCase、PascalCase、ACRONYM 拆成独立单词；token 长度限制在 3~30 之间，既滤掉了无意义的单双字母，也保留像 api、jwt 这类有信息量的短 token。

2. 读取词表并挑选扩展词。 为用户的问句从词表中挑选最多 12 个与查询意图语义匹配的 token，并遵守硬性约束：

只能挑选词表文件中真实存在的 token，不得发明；
若某个查询概念在词表中找不到可信 token，直接跳过它，不要用训练记忆中的近似同义词顶替；
若整个问句在词表中一个匹配都没有，就输出空列表并明确告知用户"语料中没有与该问题相关的词汇"，绝不伪造搜索；
跨语言翻译：俄语 "аутентификация" → 仅当词表中存在 auth、credential、token、security 时才选用；
词形处理："handlers" 仅当存在 handler 时映射过去；"todos" 仅当存在 todo 时映射过去。

3. 把扩展结果显式打印给用户，保证扩展过程可审计：

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

如果列表为空，直接如实说明并停止，不要进入遍历步骤。

为什么要"先看词表再扩展"？从源码看，_score_nodes（serve.py 中的评分函数）与内联回退的评分都是纯字面 overlap 打分，加上按文档频率计算的 IDF（_compute_idf，在 serve.py 中实现）——IDF 的思想是：logger 这类在语料中常见、信息量低的词权重低，而 wie、funktioniert 这类罕见 token 会获得很高的 IDF 权重。整个评分体系都以"真实出现在图谱里的词"为前提，因此在进入评分前把问句"翻译"成语料内词汇，是让这套 IDF 机制生效的前提。

三、Step 1：遍历（Traversal）

把选出的 token 用空格连接成扩展后的查询串，作为下面的 QUESTION——不是原始用户问句（原始问句只保留给最后一步 save-result 使用）。

3.1 优先使用 CLI

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

CLI 参数（对应 cli.py 的 graphify query 分支）说明：

参数	含义	默认值
`QUESTION`	位置参数，扩展后的查询串	必填
`--dfs`	切换 DFS 模式；缺省为 BFS	缺省 BFS
`--budget N`	输出 token 预算，超出则截断（`--budget=N` 写法同样支持）	`2000`
`--context C`	追加上下文过滤条件，可多次传入	无
`--graph path`	指定图谱 JSON 路径	默认 `graphify-out/graph.json`

在实现层面，CLI 把图加载为 NetworkX 图后调用 _query_graph_text(G, question, mode=_mode, depth=2, token_budget=budget, ...)，并把每次查询写入 query 日志、--budget 直接控制输出预算（cli.py）。

3.2 CLI 不可用时的内联回退

若 CLI 不可用，加载 graphify-out/graph.json 用 NetworkX 就地遍历：

$(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]  # 与词表阈值一致；保留 api/jwt/ios

# 找最佳匹配的起点节点
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 预算约束输出（约每 token 4 字符）
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 指定值保持一致）。

该回退脚本的算法参数（与 serve.py 的实现互为印证）：

起点节点选取：最多取 3 个"标签中包含查询词数量最多"的节点（按得分降序）；
DFS：显式深度限制为 6，防止沿调用链一路贯穿整个图谱；
BFS：层数上限为 3，保证"最近邻优先"的广谱上下文；
预算截断：约按 4 字符 / token 换算为字符预算，超出即截断并提示使用 --budget N 扩大；
输出行：NODE 行带 source_file 与 source_location，EDGE 行带 relation 与 confidence 标签；MultiGraph 情况先取第一条平行边数据，避免同节点对间多条 relation 造成歧义。

回答环节的纪律是硬性的：只依据子图输出回答，引用具体事实时标注 source_location；若图谱信息不足，明确说明"信息不足"，绝不幻觉边。

四、把答案写回图谱：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：完整回答文本（内含扩展 token 轨迹）；
NODE1 NODE2：回答中引用的节点标签列表。

从 cli.py 的 graphify save-result 分支可以看到，它内部调用 ingest.py 的 save_query_result，把问答以 YAML frontmatter 形式写入 graphify-out/memory/，支持 --answer-file（从文件读答案，适合长文）、--type（默认 query）等参数。

工作记忆（自改进循环）

给 save-result 追加 --outcome，让未来的会话从本次问答中学习（cli.py 中 --outcome 只接受三个枚举值）：

graphify save-result --question "..." --answer "..." --type query \
  --nodes NODE1 NODE2 --outcome useful|dead_end|corrected \
  [--correction "the right answer"]

useful —— 引用的节点很好地回答了问题（这些节点将变为 preferred sources / 首选来源）；
dead_end —— 该问题/路径毫无产出，下次不要再推导一遍；
corrected —— 之前保存的答案是错的；--correction 记录正确内容。

会话开始时：刷新并阅读经验库

在开始图谱相关工作时，先运行一次轻量的反思刷新（确定性执行，不消耗 LLM）：

graphify reflect --if-stale

然后阅读 graphify-out/reflections/LESSONS.md。它会列出 preferred sources（从这里开始）、已知死胡同 dead ends（直接跳过）以及历史 corrections。

从 cli.py 的 reflect 分支与 reflect.py 的实现可见其工作机理：

--if-stale：当 LESSONS.md 已经比所有输入（memory 目录、graph.json、.graphify_analysis.json、.graphify_labels.json）都新时直接跳过（no-op）——例如 post-commit git hook 刚刷新过它；此时你手动再跑一次几乎零成本；
--half-life-days（默认 30）：经验信号的权重按天做时间衰减，越久远的结果影响力越小；
--min-corroboration（默认 2）：节点需要被足够多个不同的 useful 结果佐证才会晋升为 preferred——单次保存不足以把一个节点"封神"；
结果写入 graphify-out/reflections/LESSONS.md，同时生成按节点粒度的学习 sidecar 覆盖层，graphify explain 展示节点时也会叠加显示该节点是否被标记为 preferred / contested（cli.py 的 explain 分支）——若代码已变化，还会标注 [code changed since — re-verify]。

即使没有安装 git hook，手动运行 reflect 也能让经验库保持最新；安装了 post-commit hook 时，--if-stale 保证会话开始时的开销趋近于零。

五、`/graphify path`：求两个概念间的最短路径

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

graphify path "NODE_A" "NODE_B"

5.1 CLI 参数补充

（详见 cli.py 的 path 分支）

--graph path：指定图谱文件；
--directed / --undirected：默认有向（边方向以 _src/_tgt 标记为准，缺失时回退到弧的原始端点顺序）；两者互斥。有向搜索失败时会提示"未找到有向路径，可加 --undirected 忽略方向重试"；
安全性检查：当源与目标解析到同一个节点时直接报错退出（此时最短路径为 0 跳，几乎从不是调用者想要的）；两端点得分接近（差距 <10%）时给出歧义警告；为确保确定性，路径搜索在排序后的图上进行，避免进程间结果漂移。

5.2 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 替换为用户实际提到的概念名，然后用通俗语言解释路径含义——每一跳代表什么、为什么有意义。

路径解析采用"词级 overlap 打分"选择端点（与 serve.py 的 _pick_scored_endpoint 思路一致）。解析成功后逐跳输出 label --relation--> [confidence]，把链路的真实关系与置信度一并呈现，便于验证每一步是否可信。

路径解释写完后同样保存回去：

$(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 输出示例结构（对应 cli.py 的 explain 分支）：节点标签、节点 ID、源码位置 source_file/source_location、文件类型、社区归属、度（degree），以及该节点在工作记忆中的经验叠加（preferred / contested / 带分数与陈旧标记）。若标签命中多个分布在不同文件里的节点（歧义），CLI 会列出候选并提示改用仓库相对路径或完整节点 ID 重试。

内联实现

$(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 句解释：该节点是什么、连接了什么、这些连接为何重要，并以 source_file 作为引用依据。

解释写完后保存回去：

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

七、三条流程的统一心智模型

把 query / path / explain 放到一起看，它们构成完整的三段式闭环：

锚定（Anchor）——把用户的自然语言问句映射到图谱内的真实节点。query 用"词表约束扩展 + 最多 3 起点"，path 用两个端点解析，explain 用单节点解析；三者共享同一个约束：只认图谱里真实存在的词与节点，不臆造；
遍历/推导（Traverse / Derive）——query 做 BFS（深度 3）或 DFS（深度 6）子图展开并输出带预算约束的节点-边列表；path 做确定性的最短路径并给出逐跳 relation + confidence；explain 枚举单节点的全部邻接边。输出都保留 source_location、relation、confidence 三个溯源字段，确保回答可核实；
沉淀（Persist）——统一经 save-result 写入 memory 目录（graphify-out/memory/），标记 --outcome；reflect 再将其聚合成 LESSONS.md 的 preferred sources / dead ends / corrections，并反向叠加到后续 explain 展示中。

值得强调的工程事实：从 cli.py、serve.py、reflect.py、ingest.py 的实现看，这一整套流程默认不依赖任何 LLM——匹配、IDF 加权、遍历、衰减聚合都是确定性的图算法；LLM（或 Agent）只负责两件"需要语义理解"的事：把用户问句在词表约束下翻译成图谱词汇（Step 0），以及把结构化子图输出组织成自然语言回答。这解释了为什么这套查询体系"确定性、可审计、无向量库"：查询可信度来自图谱本身的边与置信度标签，而非检索模型的黑箱排序。