Knowledge Graph Construction

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

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