Knowledge Graph Construction
Turning unstructured documents into a queryable graph has four stages: extract, link, validate, load. Skip any one and you get a graph that's either empty, hallucinated, inconsistent, or unusable.
Pipeline Overview
chunks → extract (entities + triples) → link (canonical IDs / KB entries)
→ validate (schema, LLM judge) → load (Neo4j / Neptune / TigerGraph)
→ incremental updates on new ingests
Schema First
Define the ontology up front. Free-form extraction produces noise ("rel": "is_related_to") that kills downstream querying.
from pydantic import BaseModel, Field
from typing import Literal
EntityType = Literal["Person", "Organization", "Product", "Location", "Incident", "Document"]
RelationType = Literal[
"WORKS_AT", "FOUNDED", "HEADQUARTERED_IN", "PRODUCED_BY",
"CAUSED", "MENTIONED_IN", "REPORTS_TO", "ACQUIRED",
]
class Entity(BaseModel):
name: str = Field(..., description="Canonical name as it appears in text")
type: EntityType
description: str = Field("", description="1-line grounded description")
class Triple(BaseModel):
subject: str
predicate: RelationType
object: str
evidence: str = Field(..., description="Exact quote from the source")
class Extraction(BaseModel):
entities: list[Entity]
triples: list[Triple]
Every relation type should have a clear, narrow intent; prefer many specific types over a few broad ones.
LLM-Based Extraction (Structured Output)
Anthropic tools API
import anthropic, json
client = anthropic.Anthropic()
tool = {
"name": "emit_extraction",
"description": "Return the extraction.",
"input_schema": Extraction.model_json_schema(),
}
PROMPT = """Extract entities and triples from the text.
Rules:
- Use only the schema types in the tool.
- Only extract facts EXPLICITLY stated in the text.
- Normalize names (no pronouns, no possessives).
- `evidence` must be a verbatim quote.
Text:
{chunk}"""
def extract(chunk: str) -> Extraction:
msg = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=4000,
tools=[tool],
tool_choice={"type": "tool", "name": "emit_extraction"},
messages=[{"role": "user", "content": PROMPT.format(chunk=chunk)}],
)
for block in msg.content:
if block.type == "tool_use":
return Extraction.model_validate(block.input)
raise RuntimeError("no extraction returned")
OpenAI structured outputs
from openai import OpenAI
oai = OpenAI()
resp = oai.beta.chat.completions.parse(
model="gpt-4o",
response_format=Extraction,
messages=[{"role": "user", "content": PROMPT.format(chunk=chunk)}],
)
extraction: Extraction = resp.choices[0].message.parsed
Costs
- Use a cheaper model for extraction (
claude-haiku-4-5,gpt-4o-mini) on high-volume ingestion. - Batch the call with the OpenAI/Anthropic batch APIs for 50% discount on large corpora.
- Cache by
sha256(chunk + model_version + schema_version)— re-run only on schema changes.
REBEL (Encoder-Decoder Model)
REBEL is a seq-to-seq model fine-tuned to produce triples directly; no prompt engineering, no API cost.
# pip install transformers sentencepiece
from transformers import pipeline
pipe = pipeline("text2text-generation", model="Babelscape/rebel-large", device=0)
out = pipe("Barack Obama was born in Honolulu and served as the 44th US president.",
return_tensors=False, return_text=True, max_length=256)[0]["generated_text"]
# Parser for REBEL output format
def parse_rebel(raw: str) -> list[tuple[str, str, str]]:
triples, subject, relation, obj, state = [], "", "", "", "x"
for tok in raw.replace("<s>","").replace("<pad>","").replace("</s>","").split():
if tok == "<triplet>":
if subject and obj: triples.append((subject.strip(), relation.strip(), obj.strip()))
state, subject, relation, obj = "t", "", "", ""
elif tok == "<subj>":
state = "s"
elif tok == "<obj>":
state = "o"
else:
if state == "t": subject += " " + tok
if state == "s": obj += " " + tok
if state == "o": relation += " " + tok
if subject and obj: triples.append((subject.strip(), relation.strip(), obj.strip()))
return triples
Use REBEL when: cost matters more than schema control (REBEL produces free-form relations; map to your schema with a rule table).
OpenIE (Stanford CoreNLP, OpenIE6)
Rule-based open information extraction produces high-recall but noisy triples. Pair with a downstream LLM filter to prune low-quality ones. Useful for multilingual corpora lacking strong instruction-tuned models.
Entity Linking to Wikidata / DBpedia
Raw surface forms aren't identifiers. Link each extracted entity to an external KB ID where possible so cross-document merges are trivial.
import requests
def wikidata_search(name: str, lang="en", limit=5):
r = requests.get("https://www.wikidata.org/w/api.php", params={
"action": "wbsearchentities", "search": name, "language": lang,
"format": "json", "limit": limit,
}, timeout=5)
return [(h["id"], h.get("description", "")) for h in r.json().get("search", [])]
# wikidata_search("Anthropic")
# [('Q110760624', 'American artificial intelligence safety and research company'), ...]
For ambiguous names, pass candidates to an LLM with surrounding context and let it pick the right QID. Cache the linking decision per (name, context_hash).
Alternatives: BLINK, GENRE, spaCy's entity_linker pipe with a custom KB.
Validation
Never load unvalidated triples. Three layers:
1. Schema validation (Pydantic already does this)
2. Evidence-grounded check
def grounded(triple: Triple, chunk: str) -> bool:
return triple.evidence.lower() in chunk.lower() \
and triple.subject.lower() in triple.evidence.lower() \
and triple.object.lower() in triple.evidence.lower()
3. LLM judge (spot-check + reject)
JUDGE_PROMPT = """Does the evidence support the triple? Reply JSON:
{{"supported": true|false, "confidence": 0.0-1.0}}
Triple: ({s}, {p}, {o})
Evidence: {e}"""
def judge(t: Triple) -> dict:
msg = client.messages.create(model="claude-haiku-4-5", max_tokens=150,
messages=[{"role": "user", "content": JUDGE_PROMPT.format(
s=t.subject, p=t.predicate, o=t.object, e=t.evidence)}])
return json.loads(msg.content[0].text)
Run the judge on a 10% sample; alert if supported drops below 90%.
Provenance
Every triple carries source metadata — essential for citations and for fixing bad extractions later.
record = {
"triple": (t.subject, t.predicate, t.object),
"evidence": t.evidence,
"source_doc_id": doc_id,
"source_chunk_id": chunk_id,
"source_offset": chunk_offset,
"extractor": "claude-sonnet-4-5",
"schema_version": "v3",
"extracted_at": iso_now(),
}
Loading to Graph Stores
Neo4j
from neo4j import GraphDatabase
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "pass"))
CYPHER = """
UNWIND $rows AS row
MERGE (s:Entity {canonical_id: row.s_id})
ON CREATE SET s.name = row.s_name, s.type = row.s_type
MERGE (o:Entity {canonical_id: row.o_id})
ON CREATE SET o.name = row.o_name, o.type = row.o_type
MERGE (s)-[r:REL {type: row.predicate}]->(o)
SET r.evidence = row.evidence,
r.source_chunk_id = row.source_chunk_id,
r.extracted_at = datetime(row.extracted_at)
"""
def load(rows: list[dict]):
with driver.session() as s:
for i in range(0, len(rows), 500):
s.run(CYPHER, rows=rows[i:i+500])
Create constraints first:
CREATE CONSTRAINT entity_id IF NOT EXISTS FOR (e:Entity) REQUIRE e.canonical_id IS UNIQUE;
CREATE INDEX entity_name IF NOT EXISTS FOR (e:Entity) ON (e.name);
Amazon Neptune
Use Gremlin or openCypher. Bulk-load via CSV to S3 → neptune-loader. Same shape; predicates become edge labels.
TigerGraph
Define VERTEX and EDGE types in GSQL. Bulk-load with LOADING JOB. Strongest for >100M edge graphs.
Incremental Updates
On new ingest: extract → link → validate → diff against existing triples (keyed by (s, p, o)), insert new, update evidence on existing. Never delete — mark superseded for lineage.
CREATE TABLE triple_log (
s_id TEXT, predicate TEXT, o_id TEXT,
evidence TEXT,
source_chunk_id TEXT,
extracted_at TIMESTAMPTZ,
superseded_at TIMESTAMPTZ
);
CREATE INDEX ON triple_log (s_id, predicate, o_id);
Full Pipeline Skeleton
def build_kg(chunks: list[dict]):
all_rows = []
for chunk in chunks:
extraction = extract(chunk["text"])
ents = [link_entity(e) for e in extraction.entities]
for t in extraction.triples:
if not grounded(t, chunk["text"]):
continue
s_id = ent_id_by_name[t.subject]
o_id = ent_id_by_name[t.object]
all_rows.append({
"s_id": s_id, "s_name": t.subject, "s_type": ...,
"o_id": o_id, "o_name": t.object, "o_type": ...,
"predicate": t.predicate,
"evidence": t.evidence,
"source_chunk_id": chunk["id"],
"extracted_at": iso_now(),
})
load(all_rows)
Anti-Patterns
| Anti-Pattern | Fix |
|---|---|
| Open-ended extraction (no schema) | Schema-guided via Pydantic + structured outputs |
| Accepting triples without evidence | Require evidence field + grounding check |
Using gpt-4o / claude-opus for every chunk | Haiku/mini for bulk; Opus for hard cases |
| Extracting entities but not linking to canonical IDs | Wikidata link + internal entity-resolution step |
| Full graph rebuild on every ingest | Incremental triple insert, keyed on (s, p, o) |
| Dropping provenance to "save space" | Provenance is the citation layer — keep it |
| Loading without constraints | Define uniqueness on canonical_id first |
Production Checklist
- Schema (entities + relations) in Pydantic, version-tagged
- Extraction cached by
sha256(chunk + model + schema_version) - Grounding check (evidence quote in chunk) enforced
- LLM judge sample (10%) with alert on quality regression
- Entity linking to canonical ID (internal or Wikidata)
- Provenance fields on every triple (source, offset, extractor, time)
- Graph constraints + indexes created before bulk load
- Batch load (500–1000 rows per transaction)
- Incremental updates on ingest, no full rebuilds
- Eval corpus with gold triples for regression testing