claude-cookbooks 知识图谱实战用 Claude 结构化输出完成 NER、关系抽取与实体消解全流程【免费下载链接】claude-cookbooksA collection of notebooks/recipes showcasing some fun and effective ways of using Claude.项目地址: https://gitcode.com/GitHub_Trending/an/claude-cookbooks本篇基于 claude-cookbooks 仓库中 knowledge_graph 能力指南 及其配套 notebook、数据与评测脚本展开系统讲解如何仅用提示词无需训练任何模型完成经典知识图谱构建的四个核心任务命名实体识别、关系抽取、实体消解和实体摘要。读完本文你将掌握一条可复制的端到端流水线从非结构化文档抽取三元组、用结构化输出约束 Schema、在内存中组装可多跳查询的图并用 precision/recall 对抽取质量做闭环评测。一、任务定义与仓库结构capabilities/knowledge_graph/README.md 对该模块的定位很明确当你有一堆非结构化文档而问题需要跨文档串联例如“谁和参与过项目 X 的人共事”“哪些供应商与这起事故有关联”单靠 RAG 检索无法把分散的事实链接起来——你需要一张知识图谱实体为节点、类型化关系为边多跳推理因此变成图遍历。传统做法要为领域分别训练命名实体识别模型、训练关系分类器、编写实体消解启发式规则并在数据漂移时持续维护三者。而这条指南的思路是让 Claude 的每一次调用替代其中一个阶段。模块目录结构如下路径作用guide.ipynb主教程 notebook完整流水线抽取 → 消解 → 建图 → 摘要 → 查询 → 评测data/sample_triples.json人工标注的 gold 三元组两篇 Apollo 语料用于评测data/alias_map.json表面形式变体到规范名的映射供评测归一化使用evaluation/eval_extraction.py独立的 precision/recall 打分脚本实体 关系evaluation/README.md评测说明、指标定义与基线预期环境准备按 notebook 中的说明运行环境要求Python 3.11Anthropic API key图的基本概念节点、边、遍历依赖在 notebook 首格中安装%pip install anthropic requests networkx matplotlib python-dotenv pydantic如果用仓库根目录的 uv 工作区来跑评测脚本evaluation/README.md 给出的标准流程是uv sync --all-extras cp .env.example .env # 然后在 .env 中添加 ANTHROPIC_API_KEY双模型分工notebook 中定义了贯穿全流程的两个模型常量这是成本与质量权衡的核心设计EXTRACTION_MODEL claude-haiku-4-5 # 高频、Schema 约束的抽取 SYNTHESIS_MODEL claude-sonnet-4-6 # 消解、摘要等需要权衡多文档证据的环节原文的分工理由是Haiku 负责大批量、模式约束的抽取工作速度与成本优先Sonnet 负责实体消解与摘要需要在多份文档间权衡相互冲突的证据。二、第一步构建语料指南选用 Apollo 计划作为测试床6 篇短小的 Wikipedia 摘要全部提到 NASA、月球、多位宇航员和运载火箭但每篇文章对这些实体的命名略有差异——这正是实体消解需要解决的真实问题。为了控制 token 成本只抓取 Wikipedia REST API 的 summary 而非全文生产管线应切分完整文档但抽取逻辑完全一致ARTICLE_TITLES [ Apollo program, Apollo 11, Neil Armstrong, Saturn V, Buzz Aldrin, Kennedy Space Center, ] WIKI_API https://en.wikipedia.org/api/rest_v1/page/summary/ HEADERS {User-Agent: claude-cookbooks/1.0 (https://github.com/anthropics/claude-cookbooks)} def fetch_summary(title: str) - str: slug quote(title.replace( , _), safe) r requests.get(WIKI_API slug, headersHEADERS, timeout10) r.raise_for_status() return r.json()[extract] documents [] for i, title in enumerate(ARTICLE_TITLES): try: documents.append({id: i, title: title, text: fetch_summary(title)}) except requests.RequestException as e: print(fSkipping {title}: {e}) if not documents: raise RuntimeError(No documents loaded — check network and Wikipedia API availability)注意HEADERS中必须携带可识别的 User-Agent——eval_extraction.py 第 19 行的注释明确指出Wikimedia 策略会拒绝没有身份标识 User-Agent 的请求。三、第二步基于结构化输出的实体与关系抽取经典 NER 用标签标注文本片段PERSON、ORG、LOC经典关系抽取再把片段对分类为关系类型两者传统上都需要按领域标注训练数据。指南把这两个阶段合并为每篇文档一次 Claude 调用关键手段是结构化输出structured outputs把输出形状定义为 Pydantic 模型传给client.messages.parse()返回结果保证通过 Schema 校验并直接作为带类型的 Python 对象——没有正则解析、没有 JSON 解码错误、没有防御性的isinstance检查。Schema 定义EntityType Literal[PERSON, ORGANIZATION, LOCATION, EVENT, ARTIFACT] ENTITY_TYPES [PERSON, ORGANIZATION, LOCATION, EVENT, ARTIFACT] class Entity(BaseModel): name: str type: EntityType description: str class Relation(BaseModel): source: str predicate: str target: str class ExtractedGraph(BaseModel): entities: list[Entity] relations: list[Relation]两个容易忽略但影响下游的 Schema 设计点description字段要求每个实体附带一句基于当前文档的描述。它在抽取阶段看似冗余但正是下一步实体消解的消歧依据同名不同物靠它区分。谓词约束提示词要求谓词为短动词短语commanded、launched from、part of且每条关系必须连接两个已抽取的实体。抽取提示词与调用完整的抽取提示词notebook 与评测脚本保持同步脚本版本略简EXTRACTION_PROMPT Extract a knowledge graph from the document below. document {text} /document Guidelines: - Extract only entities that are central to what this document is about — skip incidental mentions. - For each entity, write a one-sentence description grounded in this document. These descriptions are used later to disambiguate entities with similar names. - Predicates should be short verb phrases (commanded, launched from, part of). - Every relation must connect two entities you extracted. def extract(text: str) - ExtractedGraph: response client.messages.parse( modelEXTRACTION_MODEL, max_tokens2048, messages[{role: user, content: EXTRACTION_PROMPT.format(texttext)}], output_formatExtractedGraph, ) return response.parsed_output批量抽取时逐文档调用并带上 API 错误兜底raw_entities [] raw_relations [] for doc in documents: try: result extract(doc[text]) except anthropic.APIError as e: print(fSkipping {doc[title]}: {e}) continue for ent in result.entities: raw_entities.append({**ent.model_dump(), source_doc: doc[title]}) for rel in result.relations: raw_relations.append({**rel.model_dump(), source_doc: doc[title]})notebook 中的一次真实运行结果为每篇文档 310 个实体、210 条关系总计36 个原始实体、34 条原始关系。把结果按类型分组打印后可以直观看到消解要解决的问题例如 PERSON 类下同时出现Buzz Aldrin、Edwin Aldrin、Neil Armstrong、Neil Alden Armstrong等重复表面形式。四、第三步Claude 驱动的实体消解原始抽取给出的是重叠的提及NASA 与 National Aeronautics and Space Administration、the Moon 与 Moon。直接建图会得到一个碎裂的图——同一概念被拆散在互不相连的节点上。传统方法用字符串相似度编辑距离、token 的 Jaccard加阻塞规则。它对拼写错误有效但会彻底失效于Edwin Aldrin与Buzz Aldrin——这两个名字零字符重叠却指向同一人。指南的做法是让 Claude 对每种类型的实体做聚类并把抽取阶段生成的一句话描述作为消歧上下文。原文强调描述在这里是决定性的——Armstrong — 第一个登上月球的人和 Armstrong — 爵士小号手同名但绝不能合并。消解 Schema 与提示词class Cluster(BaseModel): canonical: str aliases: list[str] class ResolvedClusters(BaseModel): clusters: list[Cluster] RESOLVE_PROMPT Below are {entity_type} entities extracted from several documents. Some are different surface forms of the same real-world entity. entities {entity_list} /entities Cluster them. Each input name must appear in exactly one clusters aliases list. Entities that are genuinely distinct get their own single-element cluster. Use the descriptions to avoid merging entities that merely share a name. The canonical name should be the most complete, unambiguous form. def resolve(entity_type: str, entities: list[dict]) - list[Cluster]: unique {} for e in entities: unique.setdefault(e[name], e[description]) entity_list \n.join(f- {name}: {desc} for name, desc in unique.items()) response client.messages.parse( modelSYNTHESIS_MODEL, max_tokens2048, messages[ { role: user, content: RESOLVE_PROMPT.format(entity_typeentity_type, entity_listentity_list), } ], output_formatResolvedClusters, ) return response.parsed_output.clusters逐类型执行消解并构建alias_to_canonical映射API 失败时回退为每个名字自成一个单元素簇alias_to_canonical {} canonical_info {} for etype in ENTITY_TYPES: entities_of_type [e for e in raw_entities if e[type] etype] if not entities_of_type: continue try: clusters resolve(etype, entities_of_type) except anthropic.APIError as e: print(fResolve failed for {etype}: {e}; treating each name as its own cluster) clusters [ Cluster(canonicaln, aliases[n]) for n in {x[name] for x in entities_of_type} ] for cluster in clusters: canonical_info[cluster.canonical] {type: etype, aliases: cluster.aliases} for alias in cluster.aliases: alias_to_canonical[alias] cluster.canonical实际运行中 24 个唯一名字被合并为 22 个规范实体例如Buzz Aldrin (also: Edwin Aldrin)、Neil Alden Armstrong (also: Neil Armstrong)。原文特别提示了两种需要盯防的失效模式漏名丢节点Claude 若把某个原始名字漏出所有簇alias_to_canonical就没有它的条目该名字会静默地从图中消失。生产级消解器应对未匹配名字回退为单元素簇保证不丢东西。过度合并具体任务 Gemini 12 可能因描述与更宽泛的 Project Gemini 重叠而被折并进去。前者丢节点后者丢精度两者都值得在输出中抽查。五、第四步组装与可视化有了干净的别名映射后把所有关系端点改写到规范形式加载进 NetworkX。原文解释了结构选型使用MultiDiGraph因为两个实体之间可能有多条不同谓词的边launched from 与 operated by且方向有意义Armstrong commanded Apollo 11 与 Apollo 11 commanded Armstrong 不是同一条边。每个节点携带类型、提及它的文档集合与提及次数每条边携带谓词和来源文档G nx.MultiDiGraph() for e in raw_entities: canonical alias_to_canonical.get(e[name]) if canonical is None: continue if canonical not in G: G.add_node( canonical, typecanonical_info[canonical][type], descriptione[description], source_docs[], mentions0, ) G.nodes[canonical][source_docs].append(e[source_doc]) G.nodes[canonical][mentions] 1 for r in raw_relations: src alias_to_canonical.get(r[source]) tgt alias_to_canonical.get(r[target]) if src and tgt and src ! tgt: G.add_edge(src, tgt, predicater[predicate], source_docr[source_doc]) for n in G.nodes: G.nodes[n][source_docs] sorted(set(G.nodes[n][source_docs]))运行结果22 个节点、34 条边、1 个弱连通分量。可视化用 spring 布局节点大小按度数缩放、颜色按类型编码PERSON 蓝、ORGANIZATION 橙、LOCATION 青、EVENT 红、ARTIFACT 紫COLOR { PERSON: #4e79a7, ORGANIZATION: #f28e2c, LOCATION: #76b7b2, EVENT: #e15759, ARTIFACT: #af7aa1, } plt.figure(figsize(14, 10)) pos nx.spring_layout(G, k1.5, seed42) node_colors [COLOR[G.nodes[n][type]] for n in G.nodes] node_sizes [300 200 * G.degree(n) for n in G.nodes] nx.draw_networkx_nodes(G, pos, node_colornode_colors, node_sizenode_sizes, alpha0.9) nx.draw_networkx_labels(G, pos, font_size8) nx.draw_networkx_edges(G, pos, alpha0.3, arrowsTrue, arrowsize10)原文给出的读图方法有明确判据节点大小反映度数hub 是黏合整个语料的实体颜色反映类型分布如果图基本是一种颜色说明语料面太窄单一连通分量说明实体消解做对了——碎裂的小岛意味着本应合并的变体没有被合并。六、第五步实体摘要消解完成后每个节点只有首个提及它的文档里那句一行描述。对于出现在多篇文档中的 hub 节点可以做得更好汇集所有提及把图邻域作为上下文让 Claude 综合出正式档案。原文把这一步描述为把标签图变成知识图的关键——这些摘要就是搜索结果中展示的节点内容或是喂给下游 QA 的输入。Schema 要求摘要可溯源、时间范围结构化class TimeRange(BaseModel): start: str # YYYY 或 YYYY-MM或 unknown end: str # YYYY 或 YYYY-MM或 ongoing class EntityProfile(BaseModel): summary: str key_facts: list[str] time_range: TimeRange SUMMARIZE_PROMPT Generate a knowledge-graph profile for this entity. Entity: {name} ({etype}) Source excerpts mentioning this entity: {excerpts} Known relations in the graph: {relations} Write a 2-3 paragraph factual summary synthesized from the excerpts, resolving any contradictions by preferring the most specific claim. Include 3-5 atomic key facts, each traceable to the sources. For the time range, use YYYY or YYYY-MM format, or unknown/ongoing where appropriate. Do not invent facts not supported by the excerpts.def summarize_entity(name: str) - EntityProfile: # 读取 notebook 前文构建的模块级 G 与 documents docs_with_entity G.nodes[name][source_docs] excerpts \n\n.join( f[{d[title]}]\n{d[text]} for d in documents if d[title] in docs_with_entity ) relations ( \n.join( f- {name} --{d[predicate]}-- {tgt} for _, tgt, d in G.out_edges(name, dataTrue) ) \n \n.join( f- {src} --{d[predicate]}-- {name} for src, _, d in G.in_edges(name, dataTrue) ) ) response client.messages.parse( modelSYNTHESIS_MODEL, max_tokens1500, messages[ { role: user, content: SUMMARIZE_PROMPT.format( namename, etypeG.nodes[name][type], excerptsexcerpts, relationsrelations ), } ], output_formatEntityProfile, ) return response.parsed_output只对度数最高的 3 个 hub 节点Apollo program、Apollo 11、John F. Kennedy Space Center做摘要结果写回节点属性hub_nodes [n for n, _ in sorted(G.degree(), keylambda x: -x[1])[:3]] for node in hub_nodes: profile summarize_entity(node) G.nodes[node][profile] profile.model_dump()提示词中几个值得注意的约束矛盾时优先更具体的说法35 条原子化关键事实且每条可溯源不得虚构摘录不支持的事实。这正是把摘要控制在综合而非发挥边界内的写法。七、第六步多跳查询与图上下文对照建图的收益在于多跳推理回答那些事实从不共现于单一文档的问题。与 Apollo 11 相关的人员关联到哪些地点需要抽取器在一篇文档里找到 人→任务 边、在另一篇里找到 人→地点 边再靠消解器统一了人名节点让这些边真正相接。指南把相关子图序列化为三元组文本让 Claude 在其上推理并且做了有无图上下文的双对照实验def serialize_subgraph(center: str, hops: int 2) - str: nodes {center} frontier {center} for _ in range(hops): nxt set() for n in frontier: nxt | set(G.successors(n)) | set(G.predecessors(n)) frontier nxt - nodes nodes | frontier sub G.subgraph(nodes) lines [f({s}) --[{d[predicate]}]-- ({t}) for s, t, d in sub.edges(dataTrue)] return \n.join(sorted(set(lines))) def ask(question: str, graph_context: str | None None) - str: if graph_context is not None: prompt fAnswer using only the knowledge graph below. Cite the specific edges that support your answer. graph {graph_context} /graph Question: {question} else: prompt question response client.messages.create( modelSYNTHESIS_MODEL, max_tokens500, messages[{role: user, content: prompt}], ) text_block next((b for b in response.content if b.type text), None) if text_block is None: raise ValueError(fNo text block in response (stop_reason{response.stop_reason})) return text_block.textcenter next((n for n in G.nodes if Apollo in n), hub_nodes[0]) subgraph serialize_subgraph(center, hops2) question Which locations are connected to people who were part of Apollo 11, and how? print(WITHOUT graph context:) print(ask(question)) print(WITH graph context:) print(ask(question, subgraph))对照结果很有代表性无图上下文的回答调用了 Claude 的预训练知识对 Apollo 11 这种知名事件可能恰好正确列出了大量语料里根本不存在的地点有图上下文的回答则每条论断都引用了具体边如(Neil Alden Armstrong) --[walked on]-- (Moon)并诚实声明图中没有其他 Apollo 11 乘员的地点数据。原文的结论是接地grounded的答案是可溯源的——每个论断引用的是从特定文档抽取的边在 Claude 没有先验知识的私有语料上只有接地答案可用。八、评测对 gold 集合的 Precision/Recall质量度量采用对 gold 集合的 precision/recall。仓库附带了一个小型人工标注集 data/sample_triples.json覆盖两篇文章Apollo 11 与 Neil Armstrong的实体与关系例如{ Apollo 11: { entities: [ {name: Apollo 11, type: EVENT}, {name: Neil Armstrong, type: PERSON}, {name: NASA, type: ORGANIZATION}, ... ], relations: [ {source: Neil Armstrong, predicate: commanded, target: Apollo 11}, {source: Apollo 11, predicate: launched from, target: Kennedy Space Center}, ... ] }, Neil Armstrong: { ... } }data/alias_map.json 则把表面形式变体归一到 gold 名字保证 the Moon 与 Moon 计为同一命中映射条目例如national aeronautics and space administration → nasa、edwin aldrin → buzz aldrin、ksc → kennedy space center。notebook 内联评测原始抽取 vs 消解后核心打分函数就是标准集合 P/R/F1def norm(name: str) - str: lower name.lower().strip() return ALIASES.get(lower, lower) def prf(predicted: set, gold: set) - tuple[float, float, float]: tp len(predicted gold) p tp / len(predicted) if predicted else 0.0 r tp / len(gold) if gold else 0.0 f1 2 * p * r / (p r) if (p r) else 0.0 return p, r, f1data_dir Path(data) if not data_dir.exists(): data_dir Path(capabilities/knowledge_graph/data) with open(data_dir / sample_triples.json, encodingutf-8) as f: gold json.load(f) with open(data_dir / alias_map.json, encodingutf-8) as f: ALIASES json.load(f) for doc_title, labels in gold.items(): gold_names {norm(e[name]) for e in labels[entities]} raw {norm(e[name]) for e in raw_entities if e[source_doc] doc_title} rp, rr, rf prf(raw, gold_names) resolved { norm(alias_to_canonical.get(e[name], e[name])) for e in raw_entities if e[source_doc] doc_title } _, resolved_r, _ prf(resolved, gold_names) print(f{doc_title:20} raw F1{rf:.2f} (P{rp:.2f} R{rr:.2f}) resolved R{resolved_r:.2f}) missed gold_names - resolved if missed: print(f still missed after resolution: {, .join(sorted(missed))})这次运行的结果每篇 raw 精确率都是 1.00召回分别为 0.55 与 0.38显示抽取器偏保守——宁可少抽高置信实体也不追求穷尽覆盖。原文特别解释了一个评分伪影当消解器选定的规范形式不在 alias map 覆盖范围内比如选了 Neil Alden Armstrong 这种冗长形式消解后的召回反而会下降——因为消解前能匹配 gold 的名字消解后不再匹配。这不是消解器的 bug而是评分伪影修复方法是每当看到打分器不认识的规范形式就扩充alias_map.json。独立评测脚本notebook 内联部分只评实体evaluation/eval_extraction.py 同时评测实体与关系并在 evaluation/README.md 中明确了口径实体 P/R/F1抽取实体经归一化小写 alias map后与同文档 gold 实体匹配即计为 TP。关系 P/R/F1两端点归一化后与 gold (source, target) 对匹配即计为 TP。谓词语义被忽略commanded 与 was commander of 都算对甚至同一对实体间语义错误的 destroyed 也算对。因此报告的关系召回是一个上界——它度量的是抽取器找对了连接而不是标注对了谓词。更严格的打分需要增加谓词相似度检查例如对每个候选对做一次 Claude judge 调用。脚本从仓库根目录运行uv run python capabilities/knowledge_graph/evaluation/eval_extraction.py从源码看eval_extraction.py 的main()对 gold 中每篇文章重新抓取 summary、用与 notebook 相同的claude-haiku-4-5与client.messages.parse()抽取注意脚本版 PROMPT 不含 description 与谓词短语两条指南只保留只抽核心实体 关系必须连接已抽实体再分别对实体集合与 (source, target) 对集合计算 P/R/F1最后做跨文档宏平均macro-average并打印漏掉的 gold 实体清单以便定位短板。evaluation/README.md 给出的预期基线claude-haiku-4-5 指南中的抽取提示词MetricPRF1Entities0.80–0.900.70–0.850.75–0.85Relations0.70–0.850.55–0.700.60–0.75原文明确这些区间只是参考值模型非确定性会使实际分数逐次波动关系召回是最难的数字——抽取器倾向于保守少抽高置信边而非穷尽。若提示词调向高召回如 extract every stated relationship, even minor ones则是以精确率换召回率。九、生产环境扩展要点notebook 末尾的 Scaling up 一节给出了四条直接可操作的扩展建议抽取成本Haiku 便宜到可以在大语料上跑当抽取 Schema 与指令固定时**提示词缓存prompt caching**能进一步降本——系统提示与 Schema 走缓存价只为文档正文付全价。Message Batches API对可等待 24 小时内的任务给出 50% 折扣。消解的可扩展性一次性把一万个 PERSON 实体喂给 Claude 不可行。先做阻塞blocking用廉价信号同姓、token 重叠、embedding 相似度分组让 Claude 只在小块内部仲裁。上文消解提示词在 50100 个实体的块上无需修改即可工作。增量更新新文档到达时抽取其实体、与现有规范集合而非彼此做消解、只加新边只有当某实体的来源文档集合发生实质变化时才重新摘要。存储NetworkX 撑到几十万条边都没问题。再往上Schema 直接映射到属性图Neo4j、Neptune或三张 Postgres 表entities(id, name, type, summary)、relations(source_id, target_id, predicate)、aliases(entity_id, alias)。抽取与消解代码不变只换持久层。十、小结这条 notebook 用纯提示词搭出了完整知识图谱流水线每一环对应一个被 LLM 替代的传统组件抽取每篇文档一次结构化输出调用替代了训练的 NER 模型 训练的关系分类器Pydantic Schema 就是唯一的训练消解Claude 利用抽取阶段的描述作上下文对表面形式聚类抓住 Edwin Aldrin → Buzz Aldrin 这类字符串相似度完全漏掉的情况摘要hub 节点获得跨所有提及文档综合的档案带结构化时间范围与可溯源的关键事实查询序列化子图让 Claude 以边级引用回答多跳问题把答案接地在自建图而非预训练知识上。evaluation/ 目录下的评测装置提供反馈闭环改抽取提示词、重跑打分器、观察 F1 变化——这个闭环是把 demo 变成生产系统的关键。仓库中还有三份相关 cookbook 可作为延伸参考tool_use/extracting_structured_json.ipynb同一抽取模式的 tool-use 版本适合已在 agentic 工具调用流程中的场景、capabilities/retrieval_augmented_generation/guide.ipynb需要文档检索而非事实遍历时的互补方案、capabilities/contextual-embeddings/guide.ipynb把索引前先加上下文的同样思路应用到向量检索。【免费下载链接】claude-cookbooksA collection of notebooks/recipes showcasing some fun and effective ways of using Claude.项目地址: https://gitcode.com/GitHub_Trending/an/claude-cookbooks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考