Backend Reference — `rag-backends`

The rag-backends package ships production-ready implementations of the rag-core SPI interfaces. Import ABCs from rag_core.spi; inject backend instances at the composition root.

Overview

Class	SPI	Backend	Extra
`PgVectorStore`	`VectorStore`	PostgreSQL + pgvector	— (default)
`QdrantVectorStore`	`VectorStore`	Qdrant	— (default)
`WeaviateVectorStore`	`VectorStore`	Weaviate v4	`[weaviate]`
`PineconeVectorStore`	`VectorStore`	Pinecone (serverless)	`[pinecone]`
`ElasticsearchDenseVectorStore`	`VectorStore`	Elasticsearch 8.x kNN	`[elasticsearch]`
`ElasticsearchKeywordStore`	`KeywordStore`	Elasticsearch 8.x BM25	`[elasticsearch]`
`TantivyKeywordStore`	`KeywordStore`	tantivy (Rust core, in-process)	`[tantivy]`
`Neo4jGraphStore`	`GraphStore`	Neo4j 5.x (Bolt + Cypher)	`[neo4j]`
`MemgraphGraphStore`	`GraphStore`	Memgraph (Bolt + Cypher)	`[memgraph]`
`NetworkXGraphStore`	`GraphStore`	networkx (in-process)	`[networkx]`
`RedisCache`	`Cache`	Redis	— (default)
`RedisRetrievalCache`	`RetrievalCache`	Redis	— (default)
`RedisAnswerCache`	`AnswerCache`	Redis	— (default)
`S3Storage`	`Storage`	AWS S3 / MinIO / GCS S3-interop	— (default)
`LocalFileStorage`	`Storage`	Local filesystem	— (default)

All five vector backends accept an optional hint: IndexHint on initialize() and consult it to pick the index variant (flat / ivfflat / HNSW / IVF-PQ — see ADR-0009 and docs/architecture/vector-backends.md).

PgVectorStore

PostgreSQL-backed dense vector store using the pgvector extension.

from rag_backends.vector.pgvector import PgVectorStore
from rag_core.types import ChunkId, Embedding, TenantId

store = PgVectorStore(
    dsn="postgresql://rag:rag@localhost:5432/rag",
    table="rag_vector_store",  # default
    dimension=1536,            # must match your embedder
)
await store.initialize()      # creates table + ivfflat index (idempotent)

# Write
await store.upsert(embeddings, tenant_id)

# Read
results = await store.query(
    vector=[0.1, 0.2, ...],
    top_k=10,
    tenant_id=tenant_id,
    corpus_ids=[],            # empty = all corpora
)
# returns: list[tuple[ChunkId, float]] ordered by descending score

# Delete
await store.delete([ChunkId("chunk-abc")], tenant_id)

# Health check
ok = await store.health()

await store.close()

Configuration

Parameter	Default	Description
`dsn`	—	asyncpg connection string
`table`	`"rag_vector_store"`	Table name
`dimension`	`1536`	Vector dimensionality
`min_connections`	`1`	Pool minimum
`max_connections`	`10`	Pool maximum

QdrantVectorStore

Qdrant-backed vector store. Uses a single collection with payload-based tenant isolation.

from rag_backends.vector.qdrant import QdrantVectorStore

store = QdrantVectorStore(
    url="http://localhost:6333",
    collection="rag_embeddings",  # default
)
await store.initialize(dimension=1536)  # creates collection (idempotent)

await store.upsert(embeddings, tenant_id)
results = await store.query(vector, top_k=10, tenant_id=tenant_id, corpus_ids=[])
await store.delete(chunk_ids, tenant_id)
await store.close()

Configuration

Parameter	Default	Description
`url`	—	Qdrant HTTP URL
`collection`	`"rag_embeddings"`	Collection name
`api_key`	`None`	Qdrant Cloud API key

WeaviateVectorStore

Weaviate v4 (async client). One collection (RagEmbeddings by default) holds all tenants; isolation is enforced via a tenant_id property filter on every read / delete. ACL labels are stored as a text[] property and pushed down via FilterExpr.

from rag_backends.vector.weaviate import WeaviateVectorStore
from rag_core.types import IndexHint

store = WeaviateVectorStore(http_host="localhost", http_port=8080)
await store.initialize(dimension=1536, hint=IndexHint(estimated_size=500_000))
# ... bulk_index / retrieve_ids / bulk_delete identical to other backends
await store.close()

Requires the optional extra: uv pip install 'rag-backends[weaviate]'.

Parameter	Default	Description
`http_host`	`"localhost"`	Weaviate HTTP host
`http_port`	`8080`	Weaviate HTTP port
`grpc_host`	(= http_host)	Weaviate gRPC host
`grpc_port`	`50051`	Weaviate gRPC port
`http_secure`	`False`	Use HTTPS
`grpc_secure`	`False`	Use TLS for gRPC
`api_key`	`None`	Weaviate Cloud API key
`collection`	`"RagEmbeddings"`	Collection name

IndexHint mapping (per ADR-0009):

Variant	Weaviate config
`flat`	`Configure.VectorIndex.flat()`
`ivfflat`	`Configure.VectorIndex.hnsw(max_connections=8)` (closest analogue)
`hnsw`	`Configure.VectorIndex.hnsw()`
`ivf_pq`	`Configure.VectorIndex.hnsw(quantizer=Quantizer.pq())`

PineconeVectorStore

Pinecone (serverless). Uses one namespace per tenant as the primary isolation primitive; the explicit tenant_id filter remains as defense-in-depth. Metadata: tenant_id, chunk_id, corpus_id, model, acl_labels (list); the filter translator emits MongoDB-style operator dicts ($eq / $in / $and / $or / $ne / $nin).

from rag_backends.vector.pinecone import PineconeVectorStore

store = PineconeVectorStore(api_key=os.environ["PINECONE_API_KEY"])
await store.initialize(dimension=1536)
# ... bulk_index / retrieve_ids / bulk_delete identical to other backends
await store.close()

Requires the optional extra: uv pip install 'rag-backends[pinecone]'.

Parameter	Default	Description
`api_key`	—	Pinecone API key (required)
`index_name`	`"rag-embeddings"`	Index name (created lazily)
`cloud`	`"aws"`	Serverless cloud (`aws` / `gcp` / `azure`)
`region`	`"us-east-1"`	Serverless region
`metric`	`"cosine"`	Distance metric for new indexes

IndexHint is recorded in logs but has no API effect — Pinecone manages the index implementation internally. Operators tune the pod-type out of band; the hint is advisory.

ElasticsearchDenseVectorStore

Elasticsearch 8.x dense_vector field + native kNN search. One index holds all tenants; isolation via a tenant_id keyword filter on every search and a delete_by_query on bulk delete.

from rag_backends.vector.elasticsearch import ElasticsearchDenseVectorStore
from rag_core.types import IndexHint

store = ElasticsearchDenseVectorStore(
    hosts=["http://localhost:9200"],
    index="rag-embeddings",
)
await store.initialize(dimension=1536, hint=IndexHint(estimated_size=2_000_000))
# ... bulk_index / retrieve_ids / bulk_delete identical to other backends
await store.close()

Requires the optional extra: uv pip install 'rag-backends[elasticsearch]'.

Parameter	Default	Description
`hosts`	—	List of ES host URLs
`index`	`"rag-embeddings"`	Index name (created lazily)
`api_key`	`None`	Elastic Cloud API key
`basic_auth`	`None`	`(user, password)`
`request_timeout`	`30.0`	Per-request seconds
`similarity`	`"cosine"`	kNN similarity (`cosine` / `l2_norm` / `dot_product`)

IndexHint mapping (per ADR-0009):

Variant	Mapping
`flat`	`dense_vector` with `index=false` — exact scan
`ivfflat`	`dense_vector` with `index=false` (ES has no IVF-flat)
`hnsw`	`dense_vector` `index_options={type:hnsw, m:16, ef_construction:100}`
`ivf_pq`	`dense_vector` `index_options={type:hnsw, m:32, ef_construction:200}`

RedisCache

Redis-backed cache implementing the Cache SPI.

from rag_backends.cache.redis import RedisCache

cache = RedisCache(
    url="redis://localhost:6379/0",
    prefix="rag:",         # key namespace
    default_ttl=3600,      # seconds (None = no expiry)
)

await cache.set("key", b"value", ttl_seconds=300)
data = await cache.get("key")    # bytes or None
exists = await cache.exists("key")
await cache.delete("key")
await cache.close()

Configuration

Parameter	Default	Description
`url`	—	Redis URL
`prefix`	`"rag:"`	Key prefix
`default_ttl`	`None`	Default TTL in seconds
`socket_timeout`	`5.0`	Command timeout

RedisRetrievalCache / RedisAnswerCache

Redis implementations of the typed RetrievalCache / AnswerCache SPIs — the L1 (shared, exact) tier of the Step 4.1 tiered cache. They are injected as the l1= argument of rag_cache.TieredRetrievalCache / TieredAnswerCache (which own the in-memory L0 + similarity L2 tiers and the graceful degradation).

from rag_backends import RedisRetrievalCache, RedisAnswerCache
from rag_cache import TieredRetrievalCache

cache = TieredRetrievalCache(
    l1=RedisRetrievalCache(url="redis://localhost:6379/0"),
)

Both serialise + tenant-namespace their keys and support tag-set invalidation the in-memory tiers can't: RedisRetrievalCache.invalidate_corpus drops every entry tagged with a ChunkRef.corpus_id, and RedisAnswerCache.invalidate_policy drops every entry tagged with a policy_version. Full API + config in reference/cache.md.

S3Storage

S3-compatible object storage. Works with AWS S3 and MinIO.

from rag_backends.storage.s3 import S3Storage

# MinIO (local dev)
store = S3Storage(
    bucket="rag-dev",
    endpoint_url="http://localhost:9000",
    aws_access_key_id="minioadmin",
    aws_secret_access_key="minioadmin",
)

# AWS S3
store = S3Storage(bucket="my-rag-bucket", region="us-east-1")

await store.put("docs/a.txt", b"content", content_type="text/plain")
data = await store.get("docs/a.txt")
await store.delete("docs/a.txt")
exists = await store.exists("docs/a.txt")
async for key in store.list_keys("docs/"):
    print(key)

Configuration

Parameter	Default	Description
`bucket`	—	S3 bucket name (must exist)
`region`	`"us-east-1"`	AWS region
`endpoint_url`	`None`	Override for MinIO/custom endpoints
`aws_access_key_id`	`None`	Explicit credentials
`aws_secret_access_key`	`None`	Matching secret
`prefix`	`""`	Object key prefix

LocalFileStorage

Filesystem-based storage. Useful for local dev and CI without cloud credentials.

from rag_backends.storage.local import LocalFileStorage

store = LocalFileStorage(root="/tmp/rag-dev")

await store.put("docs/hello.txt", b"hello")
data = await store.get("docs/hello.txt")

Keys are mapped to filesystem paths relative to root. Path-traversal is blocked — any key that resolves outside root raises ValueError.

Internals

PgVectorStore schema

CREATE TABLE rag_vector_store (
    tenant_id   TEXT        NOT NULL,
    chunk_id    TEXT        NOT NULL,
    corpus_id   TEXT        NOT NULL DEFAULT '',
    model       TEXT        NOT NULL,
    vector      vector(N),           -- N = configured dimension
    dimension   INT         NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (tenant_id, chunk_id)
);
CREATE INDEX rag_vector_store_vec_idx
    ON rag_vector_store USING ivfflat (vector vector_cosine_ops)
    WITH (lists = 100);

The table is separate from the application schema defined in infra/postgres/init.sql to keep the VectorStore decoupled.

ElasticsearchKeywordStore

Step 2.3 — production BM25 keyword backend backed by an Elasticsearch cluster's native scoring. Implements both KeywordRetrievalBackend (read) and KeywordIndexBackend (write); use together as KeywordStore.

from rag_backends.keyword.elasticsearch import ElasticsearchKeywordStore

store = ElasticsearchKeywordStore(
    hosts=["http://localhost:9200"],
    index="rag-keyword",
    field_boosts={"title": 3.0, "section_path": 2.0, "body": 1.0},
)
await store.initialize()
await store.bulk_index(ctx, chunks)
refs = await store.retrieve_ids(ctx, "quick brown fox", top_k=10, corpus_ids=[])

Constructor parameters

Parameter	Default	Purpose
`hosts`	—	Elasticsearch host URLs (required)
`index`	`"rag-keyword"`	Index name (created lazily on `initialize()`)
`api_key`	`None`	Elastic Cloud API key
`basic_auth`	`None`	`(user, password)` tuple
`request_timeout`	`30.0`	Per-request timeout in seconds
`field_boosts`	`{"title": 3.0, "section_path": 2.0, "body": 1.0}`	Per-field BM25 score multiplier; only listed fields are searched
`analyzer`	`"standard"`	Analyzer applied to `body` / `title` / `section_path`; set to `"english"` for stemming

Schema

A single index stores every tenant. Isolation is enforced by a tenant_id keyword filter on every search and a composite document ID ("<tenant>::<chunk>") on every write.

Field	Type	Purpose
`tenant_id` / `chunk_id` / `document_id` / `corpus_id`	`keyword`	Exact-match filters
`acl_labels`	`keyword` (array)	Push-down via `Eq` / `AnyIn`
`trust_level`	`keyword`	Push-down for prompt-injection defense
`body` / `title` / `section_path`	`text` (operator-selected analyzer)	BM25-scored
`position`	`integer`	Stored for hydrate
`metadata`	`object` (`enabled: false`)	Free-form parser/enricher metadata; stored, not indexed

FilterExpr push-down

retrieve_ids accepts an optional filters: FilterExpr | None that the translator (rag_backends.keyword._filter_es.translate) renders into an Elasticsearch query-DSL bool.filter clause. Supported fields: tenant_id, chunk_id, document_id, corpus_id, acl_labels, trust_level. Unsupported fields raise RetrievalError — silent push-down loss would mean an ACL filter the caller asked for silently became a no-op.

Per-field boosting

Per-field boosting is a backend configuration knob, not a per-call SPI parameter. Boosts are applied to the multi_match query's fields array on every retrieval (e.g. ["title^3.0", "section_path^2.0", "body^1.0"]). The multi_match type is best_fields, so the score is the max across fields rather than a sum — a strong title hit dominates a weak body hit rather than being washed out.

TantivyKeywordStore

Step 2.3 — in-process BM25 backend powered by tantivy (Rust core with Python bindings). Targets laptop / dev-loop usage, CI runs where spinning up Docker is overkill, and air-gapped deployments. Shares the same SPI surface as ElasticsearchKeywordStore so swapping between the two is a config change.

from rag_backends.keyword.tantivy import TantivyKeywordStore

store = TantivyKeywordStore(
    field_boosts={"title": 3.0, "section_path": 2.0, "body": 1.0},
)                               # RAM index
# or:
store = TantivyKeywordStore(path="/var/data/keyword")  # persistent

await store.initialize()
await store.bulk_index(ctx, chunks)
refs = await store.retrieve_ids(ctx, "quick brown fox", top_k=10, corpus_ids=[])

Constructor parameters

Parameter	Default	Purpose
`path`	`None`	Filesystem directory for a persistent index; `None` builds a RAM index (the right default for tests and ephemeral CLI usage)
`field_boosts`	`{"title": 3.0, "section_path": 2.0, "body": 1.0}`	Per-field BM25 score multiplier (passed directly to tantivy's parser)
`heap_size`	`15_000_000`	tantivy writer heap (bytes); bump for bulk loads
`analyzer`	`"default"`	tantivy tokenizer (`"default"` = Unicode + lowercase; `"en_stem"` for English stemming)

Schema + sync wrapping

tantivy operations are synchronous; the SPI is async, so every call is wrapped in asyncio.to_thread to keep the event loop responsive. Tenant isolation matches the Elasticsearch backend exactly: a single index, a tenant_id filter clause on every search, and a composite _doc_id ("<tenant>::<chunk>") per write. bulk_index is idempotent — a repeat indexing of the same chunk first deletes the prior composite ID before adding the new doc.

FilterExpr push-down

retrieve_ids accepts an optional filters: FilterExpr | None that the translator (rag_backends.keyword._filter_tantivy.translate) renders into a tantivy Query composed via boolean_query. Supported fields: tenant_id, chunk_id, document_id, corpus_id, acl_labels, trust_level. Not(x) is paired with all_query() so it composes stand-alone — tantivy's MustNot is only meaningful alongside a positive clause. Unsupported fields raise RetrievalError.

QdrantVectorStore point structure

Each Qdrant point carries:

id      = uuid5(NAMESPACE_URL, f"{tenant_id}/{chunk_id}")  # stable, idempotent
vector  = embedding.vector
payload = {tenant_id, chunk_id, corpus_id, model}

Corpus scoping

rag_core.types.Embedding carries an optional corpus_id (the ingest pipeline stamps it from the source Chunk, since the embedder SPI has no corpus). Every VectorStore persists it ("" when unset) and honours the corpus_ids argument to retrieve_ids: a non-empty list restricts results to those corpora, an empty list means all corpora. Each returned ChunkRef carries its corpus_id. This closes the gap originally deferred in ADR-0004 §3.

Neo4jGraphStore

Step 2.4 — knowledge-graph backend on top of the official neo4j async Python driver (Bolt + Cypher). Implements both GraphRetrievalBackend (read / traverse) and GraphIndexBackend (write); use the composite GraphStore.

from rag_backends.graph.neo4j import Neo4jGraphStore

store = Neo4jGraphStore(
    uri="bolt://localhost:7687",
    auth=("neo4j", "password"),
    database="neo4j",
)
await store.initialize()
await store.upsert_node(ctx, "n1", ["Document"], {"title": "Hello"})
await store.upsert_edge(ctx, "n1", "n2", "LINKS", {"acl_labels": ["public"]})
neighbors = await store.expand(ctx, ["n1"], hops=2)
await store.close()

Constructor parameters

Parameter	Default	Purpose
`uri`	—	Bolt URI (e.g. `bolt://host:7687`)
`auth`	`None`	`(user, password)` tuple
`database`	`None`	Neo4j database name
`array_fields`	`frozenset({"acl_labels"})`	Property names the Cypher translator treats as Cypher list values
`**driver_kwargs`	—	Passed through to `AsyncGraphDatabase.driver()` (encryption, trust, etc.)

Schema

Both Cypher backends store the SPI's node + edge model the same way:

Nodes are MERGE'd by (id, tenant_id) so the same external node_id in two tenants is two distinct Neo4j nodes.
The supplied labels list is stored as a node property labels: list[str] rather than as Cypher labels — dynamic labels require either APOC (Neo4j-only) or unsafe string interpolation; the property representation round-trips through query() / expand() portably.
Every relationship carries tenant_id so multi-hop traversal can filter cross-tenant leaks at every edge, not just at the endpoints.
The relationship type is interpolated into Cypher (Cypher has no relationship-type parameter) — values are validated against _REL_TYPE_RE and backtick-quoted; arbitrary strings are rejected with RetrievalError.

`expand()` semantics

Multi-hop neighbor expansion is the canonical traversal primitive (Step 2.4). Cypher composition:

MATCH path = (seed:RagNode {tenant_id: $tenant_id})-[*1..N]->(neighbor:RagNode {tenant_id: $tenant_id})
WHERE seed.id IN $seed_ids
  AND NOT neighbor.id IN $seed_ids
  AND ALL(_r IN relationships(path) WHERE _r.tenant_id = $tenant_id)
  -- + edge_filter on every relationship along the path
  -- + node_filter on the final neighbor
WITH neighbor, length(path) AS distance, [_r IN relationships(path) | type(_r)] AS path_rels
WITH neighbor, min(distance) AS min_distance, collect({...}) AS hits
RETURN neighbor.id, neighbor.labels, properties(neighbor), min_distance, path
ORDER BY min_distance ASC, node_id ASC
LIMIT $limit

The NoopGraphStore.expand reference oracle is the conformance target — backend results agree on (node_id, distance, path) for every input.

MemgraphGraphStore

Memgraph speaks the Bolt protocol and accepts Cypher, so it reuses the official neo4j Python driver — the wire format is identical, only the server differs. Constructor parameters, schema, and expand() semantics match Neo4jGraphStore; the class exists as a thin subclass that reports "memgraph" in logs so operators can tell the two apart.

from rag_backends.graph.memgraph import MemgraphGraphStore

store = MemgraphGraphStore(
    uri="bolt://localhost:7687",
    auth=("memgraph", "memgraph"),
)

NetworkXGraphStore

Step 2.4 — in-process pure-Python knowledge-graph backend on top of networkx.MultiDiGraph. Targets laptop / dev-loop usage, CI runs, air-gapped deployments, and GraphRAG (Phase 2.9) prototyping.

from rag_backends.graph.networkx import NetworkXGraphStore

store = NetworkXGraphStore()       # in-process MultiDiGraph
await store.initialize()
await store.upsert_node(ctx, "n1", ["Document"], {"title": "Hello"})
await store.upsert_edge(ctx, "n1", "n2", "LINKS", {"acl_labels": ["public"]})
neighbors = await store.expand(ctx, ["n1"], hops=2)

Differences from the Cypher backends

No query() — NetworkX has no native query language; the method raises RetrievalError with a clear pointer to expand(). Power users can reach the underlying MultiDiGraph via the graph property for direct algorithm calls (community detection, shortest path, centrality — all the NetworkX algorithms are available).
No driver — everything lives in process memory.
Same expand() semantics — the BFS walks the underlying MultiDiGraph with the same edge-filter / node-filter rules as the NoopGraphStore reference oracle, producing identical NeighborResult rows for the same inputs.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Backend Reference — `rag-backends`

Overview

PgVectorStore

Configuration

QdrantVectorStore

Configuration

WeaviateVectorStore

PineconeVectorStore

ElasticsearchDenseVectorStore

RedisCache

Configuration

RedisRetrievalCache / RedisAnswerCache

S3Storage

Configuration

LocalFileStorage

Internals

PgVectorStore schema

ElasticsearchKeywordStore

Constructor parameters

Schema

FilterExpr push-down

Per-field boosting

TantivyKeywordStore

Constructor parameters

Schema + sync wrapping

FilterExpr push-down

QdrantVectorStore point structure

Corpus scoping

Neo4jGraphStore

Constructor parameters

Schema

`expand()` semantics

MemgraphGraphStore

NetworkXGraphStore

Differences from the Cypher backends

FilesExpand file tree

backends.md

Latest commit

History

backends.md

File metadata and controls

Backend Reference — rag-backends

Overview

PgVectorStore

Configuration

QdrantVectorStore

Configuration

WeaviateVectorStore

PineconeVectorStore

ElasticsearchDenseVectorStore

RedisCache

Configuration

RedisRetrievalCache / RedisAnswerCache

S3Storage

Configuration

LocalFileStorage

Internals

PgVectorStore schema

ElasticsearchKeywordStore

Constructor parameters

Schema

FilterExpr push-down

Per-field boosting

TantivyKeywordStore

Constructor parameters

Schema + sync wrapping

FilterExpr push-down

QdrantVectorStore point structure

Corpus scoping

Neo4jGraphStore

Constructor parameters

Schema

expand() semantics

MemgraphGraphStore

NetworkXGraphStore

Differences from the Cypher backends

Backend Reference — `rag-backends`

`expand()` semantics