Ingest documents

Add files, directories, raw text, or prebuilt Document objects to a Graph with Graph.add. This guide covers the ingestion entrypoint only. See the API Reference for every parameter.

Install

Install the core package:

uv pip install agentic-graphrag

For PDF, DOCX, PPTX, image, and XML support, install the docling extra:

uv pip install "agentic-graphrag[docling]"

Add a single file

Open a graph and add one file by path. Graph.open is async and needs no external service.

import asyncio
from agrag.ingestion import Graph

async def main():
    graph = await Graph.open()
    result = await graph.add(source="path/to/file.pdf")
    print(result.ingestion.documents)  # number of documents ingested

asyncio.run(main())

Every later snippet on this page continues inside that same async def main(): ... body, using the graph opened above, and still runs through asyncio.run(main()).

Add a directory, a glob, or a list of paths the same way:

result = await graph.add(source="./corpus")            # every supported file
result = await graph.add(source="./corpus/**/*.md")     # glob pattern
result = await graph.add(source=["a.txt", "b.pdf"])     # explicit list

Graph.add accepts exactly one of source, text, or documents.

Add raw text

Pass a string to ingest prose as a single document:

result = await graph.add(text="Raw prose to index as one document.")

Add prebuilt documents

Pass Document objects you already built (for example, documents returned by a loader or reconstructed from storage) to documents. A Document carries its source metadata, so every field is required:

from agrag.ingestion import Graph
from agrag.common.data_models.document import (
    Document,
    DocumentFamily,
    SourceFormat,
)

text = "Prebuilt document text."
doc = Document(
    text=text,
    title="Prebuilt",
    uri="memory://prebuilt",
    source_format=SourceFormat.TXT,
    family=DocumentFamily.PROSE,
    content_hash="replace-with-a-real-sha256",
    loader_name="text",
    char_count=len(text),
)
result = await graph.add(documents=[doc])

For most callers, source= or text= is simpler; reach for documents= when you already hold Document objects.

Choose how documents are chunked

Graph.open takes a chunking argument that decides which chunker splits each document. Loaders also normalize the text of each file: they remove the byte-order mark, change line endings to LF, and apply NFKC. Pass read_options=ReadOptions(normalization=...) to Graph.add to change this. PDF and Office documents chunk at up to 1024 tokens by default. See Configure chunking for both.

Add a chat file

ChatLoader reads a file of chat messages as one document. Use .jsonl or .ndjson with one JSON object per line, or .json with an array of objects. Each message needs a non-empty string role and a string content. It can also have an id, which must be unique in the file.

{"role": "user", "content": "Hi there", "id": 1}
{"role": "assistant", "content": "Hello! How can I help?", "id": 2}

The document text has one [role] content block for each message, with a blank line between blocks. Document.turns records the role, the id and the character span of each block. Pass the loader for one file:

from agrag.loaders.corpus import ChatLoader

result = await graph.add(source="session.jsonl", loader=ChatLoader())

To cut the document between turns, use TurnWindowChunker as Configure chunking shows.

Handle errors per source

Use error_policy to control what happens when a source fails to load. The policy applies per source, not per run, so one bad file does not stop the rest.

from agrag.ingestion import Graph
from agrag.loaders.corpus.types import ErrorPolicy

result = await graph.add(source="./corpus", error_policy=ErrorPolicy.SKIP)
print(result.ingestion.skipped, result.ingestion.quarantined)

• ErrorPolicy.RAISE — stop and raise on the first failure (default).
• ErrorPolicy.SKIP — drop the failing source and continue.
• ErrorPolicy.QUARANTINE — isolate the failing source and continue.

Read the outcome from the returned AddResult:

print(result.ingestion.documents)  # count of documents ingested
print(result.ingestion.sources)    # count of sources processed
print(result.ingestion.skipped)    # count of skipped sources
print(result.ingestion.quarantined)  # count of quarantined sources
print(result.ingestion.quarantined_items)  # one StageFailure per quarantined source

An unknown format raises UnsupportedFormatError; a format that needs docling without the extra installed raises MissingExtraError. With SKIP or QUARANTINE, the policy handles those sources instead of stopping the run.

Track progress

Pass on_progress to receive a LoadStats callback after each batch. Use it to log or stream status without blocking ingestion:

from agrag.ingestion import Graph
from agrag.loaders.corpus.types import LoadStats

def report(stats: LoadStats) -> None:
    print(f"loaded {stats.documents} documents from {stats.sources} sources so far")

result = await graph.add(source="./corpus", on_progress=report)

Update or delete a document

Use Graph.update to ingest new content under the same stable document_key. If the content hash and the chunker settings are unchanged, the operation does not re-chunk or rewrite the document.

result = await graph.update("memory://manual", text="Revised document text")
print(result.no_op)
print(result.chunks_closed)

Use Graph.delete_document to remove a document from current retrieval without deleting its chunks or extracted entities. The graph keeps those records for provenance.

result = await graph.delete_document("memory://manual")
print(result.chunks_closed)

Crash safety

add, update, and delete_document each run as a Cutover Job. Every node, edge, and vector a call writes is tagged with that job's id, and every retrieval path excludes tagged writes, so an in-flight call is never visible to a search. One transaction then flips the job to committed, closes the superseded version's PART_OF edges, and clears the tags; deletion-triggered pruning runs after that, scoped to the entities the replaced version mentioned.

If the process dies mid-call, the next Graph.open() recovers automatically: a job that died before its commit is rolled back, deleting its tagged writes, and a job that died after committing is rolled forward. Either way the graph converges to as-if-never-happened or as-if-completed. A document's job holds a lease that expires after CUTOVER_JOB_LEASE_TTL_SECONDS seconds (60 by default). A running worker renews its lease, so no other worker takes over a live job, however long it runs. If the worker stops, another worker takes the document over after the lease expires.

Supported formats

Loaders are chosen by file extension. Core loaders are always available; docling loaders require the docling extra.

Format	Extension(s)	Loader
Plain text	.txt, .log	core
Markdown	.md, .markdown	core
AsciiDoc	.adoc, .asciidoc	docling, falls back to core (regex)
HTML	.html, .htm	core
CSV / TSV	.csv, .tsv	core
JSON Lines	.jsonl, .ndjson	core
JSON	.json	core
PDF	.pdf	docling
Word	.docx	docling
PowerPoint	.pptx	docling
Images	.png, .jpg, .jpeg, .tif, .tiff, .bmp	docling
XML	.xml	docling

For .md, .html, and .csv, docling registers a loader but the core loader wins by default. For .adoc and .asciidoc, docling takes precedence instead: docling's structural parser wins when the docling extra is installed, and the core regex-based reader is the fallback when it is not. To customize per-source reading (encoding, size limits, column mapping, and more), build Document objects through the lower-level loaders with ReadOptions from agrag.loaders.corpus.types.

Next steps

See the API Reference for every parameter.