claude-cookbooks 知识图谱实战：用 Claude 结构化输出完成 NER、关系抽取与实体消解全流程

本篇基于 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.11+
• Anthropic 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, headers=HEADERS, timeout=10)
    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(f"Skipping {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 设计点：

1. description 字段：要求每个实体附带一句基于当前文档的描述。它在抽取阶段看似冗余，但正是下一步实体消解的消歧依据（同名不同物靠它区分）。
2. 谓词约束：提示词要求谓词为短动词短语（"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(
        model=EXTRACTION_MODEL,
        max_tokens=2048,
        messages=[{"role": "user", "content": EXTRACTION_PROMPT.format(text=text)}],
        output_format=ExtractedGraph,
    )
    return response.parsed_output

批量抽取时逐文档调用并带上 API 错误兜底：

raw_entities = []
raw_relations = []

for doc in documents:
    try:
        result = extract(doc["text"])
    except anthropic.APIError as e:
        print(f"Skipping {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 中的一次真实运行结果为每篇文档 3～10 个实体、2～10 条关系，总计 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 cluster's 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(
        model=SYNTHESIS_MODEL,
        max_tokens=2048,
        messages=[
            {
                "role": "user",
                "content": RESOLVE_PROMPT.format(entity_type=entity_type, entity_list=entity_list),
            }
        ],
        output_format=ResolvedClusters,
    )
    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(f"Resolve failed for {etype}: {e}; treating each name as its own cluster")
        clusters = [
            Cluster(canonical=n, 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)。

原文特别提示了两种需要盯防的失效模式：

1. 漏名丢节点：Claude 若把某个原始名字漏出所有簇，alias_to_canonical 就没有它的条目，该名字会静默地从图中消失。生产级消解器应对未匹配名字回退为单元素簇，保证不丢东西。
2. 过度合并：具体任务 "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,
            type=canonical_info[canonical]["type"],
            description=e["description"],
            source_docs=[],
            mentions=0,
        )
    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, predicate=r["predicate"], source_doc=r["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, k=1.5, seed=42)
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_color=node_colors, node_size=node_sizes, alpha=0.9)
nx.draw_networkx_labels(G, pos, font_size=8)
nx.draw_networkx_edges(G, pos, alpha=0.3, arrows=True, arrowsize=10)

原文给出的读图方法有明确判据：节点大小反映度数，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, data=True)
        )
        + "\n"
        + "\n".join(
            f"- {src} --{d['predicate']}--> {name}" for src, _, d in G.in_edges(name, data=True)
        )
    )

    response = client.messages.parse(
        model=SYNTHESIS_MODEL,
        max_tokens=1500,
        messages=[
            {
                "role": "user",
                "content": SUMMARIZE_PROMPT.format(
                    name=name, etype=G.nodes[name]["type"], excerpts=excerpts, relations=relations
                ),
            }
        ],
        output_format=EntityProfile,
    )
    return response.parsed_output

只对度数最高的 3 个 hub 节点（Apollo program、Apollo 11、John F. Kennedy Space Center）做摘要，结果写回节点属性：

hub_nodes = [n for n, _ in sorted(G.degree(), key=lambda x: -x[1])[:3]]

for node in hub_nodes:
    profile = summarize_entity(node)
    G.nodes[node]["profile"] = profile.model_dump()

提示词中几个值得注意的约束：矛盾时优先更具体的说法；3～5 条原子化关键事实且每条可溯源；不得虚构摘录不支持的事实。这正是把摘要控制在"综合"而非"发挥"边界内的写法。

七、第六步：多跳查询与图上下文对照

建图的收益在于多跳推理：回答那些事实从不共现于单一文档的问题。"与 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(data=True)]
    return "\n".join(sorted(set(lines)))

def ask(question: str, graph_context: str | None = None) -> str:
    if graph_context is not None:
        prompt = f"""Answer 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(
        model=SYNTHESIS_MODEL,
        max_tokens=500,
        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(f"No text block in response (stop_reason={response.stop_reason})")
    return text_block.text

center = next((n for n in G.nodes if "Apollo" in n), hub_nodes[0])
subgraph = serialize_subgraph(center, hops=2)
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/F1：

def 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, f1

data_dir = Path("data")
if not data_dir.exists():
    data_dir = Path("capabilities/knowledge_graph/data")

with open(data_dir / "sample_triples.json", encoding="utf-8") as f:
    gold = json.load(f)
with open(data_dir / "alias_map.json", encoding="utf-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 + 指南中的抽取提示词）：

Metric	P	R	F1
Entities	0.80–0.90	0.70–0.85	0.75–0.85
Relations	0.70–0.85	0.55–0.70	0.60–0.75

原文明确这些区间只是参考值，模型非确定性会使实际分数逐次波动；关系召回是最难的数字——抽取器倾向于保守（少抽高置信边而非穷尽）。若提示词调向高召回（如 "extract every stated relationship, even minor ones"），则是以精确率换召回率。

九、生产环境扩展要点

notebook 末尾的 "Scaling up" 一节给出了四条直接可操作的扩展建议：

1. 抽取成本：Haiku 便宜到可以在大语料上跑；当抽取 Schema 与指令固定时，**提示词缓存（prompt caching）**能进一步降本——系统提示与 Schema 走缓存价，只为文档正文付全价。Message Batches API 对可等待 24 小时内的任务给出 50% 折扣。
2. 消解的可扩展性：一次性把一万个 PERSON 实体喂给 Claude 不可行。先做阻塞（blocking）：用廉价信号（同姓、token 重叠、embedding 相似度）分组，让 Claude 只在小块内部仲裁。上文消解提示词在 50～100 个实体的块上无需修改即可工作。
3. 增量更新：新文档到达时，抽取其实体、与现有规范集合（而非彼此）做消解、只加新边；只有当某实体的来源文档集合发生实质变化时才重新摘要。
4. 存储：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（把"索引前先加上下文"的同样思路应用到向量检索）。