graphify 图查询实战query、path、explain 三大命令的遍历机制、词表扩展与自改进记忆闭环【免费下载链接】graphifyTurn 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 仓库中 Kiro 平台的技能参考文档 query.md系统讲解对已构建的 graphify 知识图谱执行提问的完整工作流如何先用“受约束的词表扩展”把自然语言问题翻译成图中真实存在的标识符如何用 BFS/DFS 两种遍历模式取回最小子图如何通过graphify path与graphify explain回答“两个概念之间如何连通”“某个节点是什么”两类问题以及如何把每次问答与经验教训useful/dead_end/corrected写回graphify-out/形成跨会话的自改进闭环。读完本文你可以完整复现 graphify 查询流程的每一步操作并理解 graphify/cli.py 与 graphify/serve.py 中评分、选种子、遍历、截断等底层实现。查询流程的定位与前置条件query.md 是 graphify 面向 Kiro Agent 平台的技能参考文件对应其他平台的同构文件如 graphify/skills/claude/references/query.md它在用户向已存在的图提问、或显式运行/graphify path、/graphify explain时被加载。核心技能桩文件把完整的遍历流程指向该文档这些流程优先使用graphify queryCLI在 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-out/.graphify_python记录了构建图时所用的 Python 解释器路径保证后续脚本与构建环境一致。若检查失败应停止查询并提示用户先运行/graphify path构建图。从源码看CLI 侧的默认图路径由 graphify/cli.py 的_default_graph_path()给出同样指向graphify-out/graph.json且query/path/explain三个子命令在执行前都会做“文件存在 是.json 体积不超过上限_enforce_graph_size_cap_or_exit”三重校验。两种遍历模式BFS 与 DFS文档首先给出模式选择表按问题类型二选一模式标志适用场景BFS默认无“X 和什么相连”——获取宽泛上下文近邻优先DFS--dfs“X 如何到达 Y”——追踪一条具体的链或依赖路径两种模式的具体实现可以在仓库源码中逐一对应CLI 路径graphify query最终调用 graphify/serve.py 的_query_graph_text()其内部分别委托给_bfs()serve.py#L934-L962与_dfs()serve.py#L965-L989。CLI 入口在 graphify/cli.py#L1068-L1188 中固定传depth2。值得注意的是源码注释明确说明query刻意保持无向图不像path/explain强制有向因为 BFS/DFS 需要同时探索种子节点的调用方与调用目标边的方向改由每条 link 上的_src/_tgt标记在渲染阶段恢复。内联回退路径文档提供的脚本里BFS 逐层向外扩展3 层DFS 则沿一条路径尽量深入、深度上限6 层以防遍历全图。两种实现都内置了“hub 抑制”非种子的超高连接度节点度数达到度分布 p99、且下限 50不被继续展开避免答案被全局枢纽节点淹没。Step 0受约束的查询扩展遍历前必做这是文档中最关键的一步。graphify 的queryCLI 通过大小写折叠的 substring IDF匹配节点二进制内部没有词干提取、没有同义词、没有跨语言匹配内联回退脚本也以同样的方式匹配。如果用户提问用的词汇与图中节点标签的领域词汇不同用户说“обработчик”/“authentication”图里却叫 “handler”/“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(encodingutf-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)), encodingutf-8) print(fvocab: {len(vocab)} tokens) 该脚本把所有节点标签按 camelCase/snake_case 等规则切分成词元保留长度 3~30 的片段写入graphify-out/.vocab.txt。第 2 步从词表中挑选扩展词元硬性约束读完.vocab.txt后针对用户问题从中挑选至多 12 个语义上匹配查询意图的词元。文档设定了不可违背的约束只能选词表文件中真实存在的词元不得发明词元若某个查询概念在词表里找不到任何合理对应词直接跳过——不得用训练记忆里的近义词替代若词表中没有任何词元与查询匹配输出空列表并明确告知用户“语料对该问题没有相关词汇”不得伪造搜索跨语言翻译俄语 “аутентификация” → 仅当auth、credential、token、security存在于词表中时才选形态学还原“handlers” 映射到handler的前提是它存在“todos” 映射到todo同理。第 3 步显式输出扩展结果保证可审计Query expanded to (from graph vocab, N tokens): [token1, token2, ...]在真正执行查询前必须把选中的词元列表打印给用户使扩展过程可审计若列表为空直接说明并停止不进入遍历。这一步的必要性由 CLI 的评分机制决定graphify/serve.py#L272-L291 的_query_terms()只做分词、去停用词与中文分词有 jieba 时用 jieba否则按双字切分serve.py#L294-L297 定义的匹配档位常量为精确匹配 1000、前缀 100、子串 1、源文件 0.5再叠加 serve.py#L300-L322 的 IDF 权重——error、exception这类命中几百个节点的常见词权重极低而FooBarService这类稀有标识符权重极高。整条链路对“字面不在图中”的词没有任何兜底因此词表扩展是整个查询流程的召回保障。Step 1遍历执行把选中的词元用空格连接成扩展查询串用它而不是用户的原话作为下文中的QUESTION原问题只在最后写回结果save-result时保留。优先使用 CLIgraphify query QUESTION # 或graphify query QUESTION --dfs --budget 3000结合 graphify/cli.py#L1068-L1116 的参数解析graphify query的完整参数面为参数说明默认值QUESTION必填查询问题建议传入词表扩展后的词元串—--dfs切换为 DFS 遍历缺省为 BFSBFS--budget N输出 token 预算也支持--budgetN必须是整数2000--context C上下文过滤可多次出现也支持--contextC无--graph path指定图文件路径graphify-out/graph.jsonCLI 在执行后还会做两件文档未展开、但值得注意的事通过 graphify/querylog.py 的log_query(kindquery, ...)记录本次查询含耗时毫秒数并调用_touch_query_stamp(gp)刷新图文件的查询时间戳见 cli.py#L1177-L1188。CLI 不可用时的内联回退直接加载graphify-out/graph.json在 Python 中完成“找种子 → 遍历 → 读子图 → 只用图中事实作答”。文档给出的完整脚本MODE取bfs或dfsBUDGET为 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(encodingutf-8)) G json_graph.node_link_graph(data, edgeslinks) 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(dataTrue): 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(reverseTrue) 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, keyrelevance, reverseTrue) lines [fTraversal: {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 预算后执行。作答纪律同样写在文档里值得强调找标签与扩展词元最匹配的 1~3 个节点作为起点从每个起点执行对应遍历阅读子图——节点标签、边关系relation、置信度标签confidence、源码位置source location只使用图中存在的事实作答引用具体事实时引用source_location若图中信息不足以回答明说绝不虚构边。对比 CLI 的更精细实现CLI 并不只是“子串计数取前 3”——_pick_seeds()serve.py#L666 起在按分数差距gap_ratio0.2截断种子的同时还保证每个有命中的查询词元至少占用一个种子席位防止某个词的偶然精确匹配1000 档把其他真正相关词元的子串匹配种子全部挤出输出渲染_subgraph_to_text()则把种子节点永远排在预算截断的第一位按 token 预算约 3 字符/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是答案中引用的节点标签列表。这完成了反馈闭环下一次--update会把该问答抽取为图中的一个节点。对照 CLI 实现cli.py#L1312-L1342save-result的完整参数为--question必填、--answer或--answer-file二选一答案也可来自文件、--type默认query、--nodes引用节点列表、--outcome、--correction、--memory-dir默认graphify-out/memory最终落到 graphify/ingest.py 的save_query_result()。工作记忆Work memory让后续会话从本次学习在save-result命令上追加--outcome useful|dead_end|corrected让下一次会话记住这次经验纠正时再附--correction the right answeruseful—— 被引用的节点很好地回答了问题它们将成为优先信源preferred sourcesdead_end—— 该问题/路径走不通下次不要重复推导corrected—— 保存的答案是错的--correction记录正确答案。在每次图工作开始时先刷新并阅读经验教训graphify reflect --if-stale然后读graphify-out/reflections/LESSONS.md其中列出优先信源从这里开始查、已知死路跳过与历史纠正。--if-stale使其在LESSONS.md已比所有输入都新时例如 git hook 刚刷新过直接空转成本几乎为零若未安装 post-commit hook自己跑一次reflect也能保证教训是最新的。reflect的完整参数面见 cli.py#L1343-L1399--memory-dir默认graphify-out/memory、--out默认graphify-out/reflections/LESSONS.md、--graph/--analysis/--labels默认取图文件同目录的.graphify_analysis.json/.graphify_labels.json、--half-life-days信号权重半衰期默认 30 天、--min-corroboration把一个节点晋升为优先信源所需的不同 useful 结果数默认 2、--if-stale。聚合与渲染逻辑在 graphify/reflect.py按半衰期衰减记忆信号、按社区分桶、渲染出带优先/死路/纠正三段的 Markdown并额外写出.graphify_learning.json侧车文件。这个侧车文件还会反哺查询体验graphify explain在打印节点详情时会尝试从侧车读取该节点的“Lesson”行preferred / contested / 临时判定以及代码变更后[code changed since — re-verify]标记见 cli.py#L1615-L1638。也就是说save-result → reflect → explain 展示构成了一个完整的经验闭环。/graphify path两个概念之间的最短路径当用户想知道两个命名概念在图中如何连通时使用graphify path NODE_A NODE_BCLI 不可用时的内联回退脚本$(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(encodingutf-8)) G json_graph.node_link_graph(data, edgeslinks) 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()], reverseTrue ) 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(fCould not find nodes matching: {a_term!r} or {b_term!r}) sys.exit(0) try: path nx.shortest_path(G, src, tgt) print(fShortest 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[i1]]; 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(fNo path found between {a_term!r} and {b_term!r}) except nx.NodeNotFound as e: print(fNode not found: {e}) 把NODE_A、NODE_B换成用户给出的实际概念名然后用自然语言解释路径每一跳意味着什么、为什么重要。解释写完后同样写回$(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_BCLI 版本cli.py#L1400-L1565比内联脚本多了几个工程细节实际使用时应知晓默认有向graphify path A B [--graph path] [--directed|--undirected]默认按有向图求路径方向真值来自每条 link 的_src/_tgt标记无向搜索需显式--undirected歧义保护若两个查询都解析到同一个节点直接报错要求更具体的标签避免输出“0 跳”的平凡路径若最高分与次高分差距小于 10%打印歧义警告确定性邻居顺序按节点 id 排序后构建图保证同一graph.json上每次得到同一条路径输出的每一段展示该节点对的真实存储关系可能有多条并行关系用/连接与置信度并依据_src标记显示真实的--或--方向而不是虚构一个calls。/graphify explain单节点的全连接解释解释单个节点及其全部连接时优先用graphify explain 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(encodingutf-8)) G json_graph.node_link_graph(data, edgeslinks) 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()], reverseTrue ) if not scored or scored[0][0] 0: print(fNo node matching {term!r}) sys.exit(0) nid scored[0][1] data_n G.nodes[nid] print(fNODE: {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_NAMECLI 版本cli.py#L1567-L1699的输出更丰富除标签外还打印节点 ID、Source文件位置、类型、所属社区、度数以及按真实方向--/--分类的连接列表带 relation、confidence 与边所在的调用/引用站点位置。对高连接度节点最多逐条打印 20 条连接其余按“方向 文件”分组统计如-- src/auth.py: 14 connections避免高扇出节点的真正答案被数字淹没此外还会在存在 reflect 侧车时追加 Lesson 行。节点匹配走_find_node()的四级优先档源文件名精确 → 标签精确 → 前缀 → 子串serve.py#L1259 起并用 trigram 倒排索引做候选预筛以加速大图上的查找。源码与测试索引进一步验证的依据本文所有命令行为均可在仓库中复核主要落点如下三条命令的 CLI 实现与参数解析graphify/cli.py 的queryL1068-L1188、save-resultL1312-L1342、reflectL1343-L1399、pathL1400-L1565、explainL1567-L1699评分、IDF、种子选择、BFS/DFS 与预算截断graphify/serve.py 的_query_terms、_compute_idf、_score_query、_pick_seeds、_bfs/_dfs、_subgraph_to_text、_query_graph_text记忆聚合与教训渲染graphify/reflect.py问答持久化graphify/ingest.py 的save_query_result()查询日志graphify/querylog.py行为验证用例tests/test_query_cli.py、tests/test_path_cli.py、tests/test_explain_cli.py、tests/test_query_induced_edges.py、tests/test_reflect.py、tests/test_querylog.py。小结graphify 的查询体系可以概括为四层召回层——用.vocab.txt词表把自然语言问题翻译成图中真实存在的标识符杜绝同义词幻觉遍历层——BFS 取宽上下文、DFS 追具体链配合种子选择差距截断 每词元保底、hub 抑制与 token 预算截断保证答案小而准回答层——query回答“X 相关的是什么”path回答“A 如何到 B”explain回答“X 是什么”且只用图中事实、引用source_location记忆层——save-result含 outcome/correction→reflect --if-stale→LESSONS.md与学习侧车把每次查询的有效路径、死路与纠正沉淀为下一次会话的先验知识。掌握这条链路后你既能在 Kiro/Claude/Codex 等 Agent 中直接使用/graphify技能桩也能在没有任何技能桩的环境下用文档中的内联脚本手工复现全部查询能力。【免费下载链接】graphifyTurn 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考