knowledge-graph-construction

What this skill does

# 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.

```python
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

```python
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

```python
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.

```python
# 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.

```python
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](https://github.com/facebookresearch/BLINK), [GENRE](https://github.com/facebookresearch/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

```python
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)

```python
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.

```python
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

```python
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:

```cypher
CREATE CONSTRAINT entity_id IF NOT EXISTS FOR (e:Entity) REQUIRE e.canonical_id IS UNIQUE;
C