Concepts¶

Everything on this page is importable from openodke and runs offline. The data types are frozen pydantic models: a stage returns new objects rather than editing one (DECISIONS #4).

Fact: an edge or a property¶

A Fact is one (subject, predicate, object) claim with its evidence. object_entity, another Entity, makes it an edge; object_value, a literal, makes it a property. Set exactly one (DECISIONS #2).

Evidence¶

Span.resolve(doc) returns the text at the offsets, and Span.is_faithful(doc) checks the quote against it, for free (DECISIONS #3). By default LLMGrounder reads the span alone, so a span should be the clause that states the claim (DECISIONS #23). A context span is not a citation (DECISIONS #25).

from openodke import Document, Evidence, Span, SpanOrigin

doc = Document(id="d1", text="Acme was founded in 1999. It sells anvils.")
start = doc.text.index("anvils")
span = Span(doc_id="d1", start=start, end=start + 6, quote="anvils")

assert span.resolve(doc) == "anvils" and span.is_faithful(doc)
assert not Span(doc_id="d1", start=0, end=4, quote="Acme Corp").is_faithful(doc)
assert Evidence(doc_id="d1", span=span).span_origin is SpanOrigin.CITED

Verdicts¶

The grounder stamps a GroundingVerdict on each fact and never drops one (DECISIONS #20):

The gate decides what is written. Its GateVerdict is accept or conflict, both written, or refuse, dropped and counted. VerdictGate refuses contradicted, and not_found too with refuse_not_found=True (Grounding).

The signature¶

Fact.signature is what two extractions must share to be one claim. The corroborator merges on it, the Neo4j sink MERGEs on it, and the extraction evaluator matches on it.

The ontology marks a qualifier with Qualifier(identity=True), and the extractor stamps its name onto Fact.identity_keys (Ontology).

from openodke import Entity, Fact, Polarity

acme = Entity(key="c:acme", type="Company", label="Acme")
p50 = Fact(
    subject=acme,
    predicate="uptime",
    object_value="99.9%",
    qualifiers={"percentile": "p50", "as_of": "2026-08"},
    identity_keys=("percentile",),
)
p95 = p50.model_copy(update={"qualifiers": {"percentile": "p95", "as_of": "2026-08"}})
restated = p50.model_copy(update={"qualifiers": {"percentile": "p50", "as_of": "2026-09"}})
denied = p50.model_copy(update={"polarity": Polarity.DENIED})

assert p50.signature != p95.signature  # identity-bearing: two measurements
assert p50.signature == restated.signature  # reconcilable: one claim, told twice
assert p50.signature != denied.signature  # a denial is the opposite claim

print(p50.signature)
# ('c:acme', 'Company', 'uptime', "'99.9%'", 'asserted', (('percentile', "'p50'"),))

Entities and links¶

An Entity is a node: a key, which facts merge on, a type, a label, aliases, external_id, attributes, and a resolution saying how the key was decided: caller, external_id, or linker with a score (DECISIONS #18).

A resolver proposes EntityLinks between keys. It never merges or replaces a stored node (DECISIONS #16). Only a proof re-keys incoming facts onto a stored entity. Asked to normalise a batch, it makes the batch's own look-alikes that nothing keeps apart one entity, before anything is written (DECISIONS #43).

from openodke import EntityLink, KnowledgeGraph, LinkKind

acme_gmbh = Entity(key="c:acme-gmbh", type="Company", label="Acme GmbH", external_id="DE-114322")
different = EntityLink(
    source_key="c:acme",
    target_key="c:acme-gmbh",
    kind=LinkKind.DIFFERENT,
    reason="external_id mismatch: DE-114322 vs GB-889401",
)
graph = KnowledgeGraph(entities=(acme, acme_gmbh), facts=(p50,), links=(different,))
assert (len(graph.edges), len(graph.properties), len(graph.links)) == (0, 1, 1)

KnowledgeGraph is the pipeline's output, and all a sink reads: entities, facts, links and stats.

The order the stages run in¶

Pipeline(ontology, extractor, *, retriever=, chunker=, router=, grounder=,
         normalizer=, resolver=, corroborator=, scorer=, gate=, constrainer=,
         inverses=, sinks=, coverage=, reextract=)

A stage left as None is its pass-through. openodke.Validator runs this pipeline over triples, with a default for every stage after extraction (The Validator). run(docs):

1. Chunk, route, extract, per chunk; or triples in, through TriplesExtractor.
2. Ground, per document: the free span check, then the model. The span locator (locate=True) and widen-and-retry (widen=True) are optional.
3. Coverage (coverage=True) and re-extract (reextract=Reextract()), both optional: what extraction left behind, handed back to the extractor.
4. Normalise.
5. Resolve, over the batch. It comes before corroboration because the signature merges on the subject's key.
6. Inverse partners, when the ontology declares them.
7. Corroborate, then score.
8. Gate.
9. Write to every sink.

A stage with extract_many, ground_documents or ground_many gets the whole run at once and may call its model concurrently, with the same result. Pipeline.constraints() returns the DDL for a sink to apply; run() does not. Chunking and routing are on Loading documents.

Inverse and symmetric partners¶

When the ontology declares an inverse or a symmetric predicate, each edge on it gains its partner, the same claim the other way round (DECISIONS #28). inverses=False turns it off.

• The partner cites its source's evidence and keeps its verdict. It is never grounded again, and a gate that refuses the source on its verdict refuses the partner too.
• It carries qualifiers["odke.derived"]: the rule, and the source's signature, which openodke.corroborate.derived_from(fact) reads back.
• A partner the batch already states is not derived, and a derived fact gets no partner.

from openodke import Chunk, Ontology, Pipeline

geo = Ontology.from_dict(
    {
        "types": {"Place": {}},
        "predicates": {
            "located_in": {"domain": ["Place"], "range": "Place", "inverse_of": "contains"},
            "contains": {"domain": ["Place"], "range": "Place"},
        },
    }
)


class SaysLocated:
    def extract(self, chunk: Chunk, ontology: Ontology) -> list[Fact]:
        brittany = Entity(key="brittany", type="Place", label="Brittany")
        france = Entity(key="france", type="Place", label="France")
        receipt = (Evidence(doc_id=chunk.doc_id),)
        return [
            Fact(subject=brittany, predicate="located_in", object_entity=france, evidence=receipt)
        ]


kg = Pipeline(geo, SaysLocated()).run([Document(id="geo", text="Brittany is in France.")])
print([(f.subject.label, f.predicate, f.object_entity.label) for f in kg.facts])
# [('Brittany', 'located_in', 'France'), ('France', 'contains', 'Brittany')]
print(kg.stats)
# {'documents': 1, 'chunks': 1, 'skipped': 0, 'deferred': 0, 'empty_extractions': 0, 'derived': 1, 'refused': 0}

The Protocols¶

Every stage is a typing.Protocol in openodke.stages: any object with the right method is a stage (DECISIONS #5). A pass-through returns its input unchanged unless the table says otherwise.

Initiator and Retriever, the paper's two front stages, are declared too, and optional (DECISIONS #9).

PlatformProfile and Delegated¶

Some stores do a stage themselves: neo4j-graphrag resolves after the write. A sink says so with profile = PlatformProfile(name, resolves=, constrains=, prunes=). If openodke also runs that stage, Pipeline warns once with a DoubleStageWarning and runs both. Delegated(to="...") hands the stage to the platform: a pass-through that records to in Entity.resolution.linker for a resolver, or in the verdict's reason for a router or gate (DECISIONS #21).

import warnings

from openodke import Delegated, DoubleStageWarning, PlatformProfile, Resolution


class GraphRagSink:
    profile = PlatformProfile(name="neo4j-graphrag", resolves=True)

    def write(self, kg):
        self.written = kg


class ExactKeyResolver:
    def resolve(self, facts, index):
        return facts, ()


with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")
    Pipeline(geo, SaysLocated(), resolver=ExactKeyResolver(), sinks=[GraphRagSink()])
assert [w.category for w in caught] == [DoubleStageWarning]

resolver = Delegated(to="neo4j-graphrag:FuzzyMatchResolver")
kg = Pipeline(geo, SaysLocated(), resolver=resolver, sinks=[GraphRagSink()]).run(
    [Document(id="geo", text="Brittany is in France.")]
)
assert kg.facts[0].subject.resolution == Resolution(
    method="linker", linker="neo4j-graphrag:FuzzyMatchResolver"
)