Knowledge Graphs

A knowledge graph stores your documents as facts — Jane Doe -[WORKS_AT]-> Acme Capital — instead of text chunks. Where vector search answers "what does this sound like", a graph answers "how are these connected". Dynamiq builds one with two nodes from dynamiq.nodes.knowledge_graphs: KnowledgeGraphEntityExtractor runs the LLM extraction, and KnowledgeGraphWriter assigns durable entity identity and upserts into the graph store.

documents ──► KnowledgeGraphEntityExtractor ──► KnowledgeGraphWriter ──► graph store
              (LLM extraction + ontology)        (entity resolution + upsert)

The split is deliberate: extraction is the slow, parallelizable part, while the writer must stay a single serial node — two writers racing on the same graph would each mint fresh ids for the same name and create duplicates.

The ontology

An Ontology is required — there is no free-form mode. It is enforced twice: the allowed types are injected into the extraction prompt (and sent as an enum in the LLM's structured-output schema), and the extracted graph is then hard-filtered so anything off-ontology is dropped before it reaches the store.

from dynamiq.nodes.knowledge_graphs import Ontology, Triple

ONTOLOGY = Ontology(
    entity_types=["Person", "Organization", "System", "Event", "Location"],
    relationship_types=["WORKS_AT", "USES", "PRESENTED", "PRESENTED_AT", "LOCATED_IN"],
    triples=[
        Triple(source="Person", relationship="WORKS_AT", target="Organization"),
        Triple(source="Organization", relationship="USES", target="System"),
        Triple(source="Person", relationship="PRESENTED", target="System"),
        Triple(source="System", relationship="PRESENTED_AT", target="Event"),
        Triple(source="Organization", relationship="LOCATED_IN", target="Location"),
    ],
    attributes={"Person": ["title"], "Organization": ["founded"]},
    entity_descriptions={"System": "a software product or platform"},
    relationship_descriptions={"USES": "an organization or person operating a system"},
)

Type and relationship names are sanitized into safe openCypher identifiers (non-alphanumerics collapse to _, result upper-cased), and membership is checked against the sanitized form — so Person matches PERSON in the extracted output.

Attributes become edges, not properties

An attribute declared in Ontology.attributes is reified: instead of writing salary as a property on the Person node, the writer stores (Person)-[:HAS_ATTRIBUTE {key: "salary"}]->(:AttributeValue {value: "..."}). Entity nodes are shared across documents and carry no access metadata, so a property written there would be visible to everyone who can reach the node. On its own edge, a sensitive attribute carries its own access scope and is filtered independently.

KnowledgeGraphEntityExtractor

Takes documents and returns the provider-neutral graph payload {"nodes": [...], "relationships": [...]}.

from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.nodes.knowledge_graphs import KnowledgeGraphEntityExtractor
from dynamiq.nodes.llms import OpenAI
from dynamiq.types import Document

extractor = KnowledgeGraphEntityExtractor(
    llm=OpenAI(
        connection=OpenAIConnection(),
        model="gpt-4o-mini",
        temperature=0.0,
        max_tokens=4000,
    ),
    ontology=ONTOLOGY,
)

extraction = extractor.run(
    input_data={
        "documents": [
            Document(
                content=(
                    "Acme Capital is a hedge fund based in New York. Jane Doe is the "
                    "Chief Investment Officer at Acme Capital. The firm uses an agentic "
                    "AI system called Helios for trade research."
                )
            ),
        ]
    }
)
print(extraction.output["nodes"], extraction.output["relationships"])

Each document is processed independently and gets a stable id if it has none. If one document fails extraction it is skipped and the rest continue; if every document fails, the node raises — a whole-batch failure is systemic (bad credentials, no model access) and should not be reported as an empty graph.

Provenance and access control are written onto edges

After ontology enforcement, every relationship the document produced gets that document's metadata copied onto it:

• source_doc_id (and source_doc_ids) — which document asserted this fact.
• identity_keys: ["source_doc_id"] — folds the document id into the edge's MERGE key, so the same fact asserted by two documents stays two edges with their own metadata rather than merging and overwriting.
• Every key in document.metadata rides along, flattened to graph-storable scalars (nested dicts are flattened with _-joined keys; anything non-primitive is JSON-encoded).

Nodes carry identity only — labels, id, and name. Put your access-control list in the document metadata and it lands on the edges, where the retriever enforces it:

Document(
    content="Acme Capital uses the Borealis analytics system.",
    metadata={"allowed_principals": ["group:finance"], "file_id": "f-1024"},
)

Because a node is visible exactly when it is reachable through a visible edge, one edge filter scopes the entire result. Graph Retrieval covers the enforcement side.

KnowledgeGraphWriter

Consumes the extractor's payload, assigns durable identity, and upserts. Returns {"nodes_created": int, "relationships_created": int}.

from dynamiq.connections import Neo4j as Neo4jConnection
from dynamiq.nodes.knowledge_graphs import KnowledgeGraphWriter

writer = KnowledgeGraphWriter(connection=Neo4jConnection())  # reads NEO4J_URI / _USERNAME / _PASSWORD

written = writer.run(input_data=extraction.output)
print(written.output["nodes_created"], written.output["relationships_created"])

How entities get their identity

The ids the LLM produces are throwaway wiring that links edges to nodes within a single extraction. Durable identity is decided here, in two tiers:

1. Deterministic (always). Every named entity gets uuid5(namespace, "{label}:{normalized_name}"). The same type and name hash to the same id on every machine and every run, so identical names collapse under MERGE with no graph read and re-ingestion is idempotent. The label is part of the hash, so an Apple Organization and an Apple Product never collide.
2. Fuzzy (optional, on by default). For entities whose deterministic id is not already in the graph, the writer pulls a bounded candidate set from an index — the entity vector index when embeddings are on, otherwise the entity-name full-text index — and adopts an existing entity's id when trigram similarity clears similarity_threshold. This merges spelling variants like Acme and Acme LLC.

The candidate lookup only proposes; trigram similarity always decides. That is why semantically close but distinct names (John Smith vs John Doe) are never fused. Set fuzzy_matching=False for deterministic-only behavior.

Two payload rules are worth knowing:

• Relationships must be written with their endpoint nodes. A relationship referencing an id absent from nodes cannot be resolved and would leave a dangling edge, so the writer raises ValueError instead.
• Bare nodes are not persisted. An entity referenced by no relationship is skipped: all provenance lives on edges, so such a node could never be attributed to a document, reached by retrieval, or removed by deletion. Nothing is lost — ids are content-addressed, so a later document asserting a fact about that name re-creates the identical node.

Indexes

On Neo4j the writer creates its indexes idempotently on init. Failures are logged rather than raised — a missing index only means retrieval falls back to a slower path.

Embeddings

Setting entity_embedder (Neo4j only) turns on two things at once:

• Each entity's name is embedded onto its node and backed by a vector index, so the retriever can seed traversal by semantic similarity instead of surface-form overlap.
• Each relationship's triplet text ("{source} {relation} {target}: {description}") is embedded onto the edge, so the retriever can rank facts by relevance server-side. It goes on the edge, not a new node, so the access scope stays where it belongs.

from dynamiq.nodes.embedders import OpenAIDocumentEmbedder

writer = KnowledgeGraphWriter(
    connection=Neo4jConnection(),
    entity_embedder=OpenAIDocumentEmbedder(
        connection=OpenAIConnection(), model="text-embedding-3-small"
    ),
)

Use the same embedding model on the retriever's text_embedder so the vector dimensions match. Embedding is best-effort: if the embedder fails, the write proceeds without embeddings.

A full ingestion workflow

Vector storage and the graph are independent branches of the same flow, so one pass over your documents populates both — the setup Graph Retrieval uses to combine facts with verbatim passages.

from dynamiq import Workflow
from dynamiq.connections import Neo4j as Neo4jConnection
from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.connections import Qdrant as QdrantConnection
from dynamiq.flows import Flow
from dynamiq.nodes.embedders import OpenAIDocumentEmbedder
from dynamiq.nodes.knowledge_graphs import (
    KnowledgeGraphEntityExtractor,
    KnowledgeGraphWriter,
    Ontology,
    Triple,
)
from dynamiq.nodes.llms import OpenAI
from dynamiq.nodes.node import InputTransformer, NodeDependency
from dynamiq.nodes.writers import QdrantDocumentWriter
from dynamiq.runnables import RunnableConfig, RunnableStatus
from dynamiq.types import Document

openai_connection = OpenAIConnection()

ontology = Ontology(
    entity_types=["Person", "Organization", "System"],
    relationship_types=["WORKS_AT", "USES"],
    triples=[
        Triple(source="Person", relationship="WORKS_AT", target="Organization"),
        Triple(source="Organization", relationship="USES", target="System"),
    ],
)

# Vector branch: embed, then write to Qdrant.
document_embedder = OpenAIDocumentEmbedder(
    id="document_embedder",
    connection=openai_connection,
    model="text-embedding-3-small",
    input_transformer=InputTransformer(selector={"documents": "$.documents"}),
)
vector_writer = QdrantDocumentWriter(
    id="vector_writer",
    connection=QdrantConnection(),
    index_name="kg_demo",
    create_if_not_exist=True,
    dimension=1536,
    depends=[NodeDependency(document_embedder)],
    input_transformer=InputTransformer(
        selector={"documents": "$.document_embedder.output.documents"}
    ),
)

# Graph branch: extract with the LLM, then resolve and upsert to Neo4j.
entity_extractor = KnowledgeGraphEntityExtractor(
    id="entity_extractor",
    llm=OpenAI(connection=openai_connection, model="gpt-4o-mini", temperature=0.0, max_tokens=4000),
    ontology=ontology,
    input_transformer=InputTransformer(selector={"documents": "$.documents"}),
)
graph_writer = KnowledgeGraphWriter(
    id="graph_writer",
    connection=Neo4jConnection(),
    depends=[NodeDependency(entity_extractor)],
    input_transformer=InputTransformer(
        selector={
            "nodes": "$.entity_extractor.output.nodes",
            "relationships": "$.entity_extractor.output.relationships",
        }
    ),
)

workflow = Workflow(
    flow=Flow(nodes=[document_embedder, vector_writer, entity_extractor, graph_writer])
)

result = workflow.run(
    input_data={
        "documents": [
            Document(
                content="Jane Doe is the CIO of Acme Capital. The firm uses Helios for trade research.",
                metadata={"allowed_principals": ["group:public"]},
            ),
        ]
    },
    config=RunnableConfig(request_timeout=120),
)

if result.status != RunnableStatus.SUCCESS:
    raise RuntimeError(f"Ingestion failed: {result.output}")

graph_out = result.output["graph_writer"]["output"]
print(graph_out["nodes_created"], graph_out["relationships_created"])

To run this locally you need OPENAI_API_KEY and a Neo4j reachable through NEO4J_URI, NEO4J_USERNAME, and NEO4J_PASSWORD:

docker run -p 7687:7687 -p 7474:7474 -e NEO4J_AUTH=neo4j/password neo4j:5

Parallel extraction

LLM extraction is I/O-bound and embarrassingly parallel; entity resolution is not. Fan extraction out with the Map operator, merge the payloads, and funnel everything into one writer — resolution then converges duplicates across all shards in a single pass.

from dynamiq.nodes.operators import Map

parallel_extract = Map(node=entity_extractor, max_workers=3)

shards = [documents[i::3] for i in range(3)]
map_result = parallel_extract.run(
    input_data={"input": [{"documents": shard} for shard in shards if shard]},
    config=RunnableConfig(request_timeout=120),
)

per_shard = map_result.output["output"]
merged = {
    "nodes": [n for out in per_shard for n in out["nodes"]],
    "relationships": [r for out in per_shard for r in out["relationships"]],
}
graph_writer.run(input_data=merged)

Updating and deleting documents

Writing never deletes. To replace a document's facts, delete them first and then re-write:

totals = graph_writer.delete_documents(["doc-123", "doc-456"])
print(totals["relationships_deleted"], totals["nodes_deleted"])

Deletion is clean by construction. Every edge carries the provenance of the document that asserted it, and the same fact from two documents is two separate edges — so removing one document never erases another document's identical claim. Document-scoped AttributeValue holders go with their edges, and a shared entity node survives as long as another document still cites it, then gets swept once this delete removes its last edge.

The second argument selects which edge property the ids match against. It defaults to source_doc_id (the ingestion chunk id); pass any other flattened metadata key to delete at a coarser grain:

graph_writer.delete_documents(["f-1024"], key="file_id")

The corresponding chunks in your vector store are not touched — the writer does not own that store, so delete them separately with the vector store's own delete-by-ids.

Graph backends

All three speak openCypher, and the concrete store is selected from the connection type. Index-backed seeding, entity embeddings, write_graph, and delete_documents are Neo4j-only today; the retriever degrades transparently on the others.

Next steps