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.

Markdown, HTML, AsciiDoc, Word, PowerPoint, and Excel files need no extra. For PDF and image support, install the docling extra. The first PDF downloads about 500 MB of models:

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 import DEFAULT_SCHEMA
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=DEFAULT_SCHEMA,
        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 import DEFAULT_SCHEMA
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=DEFAULT_SCHEMA,
        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 import DEFAULT_SCHEMA
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=DEFAULT_SCHEMA,
        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 import DEFAULT_SCHEMA
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=DEFAULT_SCHEMA,
        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 import DEFAULT_SCHEMA
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 import ChatLoader


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    graph = await Graph.open(
        schema=DEFAULT_SCHEMA,
        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())

A chat file has one section for each message. Short neighbouring messages share a chunk, and a long message is cut at sentence boundaries. Tune this with the chunker size, 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 import DEFAULT_SCHEMA
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.types import ErrorPolicy


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    graph = await Graph.open(
        schema=DEFAULT_SCHEMA,
        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.

With SKIP or QUARANTINE, the walk buffers one source's documents until its loader finishes successfully. A malformed row therefore discards the whole source, even after enough rows to fill a batch. Successful sources still yield bounded batches with resume cursors. Buffering uses memory proportional to the documents in one source; RAISE keeps streaming batches as they fill.

Read the outcome from the returned AddResult:

import asyncio
from pathlib import Path
from tempfile import TemporaryDirectory

from agrag.common.data_models import DEFAULT_SCHEMA
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.types import ErrorPolicy


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    graph = await Graph.open(
        schema=DEFAULT_SCHEMA,
        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())

result.escalation reports the chunks that an LLM extractor read. See Control LLM cost during extraction.

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. Each result counts the writes the graph has made so far. 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 import DEFAULT_SCHEMA
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=DEFAULT_SCHEMA,
        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())

Graph.add reads documents in batches of up to 100. It writes each batch to the graph as soon as the batch is extracted, before it reads the next batch. If a later batch fails, the earlier batches stay in the graph. There is no resume point, so calling add again starts from the first document and processes every document again. Each add call needs distinct document keys. A duplicate key found in a later batch raises an error, and the batches before it stay in the graph.

Supported formats

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

CSV, TSV, JSON Lines, and JSON record loaders accept published_at_column and content_date_column in ReadOptions. Set each option to the column that holds the date the source was published or the date the content describes. Values must use an ISO date such as 2024-05-01 or an ISO date-time. Blank, null, or missing values produce None. A malformed value raises MalformedRecordError with the column name and source.

For other sources, set published_at and content_date to datetime.date values on a Document that you pass to Graph.add(documents=...). The graph stores each date as an ISO string, such as 2024-05-01. Graph.update applies date-only changes without re-extracting the document.

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.