Configure storage backends

This guide sets up an embedder, a vector store, and a graph store. Use it before you store or search embeddings.

Before you start

Each code block on this page is complete. It opens and closes its own connections. Blocks that use a vector store read the QDRANT_* variables, and blocks that use the graph store read the NEO4J_* variables. See the Quickstart for a Neo4j database.

Install

Install the core package. Then install one extra for each backend you use.

uv pip install agentic-graphrag
uv pip install "agentic-graphrag[embed-local]"   # sentence-transformers embedding models
uv pip install "agentic-graphrag[qdrant]"        # Qdrant vector store
uv pip install "agentic-graphrag[weaviate]"      # Weaviate vector store
uv pip install "agentic-graphrag[milvus]"        # Milvus vector store
uv pip install "agentic-graphrag[neo4j]"         # Neo4j graph store

You must install at least one vector store extra to use VectorStore. You must install the neo4j extra to use GraphStore.

Set up the embedder

build_embedder takes a FastEmbed model name. It returns a ready-to-use embedder that runs the model on the CPU with ONNX Runtime. The default model is ibm-granite/granite-embedding-small-english-r2.

import asyncio

from agrag.embedding import build_embedder


async def main() -> None:
    embedder = build_embedder("ibm-granite/granite-embedding-small-english-r2")
    vectors = await embedder.embed(["first passage", "second passage"])
    print(await embedder.dimensions())  # the vector length this model produces


asyncio.run(main())

To use another model that FastEmbed supports, pass its name to build_embedder. To use a sentence-transformers model, or to set the device, build a SentenceTransformerEmbedder yourself. It needs the embed-local extra. Pass its own EmbeddingSettings.

from agrag.embedding import EmbeddingSettings, SentenceTransformerEmbedder

embedder = SentenceTransformerEmbedder(
    settings=EmbeddingSettings(model="BAAI/bge-small-en-v1.5", device="cuda"),
)

Set embedder configuration with environment variables

Set these variables instead of passing EmbeddingSettings fields in code.

Variable	Meaning
EMBEDDING_MODEL	The model name.
EMBEDDING_DEVICE	The device to load the model on. Only SentenceTransformerEmbedder uses it.
EMBEDDING_NORMALIZE	Whether to normalize output vectors.
EMBEDDING_BATCH_SIZE	The number of texts per encode call.

Set up a vector store

build_vector_store takes a backend name. It reads connection configuration from environment variables.

import asyncio

from agrag.common.data_models.vector_record import Distance
from agrag.embedding import build_embedder
from agrag.vectordb import build_vector_store


async def main() -> None:
    embedder = build_embedder("ibm-granite/granite-embedding-small-english-r2")
    store = build_vector_store("qdrant")
    await store.initialize()
    await store.ensure_collection(
        "guide_example",
        dimensions=await embedder.dimensions(),
        distance=Distance.COSINE,
        hybrid=True,
    )


asyncio.run(main())

Set dimensions to the value your embedder produces. ensure_collection raises an error if a collection with that name already exists with a different dimension. Pass hybrid=True to search the collection with hybrid_search later. If you only need search, leave it out.

For full control, build a backend class yourself with its own configuration.

from agrag.vectordb.qdrant import QdrantVectorStore
from agrag.vectordb.settings import QdrantSettings

store = QdrantVectorStore(
    settings=QdrantSettings(url="https://xyz.cloud.qdrant.io", api_key="...")
)

Set vector store credentials with environment variables

Set the variables for the one backend you use.

Backend	Variables
Qdrant	QDRANT_URL, QDRANT_API_KEY, QDRANT_REQUIRE_TLS
Weaviate	WEAVIATE_URL, WEAVIATE_API_KEY, WEAVIATE_REQUIRE_TLS
Milvus	MILVUS_URI, MILVUS_TOKEN, MILVUS_REQUIRE_TLS

The default URLs point at plaintext localhost, for the Docker Compose services make dev-services-up starts. If a URL uses a plaintext scheme (http), points at a non-local host, and carries an API key or token, the configuration check raises an error. Use https:// for any remote deployment.

An unauthenticated backend on a private network (a VPC, a cluster-internal service) is not flagged by default, since the configuration cannot tell a private host from a public one from the URL alone. If your deployment needs every remote connection encrypted, set *_REQUIRE_TLS=true. It rejects plaintext to any non-local host, with or without a credential.

Write and search vectors

Write vectors with upsert. Each record needs an id, a vector, and a payload.

import asyncio

from uuid import uuid4

from agrag.common.data_models.vector_record import Distance, VectorRecord
from agrag.embedding import build_embedder
from agrag.vectordb import build_vector_store


async def main() -> None:
    embedder = build_embedder("ibm-granite/granite-embedding-small-english-r2")
    vectors = await embedder.embed(["first passage", "second passage"])
    store = build_vector_store("qdrant")
    await store.initialize()
    await store.ensure_collection(
        "guide_example",
        dimensions=await embedder.dimensions(),
        distance=Distance.COSINE,
        hybrid=True,
    )

    record = VectorRecord(
        id=uuid4(), vector=vectors[0], payload={"text": "first passage"}
    )
    await store.upsert("guide_example", [record])


asyncio.run(main())

Search by vector alone with search. Search by vector and keyword together with hybrid_search.

import asyncio

from uuid import uuid4

from agrag.common.data_models.vector_record import Distance, VectorRecord
from agrag.embedding import build_embedder
from agrag.vectordb import build_vector_store


async def main() -> None:
    embedder = build_embedder("ibm-granite/granite-embedding-small-english-r2")
    vectors = await embedder.embed(["sepsis management protocol"])
    store = build_vector_store("qdrant")
    await store.initialize()
    await store.ensure_collection(
        "guide_example",
        dimensions=await embedder.dimensions(),
        distance=Distance.COSINE,
        hybrid=True,
    )
    await store.upsert(
        "guide_example",
        [VectorRecord(id=uuid4(), vector=vectors[0], payload={"text": "sepsis"})],
    )

    hits = await store.search("guide_example", query_vector=vectors[0], limit=5)
    hits = await store.hybrid_search(
        "guide_example",
        query_vector=vectors[0],
        query_text="sepsis management",
        limit=5,
    )
    for hit in hits:
        print(hit.id, hit.score, hit.payload)


asyncio.run(main())

Set up a graph store

Build a Neo4jGraphStore with Neo4jSettings. Call connect before you use it.

import asyncio

from agrag.graphdb.neo4j import Neo4jGraphStore
from agrag.graphdb.settings import Neo4jSettings


async def main() -> None:
    store = Neo4jGraphStore(settings=Neo4jSettings())
    await store.connect()
    await store.close()


asyncio.run(main())

Set graph store credentials with environment variables

Variable	Meaning
NEO4J_URI	The Bolt connection URI. Use neo4j+s://... for Aura.
NEO4J_USERNAME	The database username.
NEO4J_PASSWORD	The database password.
NEO4J_DATABASE	The target database name.

Neo4j always authenticates with a password. If NEO4J_URI uses a plaintext scheme (bolt:// or neo4j://) and points at a non-local host, Agentic GraphRAG raises an error. Use neo4j+s:// (or bolt+s://) for a remote instance.

Write nodes

Write nodes with upsert_nodes. Each record needs an id, its labels, and a properties map. Call setup_constraints after your first write of a label, so Neo4j knows which labels to constrain.

import asyncio

from uuid import uuid4

from agrag.common.data_models.graph_record import NodeRecord
from agrag.graphdb.neo4j import Neo4jGraphStore
from agrag.graphdb.settings import Neo4jSettings


async def main() -> None:
    store = Neo4jGraphStore(settings=Neo4jSettings())
    await store.connect()
    try:
        node = NodeRecord(
            id=uuid4(),
            labels=["Note"],
            properties={"text": "first passage"},
        )
        await store.upsert_nodes("Note", [node])
        await store.setup_constraints()
    finally:
        await store.close()


asyncio.run(main())

A node can carry more than one label. List every label that the node must have in labels. upsert_nodes writes all of them, not only the label argument. Records with different label sets in the same call are grouped and written separately.

import asyncio

from uuid import uuid4

from agrag.common.data_models.graph_record import NodeRecord
from agrag.graphdb.neo4j import Neo4jGraphStore
from agrag.graphdb.settings import Neo4jSettings


async def main() -> None:
    store = Neo4jGraphStore(settings=Neo4jSettings())
    await store.connect()
    try:
        tagged_node = NodeRecord(
            id=uuid4(),
            labels=["Note", "Draft"],
            properties={"text": "sepsis protocol"},
        )
        await store.upsert_nodes("Note", [tagged_node])
    finally:
        await store.close()


asyncio.run(main())

Labels only ever accumulate. If a later upsert of the same node id lists fewer labels, the labels already on the node are kept, not removed.

Handle partial write failures

upsert_nodes and upsert_relations return an UpsertResult. Read failures for records that did not write. The rest of the call still lands.

import asyncio

from uuid import uuid4

from agrag.common.data_models.graph_record import NodeRecord
from agrag.graphdb.neo4j import Neo4jGraphStore
from agrag.graphdb.settings import Neo4jSettings


async def main() -> None:
    store = Neo4jGraphStore(settings=Neo4jSettings())
    await store.connect()
    try:
        nodes = [
            NodeRecord(id=uuid4(), labels=["Note"], properties={"text": "example"})
        ]
        result = await store.upsert_nodes("Note", nodes)
        if result.failures:
            for failure in result.failures:
                print(f"{failure.id} failed: {failure.error_message}")
        print(f"{result.written} of {len(nodes)} nodes written")
    finally:
        await store.close()


asyncio.run(main())

A connectivity or authentication failure still raises instead of returning a partial result. Only a failure specific to one record's data is isolated and reported this way.

Search vectors stored in the graph

You can search vectors stored as node properties, without a separate vector store. First create a vector index on the property. Then search it.

import asyncio

from uuid import uuid4

from agrag.common.data_models.graph_record import NodeRecord
from agrag.common.data_models.vector_record import Distance
from agrag.embedding import build_embedder
from agrag.graphdb.neo4j import Neo4jGraphStore
from agrag.graphdb.settings import Neo4jSettings


async def main() -> None:
    embedder = build_embedder("ibm-granite/granite-embedding-small-english-r2")
    vectors = await embedder.embed(["first passage"])
    store = Neo4jGraphStore(settings=Neo4jSettings())
    await store.connect()
    try:
        node = NodeRecord(
            id=uuid4(),
            labels=["Note"],
            properties={"text": "first passage", "embedding": vectors[0]},
        )
        await store.upsert_nodes("Note", [node])
        await store.ensure_vector_index(
            label="Note",
            vector_property="embedding",
            dimensions=await embedder.dimensions(),
            distance=Distance.COSINE,
        )
        hits = await store.vector_search(
            label="Note", vector_property="embedding", query_vector=vectors[0], limit=5
        )
    finally:
        await store.close()


asyncio.run(main())

vector_search returns the same hit type VectorStore.search returns. Code that reads hits does not need to know which store produced them.

Mirror ingestion into a vector store

Graph.open accepts an optional vector_store. Pass one. Every embedding ingestion writes to the graph store is also upserted to the vector store. Entity, chunk, and community vectors all land in the collections RetrievalSettings names (agrag_entities, agrag_chunks, agrag_communities by default). open() creates a missing collection with your embedder's dimensions. After entity merges, the absorbed entities' vectors are deleted there. Each detect_communities(apply=True) run replaces the community vectors after stale-vector cleanup succeeds. That cleanup clears every Community vector in the collection, the way the graph-side recompute clears every Community node in the database. Give each graph its own Neo4j database (NEO4J_DATABASE) and its own collections. A cleanup failure can leave old community vectors in the collection. Agentic GraphRAG reports it to the caller.

import asyncio

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.vectordb import build_vector_store


async def main() -> None:
    graph_store = build_graph_store("neo4j")
    vector_store = build_vector_store("qdrant")
    graph = await Graph.open(
        schema=GENERIC,
        graph_store=graph_store,
        embedder=build_embedder("ibm-granite/granite-embedding-small-english-r2"),
        extractor=GlinerExtractor(),
        vector_store=vector_store,
    )
    await graph_store.close()


asyncio.run(main())

A SearchEngine pointed at the same vector_store then searches the same data the graph-native path serves. Ingestion reports failed vector-store writes in each stage's failures. Under ErrorPolicy.SKIP/QUARANTINE, Agentic GraphRAG records a failed mirror as a StageFailure such as chunk_vector_store. Under the default ErrorPolicy.RAISE, it propagates.

Next steps

See the API Reference for every parameter.