Production-grade RAG pipeline with dense vector search, BM25 sparse retrieval, reciprocal rank fusion, cross-encoder reranking with an LLM-judge fallback, and grounded generation with verified inline citations.
real embeddings, not an illustration
explore the full map ↓Every component designed for reliability, observability, and scale. No vendor lock-in — runs on your infrastructure.
Dense vector embeddings (Qdrant) + sparse BM25 keyword search combined via Reciprocal Rank Fusion. Configurable weights for dense/sparse balance per query.
A local cross-encoder (ms-marco MiniLM) rescores the fused candidates in one batch — ~100 ms on CPU, no API cost. RERANK_MODE=llm swaps in an LLM judge for corpora far from its training distribution.
Extract claims → verify against sources → compute composite confidence (retrieval × citation coverage × completeness). Refuses below threshold.
Three strategies: fixed sliding window, recursive markdown-aware, semantic embedding-drift. Switchable via config — no code changes.
FastAPI service with Bearer auth. Two endpoints: /v1/ask and /v1/ingest. CLI for local development. OpenAPI docs at /docs.
Smoke tests with mocked LLM. Golden Q&A dataset structure defined. Chunking strategy comparison via config sweep. Phase 4 extensible.
Modular 7-stage pipeline with clear boundaries. Each stage is independently configurable, swappable, and observable — no hard dependencies between stages.
Document loading, parsing, and chunking with configurable strategies
def ingest(path: str, strategy: ChunkStrategy = "recursive"):
docs = load_documents(path)
chunks = chunk_documents(docs, strategy)
enriched = enrich_metadata(chunks)
return vector_store.upsert(enriched)Generate dense and sparse vector representations for hybrid search
async def embed(chunks: List[Chunk]) -> EmbeddedChunks:
dense = await dense_embedder.embed_batch(chunks)
sparse = sparse_encoder.encode(chunks)
return EmbeddedChunks(dense=dense, sparse=sparse)Hybrid vector + keyword search with reciprocal rank fusion
def retrieve(query: str, k: int = 20) -> List[Result]:
dense_results = vector_search(query, k * 2)
sparse_results = bm25_search(query, k * 2)
fused = rrf_fuse(dense_results, sparse_results)
return fused[:k]Cross-encoder reranking for precision improvement
async def rerank(query: str, results: List[Result]) -> List[Result]:
pairs = [(query, r.text) for r in results]
scores = await cross_encoder.score(pairs)
return sorted(zip(results, scores), key=lambda x: x[1], reverse=True)Grounded answer synthesis with citation tracking
async def generate(query: str, context: List[Result]) -> Answer:
prompt = build_rag_prompt(query, context)
stream = llm.stream(prompt)
answer = await collect_with_citations(stream)
confidence = score_confidence(answer, context)
return Answer(text=answer, confidence=confidence, citations=context)End-to-end tracing, metrics, and real-time monitoring
from opentelemetry import trace
tracer = trace.get_tracer("rag-pipeline")
@tracer.start_as_current_span("rag_query")
async def query_with_tracing(question: str):
with tracer.start_as_current_span("retrieval") as span:
results = retrieve(question)
span.set_attribute("results.count", len(results))
with tracer.start_as_current_span("generation") as span:
answer = await generate(question, results)
span.set_attribute("tokens.used", answer.token_count)
return answerAutomated quality assessment with golden datasets
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
def evaluate_pipeline(dataset: Dataset) -> EvaluationReport:
results = []
for sample in dataset:
answer = pipeline.ask(sample.question)
results.append({
"question": sample.question,
"answer": answer.text,
"contexts": [c.text for c in answer.citations],
"ground_truth": sample.expected_answer,
})
scores = evaluate(results, metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall,
])
return EvaluationReport(scores=scores, passed=all_passed(scores))Ask questions against the sample documentation. Requests are proxied to the live pipeline from this page — no API key needed. Answers are grounded, cited and confidence-scored; the console below each answer shows the real execution.
Enter a question above or click a sample to see the grounded answer with citations
projecting stored embeddings…
real embeddings
Every point is an indexed chunk, positioned from its actual stored embedding — same vectors the dense retrieval leg searches. Nearby points are semantically near; that is the search.
legend
reranked finalists
dense matches
sparse matches
indexed, unretrieved
Drag to orbit, scroll to zoom, hover a point for its passage. Ask a question below and the retrieved set lights up against this map.
Ingest, ask, and get a grounded answer with verified citations — in three commands.
View the quickstartFrom zero to running RAG pipeline in three commands.
Install the Python package dependencies and configure your environment.
pip install -r requirements.txt
cp .env.example .env
# Edit .env with your OPENAI_API_KEYLoad your documentation into the vector store using the CLI or REST API.
# CLI
python cli.py ingest ./sample_docs
# Or via API
curl -X POST localhost:8000/v1/ingest \
-H "Authorization: Bearer dev-secret-key" \
-H "Content-Type: application/json" \
-d '{"path": "./sample_docs"}'Query the RAG pipeline and get grounded answers with citations.
# CLI
python cli.py ask "How do I authenticate?"
# Or via API
curl -X POST localhost:8000/v1/ask \
-H "Authorization: Bearer dev-secret-key" \
-H "Content-Type: application/json" \
-d '{"question": "How do I authenticate?"}'| Variable | Default | Description |
|---|---|---|
| CHUNK_STRATEGY | recursive | fixed | recursive | semantic |
| DENSE_WEIGHT | 0.7 | Vector search weight |
| SPARSE_WEIGHT | 0.3 | BM25 search weight |
| RRF_K | 60 | Reciprocal rank fusion constant |
| RERANK_CANDIDATE_POOL | 20 | Candidates rescored by the reranker |
| FINAL_TOP_K | 5 | Final chunks for generation |
| MIN_RETRIEVAL_CONFIDENCE | 0.35 | Refusal threshold |
| DEDUP_SIMILARITY_THRESHOLD | 0.95 | Cosine similarity for dedup |