Installation

NexRAG is on PyPI. Install the core plus the extras you need.

# Core + OpenAI + PDF (most common setup)
pip install "nexrag[openai,pdf]"

# Core only — bring your own adapters
pip install nexrag

# Everything
pip install "nexrag[all]"

Configuration

NexRAG is configured entirely through a YAML file — by default nexrag.yaml in your working directory.

# nexrag.yaml — full config reference (v0.5.0)

mode: sync             # sync (default) | async — enables native async pipeline

ingestion:
  loader:
    type: pdf             # auto | pdf | txt | custom
    include_metadata: true      # extract PDF metadata fields (default: true)
    # metadata_fields: [title, author, creation_date]  # whitelist — omit for all 8

  chunker:
    strategy: recursive   # recursive | fixed | token | sentence | sentence_window | markdown | code | semantic | proposition | custom
    chunk_size: 512
    chunk_overlap: 64
    min_chunk_size: 50
    # separator: "\n\n"  # top-level split character (default: double newline)
    # semantic / proposition take their OWN nested component, resolved
    # independently of the pipeline models (cheap model for chunking):
    # embedder: { provider: ollama, model: nomic-embed-text }   # semantic
    # llm:      { provider: gemini, model: gemini-2.5-flash }    # proposition

  embedder:
    provider: openai      # openai | gemini | huggingface | ollama | custom
    model: text-embedding-3-small
    api_key: ${OPENAI_API_KEY}
    # base_url: https://...  # override API base URL (OpenAI-compatible endpoints)
    # max_retries: 2         # embedding API retry attempts (default: 2)

  vector_db:
    provider: chroma           # chroma | pinecone | custom
    default_collection: documents
    on_conflict: overwrite       # overwrite | skip | append
    upsert_batch_size: 500        # max chunks per ChromaDB upsert call
    max_retries: 3               # connection retry attempts
    retry_delay: 1.0             # initial backoff in seconds (exponential)
    collections:
      documents:
        mode: persistent          # persistent | memory | server
        path: ./.nexrag/chroma
        # host: localhost   # server mode only
        # port: 8000        # server mode only

query:
  embedder: inherit             # reuse ingestion embedder
  retriever:
    provider: dense             # dense | bm25 | hybrid
    top_k: 5
    score_threshold: 0.0
    # hybrid-only options:
    # alpha: 0.7          # 0=pure BM25, 1=pure dense (default 0.7)
    # sparse_top_k: 50    # BM25 candidates before fusion
    # sparse:
    #   provider: bm25    # bm25 | custom
  # reranker:                       # optional — omit to skip reranking
  #   provider: cohere               # cohere | cross_encoder | custom
  #   model: rerank-english-v3.0
  #   api_key: ${COHERE_API_KEY}
  #   top_n: 5                       # chunks passed to LLM after reranking
  # max_query_length: 8000           # reject queries longer than this (chars); 0 = disabled
  llm:
    provider: openai          # openai | anthropic | gemini | ollama | custom
    model: gpt-4o
    api_key: ${OPENAI_API_KEY}
    # base_url: https://...  # override API base URL
    temperature: 0.2
    max_tokens: 1024
    # timeout: 30            # request timeout in seconds (default: 30)
    # max_retries: 2         # LLM API retry attempts (default: 2)
  prompt:
    system: |
      You are a helpful assistant. Answer using only
      the provided context. If unknown, say so.
    # context_format: numbered  # numbered | labeled | plain (how chunks are presented)
    # template: default          # default | qa | summarize | custom
  # cache:                            # optional query-result cache — disabled by default
  #   enabled: true
  #   backend: memory                 # memory | custom
  #   strategy: exact                 # exact | semantic (semantic requires a custom backend)
  #   max_size: 1000                  # max cached entries (memory backend)
  #   ttl_seconds: 300                # entry time-to-live
  # session:                          # optional multi-turn conversation memory — disabled by default
  #   enabled: true
  #   backend: memory                 # memory | custom
  #   session_ttl_seconds: 1800       # idle session lifetime
  #   persist: true                   # false = privacy mode (history never stored)
  #   context_strategy:               # how much history is injected into the prompt
  #     type: window                  # window | token_budget | custom
  #     max_history_turns: 6          # window: keep the last N turns
  #     max_tokens: 2000              # token_budget: fit history within N tokens
  # rate_limit:                       # optional client-side throttle — disabled by default
  #   enabled: true
  #   requests_per_minute: 60
  #   burst: 10                       # allow short bursts up to this many requests

guardrails:                  # optional security layer — every chain is disabled by default
  input:                       # runs on the user query
    enabled: true
    policy: fail_closed        # fail_open (default) | fail_closed
    guards:
      - type: access_control   # per-request retrieval filter from auth_context
        params: { mapping: { tenant: tenant } }
      - type: prompt_injection  # heuristic injection/jailbreak detection
  retrieved:                   # runs on each retrieved chunk (an injection vector)
    enabled: true
    guards:
      - type: pii               # redact PII before it reaches the prompt
  output:                      # runs on the LLM answer (buffers when streaming)
    enabled: true
    guards:
      - type: groundedness      # cheap lexical-overlap check vs context
  # ingestion:                  # runs on document text before chunking
  #   enabled: true
  #   guards: [ { type: pii, params: { mode: redact } } ]

observability:
  enabled: true
  service_name: my-rag-app
  signals:
    metrics: true
    traces:  true
    logs:    true
  exporters:
    prometheus:
      enabled: true      # pull → GET http://host:9464/metrics
      port:    9464
    otlp:
      enabled:   false     # push → OTel Collector / Grafana Alloy
      endpoint:  ${OTEL_EXPORTER_OTLP_ENDPOINT:-http://localhost:4317}
      protocol:  grpc      # grpc | http
      insecure:  true
    console:
      enabled: false      # debug exporter
  metrics:
    pricing:             # for nexrag.llm.cost_per_query_usd
      gpt-4o:      { input: 0.0025,  output: 0.01   }
      gpt-4o-mini: { input: 0.00015, output: 0.0006 }
  evaluations:           # optional LLM-as-judge tier — off by default
    enabled:      false
    sample_rate:  0.1
    faithfulness:
      enabled: true
      llm:
        provider: openai
        model:    gpt-4o-mini
        api_key:  ${OPENAI_API_KEY}

Quickstart

Ingest a document and query it in under ten lines of Python.

import os
from nexrag import NexRAG

os.environ["OPENAI_API_KEY"] = "sk-..."  # or set in your shell

pipeline = NexRAG.from_config("nexrag.yaml")

from pathlib import Path

# Ingest — loaders expect bytes or str, not a file path
result = pipeline.ingest(Path("contracts/agreement.pdf").read_bytes())
print(f"Ingested {result.documents_loaded} doc")
print(f"Wrote {result.chunks_written} chunks in {result.latency_ms:.1f}ms")

# Query
result = pipeline.query("What are the termination clauses?")
print(result.answer)

for source in result.sources:
    print(f"  [{source.rank}] score={source.score:.3f}  {source.source}")

Ingestion

pipeline.ingest() accepts raw bytes (for binary formats like PDF) or a plain string (for text content). Loaders are converter-only since v0.3.0 — they never open files themselves. Read the file first, then pass the content. pipeline.ingest_documents() accepts pre-built Document objects.

Read the file to bytes or text first, then pass the content to ingest(). Use the metadata argument to attach a source name so the result is traceable.

from pathlib import Path

# PDF — pass bytes
result = pipeline.ingest(
    Path("contracts/agreement.pdf").read_bytes(),
    metadata={"source": "agreement.pdf"},
)

# Text file — pass string
result = pipeline.ingest(
    Path("notes/onboarding.txt").read_text(),
    metadata={"source": "onboarding.txt"},
)

import boto3

s3 = boto3.client("s3")
pdf_bytes = s3.get_object(Bucket="bucket", Key="report.pdf")["Body"].read()

result = pipeline.ingest(pdf_bytes, metadata={"source": "report.pdf"})

text = "Alice joined Acme Corp in 2019..."
# pass a (text, source_name) tuple — second element becomes metadata["source"]
result = pipeline.ingest((text, "alice_bio"))

# or use the metadata argument explicitly
result = pipeline.ingest(text, metadata={"source": "alice_bio"})

Use ingest_documents() when your data comes from a database or API — anywhere a file path doesn't make sense.

from nexrag.core.models.document import Document

docs = [
    Document(
        content="NexRAG is a framework-agnostic RAG pipeline SDK.",
        metadata={"source": "internal-wiki", "team": "platform"},
    ),
]
result = pipeline.ingest_documents(docs)

Use ingest_batch() to ingest multiple sources in sequence. One IngestionResult is returned per source, in the same order. Each element in the list is passed to the loader — read bytes/text before passing.

from pathlib import Path

files = ["contracts/q1.pdf", "contracts/q2.pdf", "contracts/q3.pdf"]
results = pipeline.ingest_batch([Path(f).read_bytes() for f in files])
for r in results:
    print(f"{r.chunks_written} chunks from pipeline_id={r.pipeline_id}")

Ingestion is idempotent by default. Each chunk is fingerprinted with SHA-256. The on_conflict setting controls what happens when a chunk already exists.

Route ingestions and queries to named collections declared in YAML. Each collection is an independent namespace in ChromaDB.

# nexrag.yaml
vector_db:
  default_collection: documents
  collections:
    documents:
      mode: persistent
      path: ./.nexrag/documents
    resumes:
      mode: persistent
      path: ./.nexrag/resumes

from pathlib import Path

# ingest into a named collection
results = pipeline.ingest_batch([
    Path("resumes/alice.pdf").read_bytes(),
    Path("resumes/bob.pdf").read_bytes(),
], collection="resumes")

# query a named collection
result = pipeline.query("Candidate with Python experience", collection="resumes")
print(result.collection_used)   # → "resumes"

Querying

All query parameters can be overridden per-call without changing the YAML.

result = pipeline.query("What is the refund policy?")

# Override retrieval params
result = pipeline.query("Summarize the contract", top_k=10)
result = pipeline.query("Exact clause text", score_threshold=0.75)

# Query a specific collection
result = pipeline.query("Refund policy", collection="terms_of_service")

# Filter by metadata
result = pipeline.query(
    "What changed in 2024?",
    metadata_filter={"year": 2024},
)

Use provider: hybrid to fuse keyword-based (BM25) and vector-based (dense) search. alpha controls the balance — 1.0 is pure dense, 0.0 is pure BM25. Requires pip install "nexrag[bm25]".

# nexrag.yaml
query:
  retriever:
    provider: hybrid
    top_k: 10
    alpha: 0.7          # 0=pure BM25, 1=pure dense (default 0.7)
    sparse_top_k: 50   # BM25 candidate pool before fusion
    sparse:
      provider: bm25

Add an optional reranker stage between the retriever and the LLM. The typical pattern is to retrieve a large candidate set with top_k and rerank down to a small top_n for the LLM — combining recall from retrieval with precision from reranking.

# nexrag.yaml
query:
  retriever:
    provider: dense
    top_k: 50            # retrieve many candidates
  reranker:
    provider: cohere      # cohere | cross_encoder | custom
    model: rerank-english-v3.0
    api_key: ${COHERE_API_KEY}
    top_n: 5              # only 5 chunks reach the LLM

For a local, offline reranker use provider: cross_encoder with a sentence-transformers model. Requires pip install "nexrag[cross-encoder]".

reranker:
  provider: cross_encoder
  model: cross-encoder/ms-marco-MiniLM-L-6-v2
  top_n: 5
  # device: cuda   # optional — CPU by default

Async & streaming

NexRAG ships two pipeline modes. Sync (default) — straightforward, thread-safe, works everywhere. Async — native async/await with parallel batch embedding and live token streaming.

Add one line to your YAML. No application code changes required.

# nexrag.yaml
mode: async   # async | sync (default)

ingestion:
  embedder:
    provider: openai
    model: text-embedding-3-small
    api_key: ${OPENAI_API_KEY}
    batch_size: 50   # chunks per concurrent API call batch

stream_query() works with any sync caller — no event loop required. Tokens are yielded as they arrive from the LLM. The final item in the stream is always a RunMetrics object — use isinstance to separate tokens from metrics.

from nexrag import NexRAG, RunMetrics

pipeline = NexRAG.from_config("nexrag.yaml")

metrics = None
for item in pipeline.stream_query("Summarize the key obligations"):
    if isinstance(item, RunMetrics):
        metrics = item
    else:
        print(item, end="", flush=True)
print()
print(f"{metrics.total_latency_ms:.0f}ms — {metrics.chunks_retrieved} chunks")

# Runtime overrides work identically to query()
for item in pipeline.stream_query(
    "What are the termination clauses?",
    top_k=10,
    score_threshold=0.6,
):
    if not isinstance(item, RunMetrics):
        print(item, end="", flush=True)

astream_query() is the async variant. With mode: async configured, pre-LLM stages use native async clients and tokens stream live. With mode: sync, the sync stream runs in a thread pool — non-blocking for the event loop, but tokens buffer until the stream completes in the thread. Like the sync variant, RunMetrics is the final item yielded.

import asyncio
from nexrag import NexRAG, RunMetrics

pipeline = NexRAG.from_config("nexrag.yaml")

async def main():
    metrics = None
    async for item in pipeline.astream_query(
        "What changed in 2024?",
        metadata_filter={"year": 2024},
    ):
        if isinstance(item, RunMetrics):
            metrics = item
        else:
            print(item, end="", flush=True)
    print(f"\n{metrics.total_latency_ms:.0f}ms")

asyncio.run(main())

Use async_query() for non-blocking queries, or astream_query() for SSE / streaming responses. Configure mode: async for true async parallelism at every stage.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from nexrag import NexRAG

app = FastAPI()
pipeline = NexRAG.from_config("nexrag.yaml")  # mode: async in nexrag.yaml

@app.get("/ask")
async def ask(q: str):
    result = await pipeline.async_query(q)
    return {"answer": result.answer, "sources": [s.source for s in result.sources]}

@app.get("/stream")
async def stream(q: str):
    async def gen():
        async for token in pipeline.astream_query(q):
            yield token
    return StreamingResponse(gen(), media_type="text/plain")

Secret management

NexRAG supports ${ENV_VAR} substitution in YAML before any parsing occurs.

# Required — fails at startup if not set
api_key: ${OPENAI_API_KEY}

# Optional with empty-string fallback
base_url: ${OPENAI_BASE_URL:-}

# Default value
model: ${LLM_MODEL:-gpt-4o}

# Load from .env before from_config
from dotenv import load_dotenv
load_dotenv()
pipeline = NexRAG.from_config("nexrag.yaml")

Custom components

Any pipeline stage can be replaced. Implement the interface from nexrag.core.interfaces, then declare it in YAML with provider: custom and a dotted class: path.

# myapp/embedders.py
from nexrag.core.interfaces.embedder import BaseEmbedder

class SentenceTransformerEmbedder(BaseEmbedder):
    def __init__(self, model: str = "all-MiniLM-L6-v2") -> None:
        from sentence_transformers import SentenceTransformer
        self._model = SentenceTransformer(model)
        self._model_name = model

    @property
    def model_name(self) -> str: return self._model_name

    @property
    def dimensions(self) -> int:
        return self._model.get_sentence_embedding_dimension()

    def embed(self, texts: list[str]) -> list[list[float]]:
        return self._model.encode(texts).tolist()

    def embed_query(self, text: str) -> list[float]:
        return self._model.encode([text])[0].tolist()

# nexrag.yaml
ingestion:
  embedder:
    provider: custom
    class:    myapp.embedders.SentenceTransformerEmbedder
    params:
      model:  all-MiniLM-L6-v2

The same pattern works for every stage: loader, chunker, sanitizer, vector_db, retriever, reranker, prompt_builder, llm, observer, and evaluator. For a custom observer use the class_path key under observability.

Testing

Use in-memory ChromaDB and mock embedders/LLMs to test without any API calls.

# nexrag.test.yaml
vector_db:
  provider: chroma
  default_collection: test
  collections:
    test:
      mode: memory   # ephemeral, no disk I/O

from nexrag.core.interfaces.embedder import BaseEmbedder
from nexrag.core.interfaces.llm import BaseLLM

class FixedEmbedder(BaseEmbedder):
    DIM = 8

    @property
    def model_name(self) -> str: return "fixed-test"

    @property
    def dimensions(self) -> int: return self.DIM

    def embed(self, texts):
        return [[float(i % self.DIM) for i in range(self.DIM)] for _ in texts]

    def embed_query(self, text):
        return [float(i % self.DIM) for i in range(self.DIM)]


class EchoLLM(BaseLLM):
    # generate() returns (answer, token_usage) — token_usage is None when unknown
    def generate(self, prompt: str): return ("test answer", None)
    def stream(self, prompt): yield "test answer"

API reference

Return types for the two main pipeline methods.

Returned by pipeline.ingest(), pipeline.ingest_documents(), and each element of pipeline.ingest_batch().

Returned by pipeline.query() and pipeline.async_query(). Not returned by streaming calls.

Attached to PipelineResult.metrics and IngestionResult.metrics. Also yielded as the final item from stream_query() and astream_query() — use isinstance(item, RunMetrics) to detect it in a stream.

result = pipeline.query("What are the termination clauses?")
m = result.metrics
print(f"Total: {m.total_latency_ms:.0f}ms")
print(f"Stages: {m.stage_latencies}")
print(f"Tokens: {m.token_usage.total_tokens if m.token_usage else 'n/a'}")

Each element in result.sources.

Available when query.session.enabled: true. Each method takes a caller-supplied session_id (any opaque string — chat thread, socket id, etc.). History is injected into the prompt only; retrieval always uses the current query alone. Calling these with sessions disabled raises RuntimeError.

r1 = pipeline.query_session("What is NexRAG?", session_id="chat-42")
r2 = pipeline.query_session("And who maintains it?", session_id="chat-42")  # sees turn 1
pipeline.clear_session("chat-42")  # forget the conversation

Loaders

A loader converts raw data into a list of Document objects. Since v0.3.0, loaders are converter-only — they accept bytes or str only. File I/O is the caller's responsibility: read the file to bytes or text first, then pass it to pipeline.ingest().

The default loader. Inspects the MIME type or file extension of the source and dispatches to the appropriate loader automatically. Use when you want to ingest mixed file types without changing YAML.

# nexrag.yaml
ingestion:
  loader:
    type: auto   # default — detects pdf, txt, and more

Extracts text content and up to 8 document metadata fields from PDF files using pypdf. Requires pip install "nexrag[pdf]".

Extracted metadata fields: title, author, subject, keywords, creator, producer, creation_date, modification_date.

# nexrag.yaml
ingestion:
  loader:
    type: pdf
    include_metadata: true          # attach metadata to each chunk (default: true)
    # metadata_fields: [title, author]  # whitelist specific fields; omit for all 8

Wraps a plain string into a single Document. Use for ingesting strings from a database, API, or in-memory content.

# nexrag.yaml
ingestion:
  loader:
    type: txt

# runtime: pass a (text, source_name) tuple
result = pipeline.ingest(("NexRAG is a RAG pipeline SDK.", "intro"))

Implement BaseLoader to load any format not covered by the built-in loaders (DOCX, HTML, Excel, database rows, etc.).

from nexrag.core.interfaces.loader import BaseLoader
from nexrag.core.models.document import Document

class MarkdownLoader(BaseLoader):
    def load(self, source: str) -> list[Document]:
        # source is a markdown string
        return [Document(content=source, metadata={"source": "markdown"})]

# nexrag.yaml
ingestion:
  loader:
    type: custom
    class: myapp.loaders.MarkdownLoader

Chunkers

A chunker splits Document objects into smaller Chunk objects suitable for embedding. Chunk boundaries significantly affect retrieval quality — smaller chunks give precise matches, larger chunks provide more context.

The default chunker. Splits text recursively on a priority list of separators (\n\n, \n, . , ) until each chunk is under chunk_size. Preserves natural paragraph and sentence boundaries where possible.

# nexrag.yaml
ingestion:
  chunker:
    strategy: recursive   # the default; see all strategies below
    chunk_size: 512       # max characters per chunk
    chunk_overlap: 64    # overlap between consecutive chunks
    min_chunk_size: 50   # discard chunks shorter than this

Splits text into fixed-length character windows with a configurable overlap (sliding window). No separator logic — each chunk is exactly chunk_size characters, except the last which may be shorter. Useful when you need strict, predictable chunk boundaries.

# nexrag.yaml
ingestion:
  chunker:
    strategy: fixed
    chunk_size: 512        # characters per chunk
    chunk_overlap: 64     # overlap between consecutive chunks
    min_chunk_size: 50    # discard chunks shorter than this

v0.4.0 ships a full suite of structure- and meaning-aware chunkers. All accept the common chunk_size / chunk_overlap / min_chunk_size knobs; strategy-specific options go under params.

Nested component configs. The semantic and proposition chunkers resolve their own embedder/LLM — a full component block, resolved independently of the pipeline's main models. Run a cheap model for chunking and a strong one for generation:

# nexrag.yaml — semantic chunking with an independent (cheap) embedder
ingestion:
  chunker:
    strategy: semantic
    embedder:                        # resolved independently of ingestion.embedder
      provider: ollama
      model: nomic-embed-text
    params:
      threshold: 0.5               # or omit for percentile-based breakpoints
      buffer_size: 1

# proposition chunking uses a nested `llm:` block instead:
#   strategy: proposition
#   llm: { provider: gemini, model: gemini-2.5-flash, api_key: ${GOOGLE_API_KEY} }

Implement BaseChunker for semantic chunking, fixed-size chunking, or any domain-specific splitting strategy (e.g. split legal contracts by clause, code files by function).

from nexrag.core.interfaces.chunker import BaseChunker
from nexrag.core.models.document import Document
from nexrag.core.models.chunk import Chunk

class ParagraphChunker(BaseChunker):
    def chunk(self, document: Document) -> list[Chunk]:
        paragraphs = [p.strip() for p in document.content.split("\n\n") if p.strip()]
        return [
            Chunk(
                text=p,
                chunk_index=i,
                total_chunks=len(paragraphs),
                parent_doc_id=document.doc_id,
                metadata={**document.metadata, "paragraph_index": i},
            )
            for i, p in enumerate(paragraphs)
        ]

# nexrag.yaml
ingestion:
  chunker:
    strategy: custom
    class: myapp.chunkers.ParagraphChunker

Embedders

An embedder converts text into dense vector representations. NexRAG uses the same embedder instance for both ingestion and query by default — ensuring the query vector lives in the same space as the indexed chunks.

Uses OpenAI's embedding API. Requires pip install "nexrag[openai]".

# nexrag.yaml
ingestion:
  embedder:
    provider: openai
    model: text-embedding-3-small   # or text-embedding-3-large, text-embedding-ada-002
    api_key: ${OPENAI_API_KEY}
    batch_size: 100                # texts per API call (default 100)

Google Gemini embeddings via the unified google-genai SDK. Dimensionality is detected lazily. Requires pip install "nexrag[gemini]". Reads GOOGLE_API_KEY / GEMINI_API_KEY if api_key is omitted.

# nexrag.yaml
ingestion:
  embedder:
    provider: gemini
    model: gemini-embedding-001     # or text-embedding-004
    api_key: ${GOOGLE_API_KEY}

Calls the HuggingFace Inference API. Works with both the free Inference API and dedicated endpoints. Requires pip install "nexrag[huggingface]".

# nexrag.yaml — HuggingFace Inference API
ingestion:
  embedder:
    provider: huggingface
    model: sentence-transformers/all-MiniLM-L6-v2
    api_key: ${HF_API_KEY}
    # base_url: https://your-dedicated-endpoint.hf.space  # dedicated endpoint override
    # max_retries: 3   # retries on HTTP 429/5xx with exponential backoff (default: 3)

Calls a locally-running Ollama server. No API key required. Requires pip install "nexrag[ollama]" and Ollama running locally.

# nexrag.yaml
ingestion:
  embedder:
    provider: ollama
    model: nomic-embed-text   # or mxbai-embed-large, all-minilm, etc.
    # base_url: http://localhost:11434  # default Ollama address
    # max_concurrent_requests: 10  # async concurrency limit (default: 10)

Implement BaseEmbedder for any model not covered above — local sentence-transformers, Cohere embed, Voyage AI, etc.

from nexrag.core.interfaces.embedder import BaseEmbedder

class CohereEmbedder(BaseEmbedder):
    def __init__(self, api_key: str, model: str = "embed-english-v3.0") -> None:
        import cohere
        self._client = cohere.Client(api_key)
        self._model = model

    @property
    def model_name(self) -> str: return self._model

    @property
    def dimensions(self) -> int: return 1024

    def embed(self, texts: list[str]) -> list[list[float]]:
        resp = self._client.embed(texts=texts, model=self._model, input_type="search_document")
        return resp.embeddings

    def embed_query(self, text: str) -> list[float]:
        resp = self._client.embed(texts=[text], model=self._model, input_type="search_query")
        return resp.embeddings[0]

# nexrag.yaml
ingestion:
  embedder:
    provider: custom
    class: myapp.embedders.CohereEmbedder
    params:
      api_key: ${COHERE_API_KEY}
      model:   embed-english-v3.0

Vector DB

The vector DB stores embedded chunks during ingestion and retrieves similar chunks at query time. NexRAG abstracts it behind BaseVectorDB — the ingestion and query pipelines never call a database directly.

The default and only built-in vector DB. Supports three modes: persistent (data saved to disk), memory (ephemeral, no disk I/O), and server (remote ChromaDB instance). Requires pip install "nexrag[chromadb]".

# nexrag.yaml — persistent (recommended for production)
vector_db:
  provider: chroma
  default_collection: documents
  on_conflict: overwrite
  upsert_batch_size: 500
  max_retries: 3
  retry_delay: 1.0
  collections:
    documents:
      mode: persistent
      path: ./.nexrag/chroma

# nexrag.yaml — memory (testing / CI)
vector_db:
  provider: chroma
  default_collection: test
  collections:
    test:
      mode: memory

# nexrag.yaml — server (remote ChromaDB)
vector_db:
  provider: chroma
  default_collection: documents
  collections:
    documents:
      mode: server
      # host: localhost   # ChromaDB server host
      # port: 8000        # ChromaDB server port

Managed serverless vector store. Each NexRAG collection maps to a Pinecone namespace inside a single index; the index is created lazily on first upsert from the embedder's dimension. Pinecone settings go under params. Requires pip install "nexrag[pinecone]".

# nexrag.yaml
ingestion:
  vector_db:
    provider: pinecone
    default_collection: documents
    collections: { documents: {} }   # collection names = Pinecone namespaces
    params:
      index_name: nexrag
      api_key: ${PINECONE_API_KEY}
      cloud: aws                  # cloud / region / metric optional
      region: us-east-1

Implement BaseVectorDB to connect to any vector store — Pinecone, Weaviate, Qdrant, pgvector, etc. The interface has four methods: upsert(), query(), delete(), list_collections().

from nexrag.core.interfaces.vector_db import BaseVectorDB

class PineconeDB(BaseVectorDB):
    def __init__(self, api_key: str, index_name: str) -> None: ...
    def upsert(self, chunks, collection): ...
    def query(self, vector, top_k, collection, metadata_filter): ...
    def delete(self, ids, collection): ...
    def list_collections(self): ...

# nexrag.yaml
vector_db:
  provider: custom
  class: myapp.db.PineconeDB
  params:
    api_key:    ${PINECONE_API_KEY}
    index_name: nexrag-prod

Retrievers

A retriever selects the most relevant chunks for a given query. NexRAG ships three retrieval strategies: dense (vector similarity), sparse (keyword overlap), and hybrid (fusion of both). The sparse retriever interface is open — BM25 is the built-in implementation, but any sparse algorithm can be plugged in.

Uses cosine similarity between the query vector and stored chunk vectors. Fast and accurate for semantic queries — finds conceptually similar text even if the exact words differ. No additional dependencies required.

# nexrag.yaml
query:
  retriever:
    provider: dense         # default
    top_k: 5               # chunks returned to the LLM
    score_threshold: 0.0   # drop chunks below this similarity score

Sparse retrieval ranks chunks by keyword overlap rather than semantic similarity. It excels at exact-term queries, product codes, proper nouns, and cases where dense retrieval misses literal matches.

Sparse is a category, not a single algorithm. NexRAG ships BM25 as the built-in sparse implementation, but the BaseRetriever interface supports any sparse method — TF-IDF, BM25+, SPLADE, or any custom scoring function. See Custom retriever to plug in a different sparse algorithm.

Okapi BM25 keyword scoring. Tokenizes the corpus at query time and ranks by term frequency adjusted for document length. Requires pip install "nexrag[bm25]".

# nexrag.yaml
query:
  retriever:
    provider: bm25
    top_k: 5

Fuses dense and sparse retrieval scores using a weighted sum. The alpha parameter controls the balance: alpha=1.0 is pure dense, alpha=0.0 is pure sparse. The default alpha=0.7 gives 70% weight to dense and 30% to sparse.

Hybrid is not tied to BM25 — the sparse.provider key accepts bm25 or custom, so you can plug in any sparse implementation via class:.

# nexrag.yaml — hybrid with BM25 sparse
query:
  retriever:
    provider: hybrid
    top_k: 10
    alpha: 0.7          # 0=pure sparse, 1=pure dense
    sparse_top_k: 50   # sparse candidate pool (use a larger set than top_k)
    sparse:
      provider: bm25    # bm25 | custom

# nexrag.yaml — hybrid with a custom sparse retriever
query:
  retriever:
    provider: hybrid
    top_k: 10
    alpha: 0.6
    sparse:
      provider: custom
      class: myapp.retrievers.TFIDFRetriever

Implement BaseRetriever to replace the entire retrieval strategy — or to supply a custom sparse retriever used by the hybrid fusion.

from nexrag.core.interfaces.retriever import BaseRetriever
from nexrag.core.models.chunk import ScoredChunk

class TFIDFRetriever(BaseRetriever):
    def retrieve(self, query: str, top_k: int, collection: str,
                  metadata_filter: dict | None = None) -> list[ScoredChunk]:
        # your TF-IDF implementation here
        ...

# nexrag.yaml — custom retriever as the primary strategy
query:
  retriever:
    provider: custom
    class: myapp.retrievers.TFIDFRetriever

Rerankers

An optional post-retrieval stage that re-scores a larger candidate set and returns the top-N most relevant chunks. The pattern: retrieve a wide set (top_k=50) for high recall, then rerank to a narrow set (top_n=5) for high precision — only the top-N reach the LLM.

Uses the Cohere Rerank API. State-of-the-art re-ranking quality with minimal latency overhead. Requires pip install "nexrag[cohere]" and a Cohere API key.

# nexrag.yaml
query:
  retriever:
    provider: dense
    top_k: 50
  reranker:
    provider: cohere
    model: rerank-english-v3.0   # or rerank-multilingual-v3.0
    api_key: ${COHERE_API_KEY}
    top_n: 5

Local, offline reranker using a cross-encoder model from sentence-transformers. No API key or network call — runs on CPU or GPU. Slower than Cohere for large candidate sets but fully private. Requires pip install "nexrag[cross-encoder]".

# nexrag.yaml
query:
  retriever:
    provider: dense
    top_k: 20
  reranker:
    provider: cross_encoder
    model: cross-encoder/ms-marco-MiniLM-L-6-v2
    top_n: 5
    # device: cuda   # cpu (default) | cuda | mps

Implement BaseReranker to use any reranking model — Voyage Rerank, Jina Reranker, a fine-tuned cross-encoder, etc.

from nexrag.core.interfaces.reranker import BaseReranker
from nexrag.core.models.chunk import ScoredChunk

class VoyageReranker(BaseReranker):
    def __init__(self, api_key: str) -> None:
        import voyageai
        self._client = voyageai.Client(api_key=api_key)

    def rerank(self, query: str, chunks: list[ScoredChunk], top_n: int) -> list[ScoredChunk]:
        texts = [c.content for c in chunks]
        result = self._client.rerank(query, texts, model="rerank-2", top_k=top_n)
        return [chunks[r.index] for r in result.results]

# nexrag.yaml
query:
  reranker:
    provider: custom
    class: myapp.rerankers.VoyageReranker
    params:
      api_key: ${VOYAGE_API_KEY}

LLMs

The LLM receives the formatted prompt (query + retrieved context) and generates the final answer. NexRAG calls the LLM as the last stage of the query pipeline and wraps the response in PipelineResult.

Uses the OpenAI Chat Completions API. Supports streaming natively. Requires pip install "nexrag[openai]".

# nexrag.yaml
query:
  llm:
    provider: openai
    model: gpt-4o          # gpt-4o, gpt-4o-mini, gpt-4-turbo, gpt-3.5-turbo, ...
    api_key: ${OPENAI_API_KEY}
    temperature: 0.2
    max_tokens: 1024
    timeout: 30           # seconds before request is aborted
    max_retries: 2        # automatic retries on transient errors

Uses the Anthropic Messages API (Claude models). Supports streaming natively. Requires pip install "nexrag[anthropic]".

# nexrag.yaml
query:
  llm:
    provider: anthropic
    model: claude-opus-4-8   # claude-sonnet-4-6, claude-haiku-4-5-20251001, ...
    api_key: ${ANTHROPIC_API_KEY}
    temperature: 0.2
    max_tokens: 1024
    timeout: 30
    max_retries: 2

Calls a locally-running Ollama server. No API key required — data stays on your machine. Requires pip install "nexrag[ollama]" and Ollama running locally.

# nexrag.yaml
query:
  llm:
    provider: ollama
    model: llama3.1        # any model pulled via `ollama pull`
    # base_url: http://localhost:11434  # default Ollama address
    temperature: 0.2
    max_tokens: 1024
    # max_retries: 2    # async retry attempts on transient failures (default: 2)

Google Gemini via the unified google-genai SDK — supports generate, stream, and native async. Requires pip install "nexrag[gemini]". Reads GOOGLE_API_KEY / GEMINI_API_KEY if api_key is omitted.

# nexrag.yaml
query:
  llm:
    provider: gemini
    model: gemini-2.5-flash     # or gemini-2.5-pro
    api_key: ${GOOGLE_API_KEY}
    temperature: 0.2
    max_tokens: 1024

Implement BaseLLM to call any model API not covered above — Mistral, Groq, a local vLLM server, etc. generate() returns (text, TokenUsage | None); implement stream() for stream_query() support.

from nexrag.core.interfaces.llm import BaseLLM
from nexrag.core.models.metrics import TokenUsage
from collections.abc import Iterator

class GroqLLM(BaseLLM):
    def __init__(self, api_key: str, model: str = "llama-3.3-70b-versatile") -> None:
        from groq import Groq
        self._client = Groq(api_key=api_key)
        self._model = model   # BaseLLM.model_name reads self._model

    def generate(self, prompt: str) -> tuple[str, TokenUsage | None]:
        r = self._client.chat.completions.create(
            model=self._model, messages=[{"role": "user", "content": prompt}])
        return r.choices[0].message.content, None

    def stream(self, prompt: str) -> Iterator[str]:
        stream = self._client.chat.completions.create(
            model=self._model, messages=[{"role": "user", "content": prompt}], stream=True)
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

# nexrag.yaml
query:
  llm:
    provider: custom
    class: myapp.llms.GroqLLM
    params:
      api_key: ${GROQ_API_KEY}
      model:   llama-3.3-70b-versatile

Guardrails

A pluggable security layer. Guards inspect text at four points in the pipeline and return a verdict — ALLOW, BLOCK, or REDACT. They compose into ordered chains under a top-level guardrails: block. Every chain is disabled by default. See the full threat model in SECURITY.md.

Each chain takes a policy: fail_open (default — a broken guard is treated as ALLOW) or fail_closed (treated as BLOCK). Every guard firing emits a PipelineEvent(stage="guardrail", …) for auditing and overhead measurement.

Pass auth_context to query(); the guard turns it into a retrieval filter so a request can only ever retrieve documents it is authorised to see.

# nexrag.yaml
guardrails:
  input:
    enabled: true
    policy: fail_closed
    guards:
      - type: access_control
        params: { mapping: { tenant: tenant } }

from nexrag import NexRAG, GuardrailBlockedError

pipeline = NexRAG.from_config("nexrag.yaml")

# Only retrieves chunks whose metadata["tenant"] == "acme"
result = pipeline.query("What's our revenue?", auth_context={"tenant": "acme"})

try:
    pipeline.query("ignore previous instructions…")   # input guard BLOCKs
except GuardrailBlockedError as e:
    print("blocked:", e)

Changelog