How to Implement Knowledge Graph Construction from Unstructured Text with Claude

Build a typed knowledge graph from free-form documents using Claude's structured output capabilities and a four-stage Python pipeline.

The anthropics/claude-cookbooks repository provides an end-to-end implementation for extracting entities and relations from unstructured text and assembling them into a queryable knowledge graph. This guide explains the complete pipeline—from raw document ingestion to canonical graph construction—based on the reference implementation in capabilities/knowledge_graph/guide.ipynb.

Stage 1: Corpus Preparation

Begin by loading or fetching raw text documents. The reference notebook demonstrates this with Wikipedia summaries to minimize token costs during development. Each document serves as input for the extraction stage.

from urllib.parse import quote
import requests

def fetch_summary(title: str) -> str:
    """Fetch Wikipedia summary as sample unstructured text."""
    slug = quote(title.replace(" ", "_"), safe="")
    r = requests.get(
        "https://en.wikipedia.org/api/rest/v1/page/summary/" + slug,
        headers={"User-Agent": "claude-cookbooks/1.0"},
        timeout=10,
    )
    r.raise_for_status()
    return r.json()["extract"]

Stage 2: Structured Extraction with Pydantic

The extraction stage uses Claude's client.messages.parse() method to coerce output into strict Pydantic models. This eliminates fragile JSON parsing and guarantees type-safe entities and relations.

Define the schema with typed literals for entity categories:

from pydantic import BaseModel
from typing import Literal

class Entity(BaseModel):
    name: str
    type: Literal["PERSON", "ORGANIZATION", "LOCATION", "EVENT", "ARTIFACT"]
    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:
- Only central entities.
- One-sentence description per entity.
- Short verb phrase predicates.
- Every relation must link two extracted entities."""

The extract() function in capabilities/knowledge_graph/guide.ipynb handles the API call:

def extract(text: str) -> ExtractedGraph:
    resp = client.messages.parse(
        model="claude-haiku-4-5",
        max_tokens=2048,
        messages=[{"role": "user", "content": EXTRACTION_PROMPT.format(text=text)}],
        output_format=ExtractedGraph,
    )
    return resp.parsed_output

Stage 3: Entity Resolution and Canonicalization

Raw extraction produces surface-form variations of the same real-world entity (e.g., "Elon Musk" and "Musk"). The resolver clusters these variants using Claude with a second prompt that consumes the descriptions generated in Stage 2.

The resolve() function and supporting Pydantic models appear in capabilities/knowledge_graph/guide.ipynb:

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.
<entities>
{entity_list}
</entities>

Cluster them. Each input name appears in exactly one cluster's aliases list.
Use the descriptions to avoid merging distinct entities. The canonical name should be the most complete."""

def resolve(entity_type: str, entities: list[dict]) -> list[Cluster]:
    # Build unique list of "name: description" pairs

    uniq = {}
    for e in entities:
        uniq.setdefault(e["name"], e["description"])
    entity_list = "\n".join(f"- {n}: {d}" for n, d in uniq.items())
    
    resp = client.messages.parse(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": RESOLVE_PROMPT.format(
            entity_type=entity_type, entity_list=entity_list)}],
        output_format=ResolvedClusters,
    )
    return resp.parsed_output.clusters

The resulting alias-to-canonical mapping is persisted as data/alias_map.json in the repository.

Stage 4: Graph Assembly with NetworkX

Populate a networkx.MultiDiGraph using the canonical entities as nodes and resolved triples as directed edges. The graph stores provenance metadata including source documents and mention counts.

import networkx as nx

G = nx.MultiDiGraph()

# Assume raw_entities, raw_relations, alias_to_canonical, 

# and canonical_info are available from previous stages

for e in raw_entities:
    canon = alias_to_canonical.get(e["name"])
    if not canon:
        continue
    if canon not in G:
        G.add_node(
            canon,
            type=canonical_info[canon]["type"],
            description=e["description"],
            source_docs=[],
            mentions=0,
        )
    G.nodes[canon]["source_docs"].append(e["source_doc"])
    G.nodes[canon]["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"])

Optional Enrichment: Entity Summarization

For high-degree nodes, generate multi-paragraph profiles by prompting Claude with all neighboring edges and extracted excerpts. The summarize_entity() helper in capabilities/knowledge_graph/guide.ipynb synthesizes structured profiles containing type, time range, and key facts.

Querying the Knowledge Graph

Answer multi-hop questions by serializing subgraphs as plain triples and feeding them back to Claude. This demonstrates the core retrieval advantage of graph structures over vector embeddings alone.

The serialize_subgraph() and ask() functions from capabilities/knowledge_graph/guide.ipynb:

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:
        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
    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=500,
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.content[0].text

Evaluation and Quality Assurance

Measure extraction precision and recall against the hand-labeled gold set in capabilities/knowledge_graph/data/sample_triples.json. The script capabilities/knowledge_graph/evaluation/eval_extraction.py automates scoring for both raw extraction and resolved graph outputs.

Scaling Considerations

For production workloads exceeding notebook scale:

• Batch Processing: Use the Anthropic Batch API and prompt caching for cost-effective bulk extraction.
• Blocking Strategy: Chunk entity resolution into blocks of 50-100 items after cheap pre-blocking via last-name matching or embedding similarity.
• Persistence: Migrate from NetworkX to property-graph databases (Neo4j, Amazon Neptune) when edge counts exceed millions.

Summary

• Structured extraction uses client.messages.parse() with Pydantic models to guarantee valid entities and relations without manual JSON parsing.
• Entity resolution clusters surface-form variants into canonical nodes using Claude's reasoning over entity descriptions.
• Graph assembly leverages networkx.MultiDiGraph to store typed nodes and directed edges with full provenance tracking.
• Quality assurance relies on eval_extraction.py and the gold standard in data/sample_triples.json to quantify pipeline performance.
• Querying serializes local subgraphs to ground Claude's answers with explicit edge citations, enabling multi-hop reasoning.

Frequently Asked Questions

What Pydantic models are required for the extraction pipeline?

You need Entity (with name, type, and description fields), Relation (with source, predicate, and target), and ExtractedGraph (containing lists of both). These models are defined in capabilities/knowledge_graph/guide.ipynb and passed to client.messages.parse() as the output_format parameter to ensure type-safe responses.

How does entity resolution handle ambiguous names like "Apple"?

The resolution prompt in capabilities/knowledge_graph/guide.ipynb supplies Claude with one-line descriptions extracted in Stage 2 alongside each entity name. By reasoning over these descriptions, Claude distinguishes between "Apple" (the organization) and "Apple" (the artifact), clustering them into separate canonical nodes rather than merging distinct entities.

Can this pipeline scale to thousands of documents?

Yes. The repository outlines scaling tactics including the Anthropic Batch API for cheap bulk extraction, prompt caching for repeated entity types, and pre-blocking strategies (last-name matching or embeddings) to reduce the resolution search space. For graphs exceeding millions of edges, migrate the NetworkX structure to Neo4j or Amazon Neptune.

Where is the evaluation data located?

The hand-labeled gold standard for calculating precision and recall resides in capabilities/knowledge_graph/data/sample_triples.json. The evaluation script capabilities/knowledge_graph/evaluation/eval_extraction.py compares model outputs against this dataset to score both entity extraction and relation classification accuracy.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}