NexRAG

from nexrag import NexRAG

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

# ── Sync streaming — tokens printed as they arrive ───────────
for token in pipeline.stream_query("Summarize the contract"):
    print(token, end="", flush=True)

# ── Async streaming — for FastAPI / async frameworks ─────────
import asyncio

async def stream_answer():
    async for token in pipeline.astream_query(
        "What are the termination clauses?",
        top_k=8,
    ):
        print(token, end="", flush=True)

asyncio.run(stream_answer())

# ── FastAPI SSE endpoint — wire it straight in ────────────────
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

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

# nexrag.yaml — minimal config

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

ingestion:
  loader:
    type: pdf

  chunker:
    strategy: recursive
    chunk_size: 512
    chunk_overlap: 64

  embedder:
    provider: openai
    model: text-embedding-3-small
    api_key: ${OPENAI_API_KEY}

  vector_db:
    provider: chroma
    default_collection: documents
    collections:
      documents:
        mode: persistent
        path: ./.nexrag/chroma

query:
  embedder: inherit         # reuse ingestion embedder
  retriever:
    provider: dense
    top_k: 5
  llm:
    provider: openai
    model: gpt-4o
    api_key: ${OPENAI_API_KEY}
    temperature: 0.2

# Swap to Ollama — no API keys, fully local:
#   embedder.provider: ollama / model: nomic-embed-text
#   llm.provider: ollama  / model: llama3.2

# Drop in any custom component. Implement the interface,
# declare it in YAML. Works for all 9 pipeline stages.

# 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)

    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 — wire it in:
ingestion:
  embedder:
    provider: custom
    class:    myapp.embedders.SentenceTransformerEmbedder
    params:
      model:  all-MiniLM-L6-v2