使用Claude构建知识图谱：从非结构化文本到多跳推理

本教程面向有编程基础的开发者，讲解如何用 Claude 的结构化输出从非结构化文档中抽取实体与关系，通过 Claude 驱动的实体消解合并表面形式变体，组装内存知识图谱并支持多跳推理。内容涵盖环境准备、抽取与消解代码分析、图组装、实体摘要、子图序列化查询、基于金标准集的精确率/召回率评估，以及生产环境的成本优化、增量更新与存储扩展方案。

技术背景

传统知识图谱构建需要为每个领域训练命名实体识别器、关系分类器和实体消解启发式规则，并在数据变化时持续维护。RAG 检索无法串联跨文档事实，多跳问题（如“谁与参与项目 X 的人一起工作”）需要图遍历。Claude 的结构化输出让每个阶段变成一次提示调用，无需训练数据即可从任意文本抽取类型化实体和主谓宾三元组，并用语义描述进行实体消解，解决字符串相似度无法处理的别名问题。

核心原理

核心原理分四步：1）抽取阶段用 Pydantic 模型定义输出 schema，通过 client.messages.parse() 保证响应通过验证并返回类型化对象，一次调用同时完成 NER 和关系分类；2）消解阶段按实体类型分组，将名称与描述列表交给 Claude 聚类，利用描述区分同名不同实体，生成 alias_to_canonical 映射；3）组装阶段用 MultiDiGraph 存储，节点携带类型、源文档和提及次数，边携带谓词和源文档，方向敏感；4）查询阶段将 N 跳子图序列化为三元组文本，让 Claude 基于边级引用回答多跳问题，答案可追溯。

架构分析

系统采用两模型分工：Haiku 负责高吞吐、受 schema 约束的抽取，Sonnet 负责需要跨文档权衡的消解与摘要。数据流为：Wikipedia REST API 获取摘要 → 逐文档结构化抽取 → 按类型聚类消解 → 别名映射重写关系端点 → NetworkX MultiDiGraph 组装 → 枢纽节点摘要生成 → 子图序列化查询。设计决策包括：用 MultiDiGraph 支持多谓词和方向性；节点保留 source_docs 和 mentions 支持溯源与摘要；消解失败时回退为单元素聚类避免节点丢失；评估时用 alias_map 归一化表面形式。

完整实战教程

你有一堆非结构化文档，需要回答跨越这些文档的问题——"谁与参与项目 X 的人一起工作"、"哪些供应商与这起事件有关"。没有任何单一文档包含答案。RAG 检索不会为你串联事实。你需要一个知识图谱：实体作为节点，类型化关系作为边，这样多跳推理就变成了图遍历。

构建一个知识图谱过去意味着在你的领域上训练命名实体识别器、训练关系分类器、编写实体消解启发式规则，并在数据变化时维护这三者。有了 Claude，每个阶段都变成了一个提示。

你将学到什么

到本指南结束时，你将能够：

一切都在内存中运行，无需数据库。当你需要扩展时，这些技术可直接迁移到 Neo4j、Neptune 或 Postgres 邻接表。

先决条件

设置

%%capture
%pip install anthropic requests networkx matplotlib python-dotenv pydantic

import json
from collections import defaultdict
from pathlib import Path
from typing import Literal
from urllib.parse import quote
import anthropic
import matplotlib.pyplot as plt
import networkx as nx
import requests
from dotenv import load_dotenv
from pydantic import BaseModel
load_dotenv()
client = anthropic.Anthropic()
EXTRACTION_MODEL = "claude-haiku-4-5"
SYNTHESIS_MODEL = "claude-sonnet-4-6"

我们使用两个模型。Haiku 负责高吞吐量、受模式约束的提取工作，在这类任务中，速度和成本比细微差别更重要。Sonnet 负责实体解析和摘要，在这类任务中，模型需要权衡跨文档的相互矛盾的证据。

构建语料库

我们需要一些讨论重叠实体的文档，这样实体解析才有真正的工作要做。阿波罗计划是一个很好的试验台：六篇简短的维基百科摘要，它们都提到了 NASA、月球、几位宇航员以及一种运载火箭——但每篇文章对它们的命名略有不同。

我们从 Wikipedia REST API 获取摘要，而不是完整文章，以降低 token 成本。对于生产级流水线，你会对完整文档进行分块；提取逻辑是相同的。

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")
print(f"Loaded {len(documents)} documents\n")
print(f"Sample — {documents[0]['title']}:\n{documents[0]['text'][:300]}...")

实体与关系抽取

经典 NER 为文本片段打上标签（PERSON、ORG、LOC）。经典关系抽取随后将片段对分类为关系类型。两者传统上都需要每个领域的标注训练数据。

我们将这两个阶段合并为对每个文档的一次 Claude 调用。关键在于结构化输出：我们将输出形状定义为 Pydantic 模型，并将其传递给 client.messages.parse()。Claude 的响应保证能通过该 schema 的验证，并以类型化的 Python 对象返回——无需正则解析，无需 JSON 解码错误，无需防御性的 isinstance 检查。

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]
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

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"]})
    print(
        f"{doc['title']:<25} {len(result.entities):>3} entities  {len(result.relations):>3} relations"
    )
print(f"\nTotal: {len(raw_entities)} raw entities, {len(raw_relations)} raw relations")

让我们看看提取出了什么。注意同一个现实世界实体在不同文档中如何以不同的表面形式出现——这就是我们接下来要解决的实体消解问题。

by_type = defaultdict(list)
for e in raw_entities:
    by_type[e["type"]].append(e["name"])
for etype, names in sorted(by_type.items()):
    print(f"{etype} ({len(names)}):")
    for name in sorted(set(names)):
        print(f"  {name}")
    print()

实体消解

原始抽取结果中会出现重叠的提及项：“NASA”与“National Aeronautics and Space Administration”、“Neil Armstrong”与“Armstrong”，可能还有“the Moon”与“Moon”。如果我们直接基于这些结果构建图谱，就会得到一个支离破碎的烂摊子——同一个概念被拆散在互不连通的节点上。

传统方法使用字符串相似度（编辑距离、基于 token 的 Jaccard 相似度）加上分块规则。这对拼写错误有效，但在“Edwin Aldrin”与“Buzz Aldrin”上就会失效——这两个名字在字符层面毫无重叠，却指向同一个人。

我们改为让 Claude 对每种类型的实体进行聚类，利用抽取阶段生成的一行描述作为消歧上下文。这些描述至关重要：“Armstrong — 第一个在月球上行走的人”和“Armstrong — 爵士小号手”名字相同，但不应该合并。

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

有两种失败模式需要注意。第一，Claude 从每个聚类中遗漏的任何原始名称都会从图中悄然消失，因为 alias_to_canonical 中没有它的条目——生产级解析器应当为未匹配的名称回退到单元素聚类，以免丢失任何内容。第二，解析器可能会过度合并：像 "Gemini 12" 这样的具体任务可能会因为描述重叠而被并入更宽泛的 "Project Gemini"。前者会丢失节点，后者会损失精度。两者都值得在下面的输出中进行抽查。

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
before = len({e["name"] for e in raw_entities})
after = len(canonical_info)
print(f"Entity resolution: {before} unique names → {after} canonical entities\n")
for canonical, info in sorted(canonical_info.items()):
    aliases = [a for a in info["aliases"] if a != canonical]
    alias_str = f"  (also: {', '.join(aliases)})" if aliases else ""
    print(f"{info['type']:<14} {canonical}{alias_str}")

组装图

有了干净的别名映射后，我们将每个关系端点重写为其规范形式，并将结果加载到 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"]))
print(f"Graph: {G.number_of_nodes()} nodes, {G.number_of_edges()} edges")
print(f"Connected components: {nx.number_weakly_connected_components(G)}")
print("\nMost connected entities:")
for node, deg in sorted(G.degree(), key=lambda x: -x[1])[:5]:
    print(f"  {node:<35} degree {deg:>2}  ({G.nodes[node]['type']})")

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)
handles = [
    plt.Line2D([0], [0], marker="o", color="w", markerfacecolor=c, markersize=10, label=t)
    for t, c in COLOR.items()
    if any(G.nodes[n]["type"] == t for n in G.nodes)
]
plt.legend(handles=handles, loc="upper left")
plt.title("Apollo Program Knowledge Graph")
plt.axis("off")
plt.tight_layout()
plt.show()

Notebook output

节点大小随度数缩放——枢纽节点是将语料库连接在一起的实体。颜色编码类型：如果你的图大部分是一种颜色，说明你的语料库范围狭窄；良好的混合意味着抽取器正在发现完整的人物、地点和事物阵容。单一连通分量意味着实体解析发挥了作用——碎片化的孤岛将表明本应合并但未能合并的变体。

实体摘要

目前每个节点只携带了来自最先提及它的文档的一行描述。对于枢纽节点——即在许多文档中出现的节点——我们可以做得更好：汇集所有提及，将图邻域作为上下文加入，并让 Claude 合成一份完整的画像。

这一步将标签图转变为知识图。摘要成为你在搜索结果中呈现或馈送给下游问答的节点内容。

class TimeRange(BaseModel):
    start: str  # YYYY or YYYY-MM, or "unknown"
    end: str  # YYYY or YYYY-MM, or "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:
    # Reads module-level G and documents built earlier in the notebook.
    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

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()
    print(f"═══ {node} ═══")
    print(profile.summary)
    print(f"\nTime range: {profile.time_range.start} – {profile.time_range.end}")
    print("Key facts:")
    for fact in profile.key_facts:
        print(f"  • {fact}")
    print()

查询图

构建知识图谱的回报在于多跳推理：回答那些需要串联从未在同一文档中共现的事实的问题。“哪些地点与乘坐阿波罗 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])
print(f"Querying 2-hop neighborhood of: {center}\n")
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("\n" + "─" * 60 + "\n")
print("WITH graph context:")
print(ask(question, subgraph))

未接地的回答依赖于 Claude 的预训练知识，可能是正确的——阿波罗 11 号确实著名。但接地的回答是可追溯的：每一项声明都引用了我们从特定文档中提取的一条边。在 Claude 没有先验知识的私有语料库上，只有接地的回答才能起作用。

评估

知识图谱的质量通过针对黄金标准集的精确率和召回率来衡量。我们在 data/sample_triples.json 中提供了一个小型的人工标注集，覆盖其中两篇文章，另外还有 data/alias_map.json，它将表面形式变体归一化为黄金标准名称，使得“the Moon”和“Moon”被视为同一次命中。

下面的检查并排评估两件事：原始提取器输出，以及经过解析步骤中 alias_to_canonical 处理后的相同实体。当解析器选择了一个别名映射已知的规范形式（或与黄金标准逐字匹配的形式）时，解析后的召回率会上升。当它选择了一个别名映射未覆盖的冗长形式——比如“Neil Alden Armstrong”——解析后的召回率可能会*下降*，因为一个在解析前匹配黄金标准的名称在解析后不再匹配。这不是解析器的缺陷；这是评分上的假象。解决办法是，每当你看到一个评分器不认识的规范形式时，就扩展 alias_map.json。

这个单元格只对实体进行评分。独立脚本还会对关系进行评分，匹配 (source, target) 对，忽略谓词措辞——因此其关系召回率是一个上界。从仓库根目录运行它：

uv run python capabilities/knowledge_graph/evaluation/eval_extraction.py

# Expects the kernel launched from this notebook's directory
# (capabilities/knowledge_graph/). Falls back to repo root.
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)
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
print("Raw extraction vs resolved-graph recall against gold:\n")
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))}")

扩展规模

本 notebook 在内存中处理了六份文档。而生产环境中的知识图谱是由数千份文档构建的。关于扩展，有几点说明：

抽取成本。 Haiku 足够便宜，可以在大规模语料上运行，但当你的抽取 schema 和指令保持固定时，提示缓存能进一步降低成本——缓存系统提示和 schema，只为文档文本支付全价。Message Batches API 对于可以等待最多 24 小时的任务提供 50% 的折扣。

大规模实体消解。 在一个提示中把一万个 PERSON 实体喂给 Claude 是行不通的。先分块：按廉价信号（相同姓氏、重叠词元、嵌入相似度）对候选实体分组，这样 Claude 只需在小块内进行仲裁。上面的消解提示在 50–100 个实体的块上无需修改即可使用。

增量更新。 当新文档到达时，抽取其实体，将它们与现有的规范实体集进行消解（而不是彼此之间消解），并且只添加新的边。仅当某个实体的源文档集合发生实质性变化时，才重新生成该实体的摘要。

存储。 NetworkX 在几十万条边以内表现良好。超出之后，该 schema 可以直接映射到属性图（Neo4j、Neptune）或三张 Postgres 表：entities(id, name, type, summary)、relations(source_id, target_id, predicate)、aliases(entity_id, alias)。抽取和消解代码无需更改——只有持久化层需要改变。

总结

你仅用提示词就构建了一个完整的知识图谱流水线：

evaluation/ 中的评估工具为你提供了反馈循环：修改抽取提示词，重新运行评分器，观察 F1 的变化。正是这个循环把演示变成了生产系统。

相关 cookbook

常见问题

常见错误：1）抽取时 API 报错导致文档跳过，需捕获 anthropic.APIError 并记录；2）消解遗漏原始名称导致节点静默消失，生产环境应为未匹配名称回退单元素聚类；3）消解过度合并，如具体任务被并入宽泛项目，需抽查输出；4）查询时无文本块，需检查 stop_reason 并处理；5）评估中解析后召回率下降，通常是规范形式未被 alias_map 覆盖，应扩展别名映射；6）网络问题导致 Wikipedia API 失败，需检查 User-Agent 和超时设置。

企业应用扩展

生产扩展方向：1）成本优化，使用提示缓存缓存系统提示和 schema，用 Message Batches API 获得 50% 折扣；2）大规模消解，先按姓氏、词元重叠或嵌入相似度分块，再让 Claude 在 50-100 个实体的块内仲裁；3）增量更新，新文档抽取后与现有规范实体集消解，仅当源文档集合实质变化时重新生成摘要；4）存储迁移，NetworkX 支持几十万条边，超出后映射到 Neo4j、Neptune 或 Postgres 三表结构，抽取和消解代码无需更改；5）评估反馈循环，修改抽取提示后重新运行评分器观察 F1 变化。

总结

本教程面向有编程基础的开发者，讲解如何用 Claude 的结构化输出从非结构化文档中抽取实体与关系，通过 Claude 驱动的实体消解合并表面形式变体，组装内存知识图谱并支持多跳推理。内容涵盖环境准备、抽取与消解代码分析、图组装、实体摘要、子图序列化查询、基于金标准集的精确率/召回率评估，以及生产环境的成本优化、增量更新与存储扩展方案。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部API 与工程实践教程