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.

For how the pipeline works, see Ingestion pipeline.

Before you start

This guide assumes that you finished the Quickstart: you have a Neo4j database, its NEO4J_* variables in a .env file, and the neo4j and extract extras installed. Each code block on this page is complete. It opens its own graph and closes it at the end.

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

uv pip install "agentic-graphrag[docling]"

Add a single file

Graph.open connects to Neo4j and creates the constraints and indexes that the graph needs. Graph.add then loads, chunks, extracts, and stores the file.

import asyncio
from pathlib import Path
from tempfile import TemporaryDirectory

from agrag.common.data_models.graph_schema import GENERIC
from agrag.embedding import build_embedder
from agrag.graphdb import build_graph_store
from agrag.ingestion import Graph
from agrag.ingestion.extract import GlinerExtractor


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    graph = await Graph.open(
        schema=GENERIC,
        graph_store=graph_store,
        embedder=build_embedder("ibm-granite/granite-embedding-small-english-r2"),
        extractor=GlinerExtractor(),
    )
    try:
        with TemporaryDirectory() as folder:
            path = Path(folder) / "notes.md"
            path.write_text("# Notes\n\nAda Lovelace wrote the first published program.\n")
            result = await graph.add(source=str(path))
            print(result.ingestion.documents)
    finally:
        await graph_store.close()


asyncio.run(main())

Add many files

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

import asyncio
from pathlib import Path
from tempfile import TemporaryDirectory

from agrag.common.data_models.graph_schema import GENERIC
from agrag.embedding import build_embedder
from agrag.graphdb import build_graph_store
from agrag.ingestion import Graph
from agrag.ingestion.extract import GlinerExtractor


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    graph = await Graph.open(
        schema=GENERIC,
        graph_store=graph_store,
        embedder=build_embedder("ibm-granite/granite-embedding-small-english-r2"),
        extractor=GlinerExtractor(),
    )
    try:
        with TemporaryDirectory() as folder:
            root = Path(folder)
            (root / "a.txt").write_text("Charles Babbage designed the Analytical Engine.")
            (root / "b.md").write_text("# Engine\n\nThe engine was never finished.")
            result = await graph.add(source=folder)  # every supported file
            print(result.ingestion.documents)
            result = await graph.add(source=f"{folder}/**/*.md")  # glob pattern
            print(result.ingestion.documents)
            result = await graph.add(source=[root / "a.txt", root / "b.md"])  # explicit list
            print(result.ingestion.documents)
    finally:
        await graph_store.close()


asyncio.run(main())

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

Add raw text

Pass a string to ingest prose as a single document:

import asyncio
from pathlib import Path
from tempfile import TemporaryDirectory

from agrag.common.data_models.graph_schema import GENERIC
from agrag.embedding import build_embedder
from agrag.graphdb import build_graph_store
from agrag.ingestion import Graph
from agrag.ingestion.extract import GlinerExtractor


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    graph = await Graph.open(
        schema=GENERIC,
        graph_store=graph_store,
        embedder=build_embedder("ibm-granite/granite-embedding-small-english-r2"),
        extractor=GlinerExtractor(),
    )
    try:
        result = await graph.add(text="Raw prose to index as one document.")
        print(result.ingestion.documents)
    finally:
        await graph_store.close()


asyncio.run(main())

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:

import asyncio
from pathlib import Path
from tempfile import TemporaryDirectory

from agrag.common.data_models.graph_schema import GENERIC
from agrag.embedding import build_embedder
from agrag.graphdb import build_graph_store
from agrag.ingestion import Graph
from agrag.ingestion.extract import GlinerExtractor
import hashlib

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


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    graph = await Graph.open(
        schema=GENERIC,
        graph_store=graph_store,
        embedder=build_embedder("ibm-granite/granite-embedding-small-english-r2"),
        extractor=GlinerExtractor(),
    )
    try:
        text = "Prebuilt document text."
        doc = Document(
            text=text,
            title="Prebuilt",
            uri="memory://prebuilt",
            source_format=SourceFormat.TXT,
            family=DocumentFamily.PROSE,
            content_hash=hashlib.sha256(text.encode()).hexdigest(),
            loader_name="text",
            char_count=len(text),
        )
        result = await graph.add(documents=[doc])
        print(result.ingestion.documents)
    finally:
        await graph_store.close()


asyncio.run(main())

For most callers, source= or text= is simpler. Use 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 on each 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:

import asyncio
from pathlib import Path
from tempfile import TemporaryDirectory

from agrag.common.data_models.graph_schema import GENERIC
from agrag.embedding import build_embedder
from agrag.graphdb import build_graph_store
from agrag.ingestion import Graph
from agrag.ingestion.extract import GlinerExtractor
from agrag.loaders.corpus import ChatLoader


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    graph = await Graph.open(
        schema=GENERIC,
        graph_store=graph_store,
        embedder=build_embedder("ibm-granite/granite-embedding-small-english-r2"),
        extractor=GlinerExtractor(),
    )
    try:
        with TemporaryDirectory() as folder:
            path = Path(folder) / "session.jsonl"
            path.write_text(
                '{"role": "user", "content": "Hi there", "id": 1}\n'
                '{"role": "assistant", "content": "Hello! How can I help?", "id": 2}\n'
            )
            result = await graph.add(source=str(path), loader=ChatLoader())
            print(result.ingestion.documents)
    finally:
        await graph_store.close()


asyncio.run(main())

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 to each source, not to the whole run, so one bad file does not stop the rest.

import asyncio
from pathlib import Path
from tempfile import TemporaryDirectory

from agrag.common.data_models.graph_schema import GENERIC
from agrag.embedding import build_embedder
from agrag.graphdb import build_graph_store
from agrag.ingestion import Graph
from agrag.ingestion.extract import GlinerExtractor
from agrag.loaders.corpus.types import ErrorPolicy


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    graph = await Graph.open(
        schema=GENERIC,
        graph_store=graph_store,
        embedder=build_embedder("ibm-granite/granite-embedding-small-english-r2"),
        extractor=GlinerExtractor(),
    )
    try:
        with TemporaryDirectory() as folder:
            (Path(folder) / "good.txt").write_text("Ada Lovelace was born in London.")
            (Path(folder) / "unknown.xyz").write_text("No loader claims this extension.")
            result = await graph.add(source=folder, error_policy=ErrorPolicy.SKIP)
            print(result.ingestion.skipped, result.ingestion.quarantined)
    finally:
        await graph_store.close()


asyncio.run(main())

• ErrorPolicy.RAISE stops and raises on the first failure. This is the default.
• ErrorPolicy.SKIP drops the failing source and continues.
• ErrorPolicy.QUARANTINE sets the failing source aside and continues.

Read the outcome from the returned AddResult:

import asyncio
from pathlib import Path
from tempfile import TemporaryDirectory

from agrag.common.data_models.graph_schema import GENERIC
from agrag.embedding import build_embedder
from agrag.graphdb import build_graph_store
from agrag.ingestion import Graph
from agrag.ingestion.extract import GlinerExtractor
from agrag.loaders.corpus.types import ErrorPolicy


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    graph = await Graph.open(
        schema=GENERIC,
        graph_store=graph_store,
        embedder=build_embedder("ibm-granite/granite-embedding-small-english-r2"),
        extractor=GlinerExtractor(),
    )
    try:
        with TemporaryDirectory() as folder:
            (Path(folder) / "good.txt").write_text("Ada Lovelace was born in London.")
            (Path(folder) / "unknown.xyz").write_text("No loader claims this extension.")
            result = await graph.add(source=folder, error_policy=ErrorPolicy.QUARANTINE)
            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
            for item in result.ingestion.quarantined_items:  # one StageFailure per source
                print(item.item_id, item.error_message)
    finally:
        await graph_store.close()


asyncio.run(main())

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 an AddResult after each batch. Use it to log or stream status without blocking ingestion:

import asyncio
from pathlib import Path
from tempfile import TemporaryDirectory

from agrag.common.data_models.graph_schema import GENERIC
from agrag.embedding import build_embedder
from agrag.graphdb import build_graph_store
from agrag.ingestion import Graph
from agrag.ingestion.extract import GlinerExtractor
from agrag.ingestion.reports.add_result import AddResult


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    graph = await Graph.open(
        schema=GENERIC,
        graph_store=graph_store,
        embedder=build_embedder("ibm-granite/granite-embedding-small-english-r2"),
        extractor=GlinerExtractor(),
    )
    try:
        def report(result: AddResult) -> None:
            print(f"loaded {result.ingestion.documents} documents so far")


        with TemporaryDirectory() as folder:
            (Path(folder) / "a.txt").write_text("Charles Babbage designed the Analytical Engine.")
            result = await graph.add(source=folder, on_progress=report)
    finally:
        await graph_store.close()


asyncio.run(main())

Supported formats

Loaders are chosen by file extension. See Loading documents for the formats and the extra each one needs.

Next steps

• See the API Reference for every parameter.
• Change how documents are split in Configure chunking.
• Update and delete documents replaces a changed document or removes one.
• Ingestion pipeline explains how an interrupted call recovers.