Build a RAG Pipeline from Scratch: Production Python Tutorial

Q: How do you build a RAG pipeline from scratch?

Build a RAG pipeline in 6 steps. [DataCamp](https://www.datacamp.com/tutorial/building-a-rag-system-with-langchain-and-fastapi): (1) Set up FastAPI backend with async endpoints. (2) Load and chunk documents (512 tokens, 64 overlap). (3) Generate embeddings with text-embedding-3-small. (4) Store in Qdrant vector database with metadata. (5) Implement retrieval (hybrid: vector + BM25 + RRF fusion). (6) Generate with grounded prompt and citations. [Qdrant](https://qdrant.tech/documentation/tutorials-build-essentials/rag-deepseek/): 'Qdrant for vector storage, SentenceTransformers for embeddings, FastAPI for the web framework.' [Eduonix](https://blog.eduonix.com/2026/04/how-to-build-a-rag-system-step-by-step-guide/): 'For production systems, tools like Pinecone, Qdrant, or Weaviate offer scalability, filtering, and better performance. Accuracy depends on retrieval quality — with proper chunking, reranking, and good embeddings, RAG systems can significantly reduce hallucinations.' [AhmadWKhan](https://blog.ahmadwkhan.com/how-i-built-a-fully-local-rag-app-with-ollama-fastapi-and-qdrant): 'FastAPI for lightning-fast API, Qdrant for vector database, LangChain for RAG orchestration, Sentence Transformers for embeddings.'

Q: What tech stack should I use for a RAG pipeline?

The recommended 2026 production RAG stack is Python + FastAPI + Qdrant + text-embedding-3-small + cross-encoder reranker + RAGAS. [Inventiple](https://www.inventiple.com/blog/how-to-build-rag-pipeline-python): 'Stack: Python + Qdrant + text-embedding-3-small + BM25 + Cohere reranker + RAGAS. Qdrant is the recommended default for production RAG in 2026: it supports both dense and sparse (BM25) vectors natively, has strong filtering and metadata support, ships as a single binary or managed cloud service, and has an excellent Python client.' [DevTo](https://dev.to/martin_palopoli/how-i-built-a-production-rag-pipeline-with-fastapi-pgvector-and-cross-encoder-reranking-j2o): 'Backend: Python 3.12 + FastAPI (100% async). Database: PostgreSQL 16 + pgvector + tsvector. Embeddings: paraphrase-multilingual-MiniLM-L12-v2 (384d). Reranker: cross-encoder/ms-marco-MiniLM-L-6-v2. LLM: Groq (llama-3.3-70b-versatile). Cache: Redis + PostgreSQL (semantic cache). Streaming: Server-Sent Events (SSE).' [HamLuk](https://dev.to/hamluk/building-production-ready-rag-in-fastapi-with-vector-databases-39gf): 'The goal is not to explain how to generate embeddings, but how to integrate a vector-based RAG component cleanly into a FastAPI backend.'

Q: How do you implement hybrid search in RAG?

Implement hybrid search by combining dense vector search with sparse BM25 keyword search using Reciprocal Rank Fusion. [DevTo](https://dev.to/martin_palopoli/how-i-built-a-production-rag-pipeline-with-fastapi-pgvector-and-cross-encoder-reranking-j2o): 'Stage 1: Hybrid Search — pgvector (semantic) + BM25 (lexical) with Reciprocal Rank Fusion. Vector search uses pgvector HNSW with cosine distance. Lexical search uses PostgreSQL tsvector + ts_rank_cd. Fusion with RRF merges results: score[doc] = sum of weight / (k + rank + 1) for each result list.' [Inventiple](https://www.inventiple.com/blog/how-to-build-rag-pipeline-python): 'Pure dense vector search is good but misses exact keyword matches — acronyms, product names, version numbers. Sparse BM25 retrieval catches these. Combining both with Reciprocal Rank Fusion (RRF) gives you the best of both worlds.' [Qdrant](https://qdrant.tech/documentation/tutorials-build-essentials/rag-deepseek/): Qdrant supports both dense and sparse vectors natively, making hybrid search straightforward. Steps: (1) Generate dense embedding for query. (2) Run BM25 keyword search. (3) Fuse with RRF: score = (1-bm25_weight)/(k+rank_dense+1) + bm25_weight/(k+rank_sparse+1). (4) Return fused top-K candidates.

Q: How do you add reranking to a RAG pipeline?

Add reranking by using a cross-encoder model to re-score retrieved candidates. [DevTo](https://dev.to/martin_palopoli/how-i-built-a-production-rag-pipeline-with-fastapi-pgvector-and-cross-encoder-reranking-j2o): 'Stage 2: Cross-Encoder Reranking — a cross-encoder processes query + chunk together as a single input, capturing cross-attention relationships. It is slower but significantly more accurate. Pairs = [(query, chunk_content) for chunk in candidates]. Scores = reranker.predict(pairs). Sort by rerank_score descending.' [Inventiple](https://www.inventiple.com/blog/how-to-build-rag-pipeline-python): 'Retrieval returns candidates; reranking picks the best ones. A cross-encoder reranker like Cohere rerank-english-v3.0 reads both the query and each passage together and scores relevance more accurately than embedding-based similarity. Our benchmark finding: 256-token chunks with Cohere reranking consistently outperformed 1024-token chunks without reranking.' [Glukhov](https://www.glukhov.org/rag/): 'Reranking with Embedding Models, Qwen3 Embedding + Qwen3 Reranker on Ollama. Reranking with Ollama + Qwen3 Reranker in Go.' Steps: (1) Over-fetch candidates (3x top_k). (2) Create (query, document) pairs. (3) Score with cross-encoder. (4) Sort by score. (5) Return top_k.

Q: How do you evaluate a RAG pipeline in production?

Evaluate a RAG pipeline with RAGAS metrics and continuous monitoring. [Inventiple](https://www.inventiple.com/blog/how-to-build-rag-pipeline-python): 'Evaluate before shipping — RAGAS faithfulness below 0.8 means hallucination risk is unacceptable. 512-token chunks with Cohere reranker outperforms naive 1024-token retrieval consistently. Hybrid retrieval is the production default — pure dense retrieval misses keyword-specific queries.' [Glukhov](https://www.glukhov.org/rag/): 'Track retrieval metrics — recall@k and precision@k continuously. Log retrieval set, reranked set, final context, latency, cost. Evaluate online/offline. Mastering RAG architecture, reranking, vector databases, hybrid search, and evaluation will determine whether your AI system remains a demo or becomes production-ready.' [NerdLevelTech](https://nerdleveltech.com/guides/rag-systems): 'Add evaluation: create a test set of 20 questions and measure RAGAS scores. Optimize chunking, add hybrid search, add reranking, go to production. Add caching, monitoring, and access controls. Scale: move to a managed vector database and optimize costs.' RAGAS metrics: faithfulness (0.8+), answer relevancy, context precision, context recall. Track in CI/CD pipeline.

By Fadi Updated Oct 2, 2025 13 min read RAG Fundamentals

TL;DR — Build a production RAG pipeline from scratch with Python, FastAPI, Qdrant, and cross-encoder reranking. DataCamp: "FastAPI processes other requests while RAG works in the background. Documents are loaded, chunked, embeddings generated, and stored." Inventiple: "Stack: Python + Qdrant + text-embedding-3-small + BM25 + Cohere reranker + RAGAS. 512-token chunks + reranker beats 1024-token without. RAGAS faithfulness below 0.8 means hallucination risk." DevTo: "A production RAG pipeline is not embed + search + prompt. It is a multi-stage system: hybrid search covers semantic and lexical, cross-encoder ranks by relevance, MMR ensures diversity, cache eliminates redundant work." Qdrant: "Qdrant for vector storage, SentenceTransformers for embeddings, FastAPI for the web framework." Eduonix: "With proper chunking, reranking, and good embeddings, RAG systems can significantly reduce hallucinations and achieve high reliability." Learn more with what is RAG, how RAG works, chunking strategies, and reranking.

DevTo sets expectations: "A production RAG pipeline is not 'embed + search + prompt'. It's a multi-stage system where each component solves a specific weakness of the previous one. Hybrid search covers semantic and lexical gaps, the cross-encoder ranks by actual relevance, MMR ensures diversity, cache eliminates redundant work, and FAQs shortcut when the answer already exists."

Inventiple frames the guide: "Most RAG tutorials show you a working demo in 20 lines. This is not that guide. We cover what actually matters for production: chunking strategy, hybrid retrieval, reranking, and evaluation. These are the decisions that separate a demo that impresses in a slide deck from a system that handles real user queries reliably."

RAG Pipeline Architecture

flowchart TD subgraph API["FastAPI Backend"] Upload["POST /documents
Upload and index"] Query["POST /query
RAG query"] Stream["GET /stream
SSE streaming"] Health["GET /health
Health check"] end subgraph Indexing["Indexing Pipeline"] Load["Load Documents
(PDF, markdown, text)"] Chunk["Chunk
(512 tokens, 64 overlap)"] Embed["Embed
(text-embedding-3-small)"] Store["Store in Qdrant
(vectors + metadata)"] end subgraph Retrieval["Retrieval Pipeline"] EmbedQuery["Embed Query"] Hybrid["Hybrid Search
(dense + BM25 + RRF)"] Rerank["Cross-Encoder Rerank
(top-K re-scoring)"] MMR["MMR Diversity
(optional)"] Context["Assemble Context
(dedupe + citations)"] end subgraph Generation["Generation"] Prompt["Grounded Prompt
(context + refusal)"] LLM["LLM Generate
(streaming)"] Response["Response + Sources"] end subgraph Infra["Infrastructure"] Qdrant["Qdrant
(vector database)"] Redis["Redis
(semantic cache)"] RAGAS["RAGAS
(evaluation)"] end Upload --> Indexing Load --> Chunk --> Embed --> Store Store --> Qdrant Query --> Retrieval EmbedQuery --> Hybrid Hybrid --> Qdrant Hybrid --> Rerank --> MMR --> Context Context --> Prompt --> LLM --> Response Stream --> Generation Response --> Redis Response --> RAGAS

Production RAG Stack

Component	Technology	Why
API framework	FastAPI (async)	Lightning-fast, type hints, auto docs
Vector database	Qdrant	Native dense + sparse, self-hosted or managed
Embeddings	text-embedding-3-small (1536d)	Best price/performance, multilingual
Reranker	Cohere rerank-english-v3.0	Cross-encoder, 256-token + reranker beats 1024-token
LLM	GPT-4o / Claude / Llama (self-hosted)	Flexible, multi-model
Cache	Redis (semantic cache)	Eliminates redundant queries
Streaming	Server-Sent Events (SSE)	Better UX, progressive response
Evaluation	RAGAS	Faithfulness, relevancy, precision, recall
Orchestration	LangChain (optional)	RAG logic chaining

Implementation

import asyncio
import hashlib
import json
import time
from datetime import datetime, timezone
from typing import Optional
from dataclasses import dataclass, field
from enum import Enum

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI(title="RAG Pipeline API", version="1.0.0")

# ─── Data Models ───

class DocumentUpload(BaseModel):
    content: str
    title: str = ""
    source: str = "upload"
    metadata: dict = {}

class QueryRequest(BaseModel):
    query: str
    top_k: int = 5
    use_hybrid: bool = True
    use_reranking: bool = True
    stream: bool = False

class QueryResponse(BaseModel):
    answer: str
    sources: list
    retrieved_chunks: int
    timing: dict

# ─── RAG Pipeline ───

@dataclass
class RAGPipeline:
    """Production RAG pipeline with FastAPI, Qdrant, and reranking."""

    chunk_size: int = 512
    chunk_overlap: int = 64
    bm25_weight: float = 0.3
    rrf_k: int = 60
    over_fetch_multiplier: int = 3

    def __init__(self, embedder, vdb, llm, reranker=None, cache=None):
        self.embedder = embedder
        self.vdb = vdb
        self.llm = llm
        self.reranker = reranker
        self.cache = cache

    async def index_document(self, doc: DocumentUpload) -> dict:
        """Index a document: chunk, embed, store."""
        start = time.time()

        # Step 1: Normalize and chunk
        text = " ".join(doc.content.split())
        chunks = self._chunk_text(text, self.chunk_size, self.chunk_overlap)

        # Step 2: Embed chunks
        embeddings = await self.embedder.embed_batch(chunks)

        # Step 3: Store in Qdrant with metadata
        points = []
        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
            chunk_id = hashlib.sha256(
                f"{doc.title}:{i}".encode()).hexdigest()[:16]
            points.append({
                "id": chunk_id,
                "vector": embedding,
                "payload": {
                    "content": chunk,
                    "title": doc.title,
                    "source": doc.source,
                    "chunk_index": i,
                    "token_count": len(chunk.split()),
                    "metadata": doc.metadata,
                    "indexed_at": datetime.now(timezone.utc).isoformat(),
                },
            })

        await self.vdb.upsert(
            collection_name="knowledge_base",
            points=points,
        )

        return {
            "title": doc.title,
            "chunks_indexed": len(points),
            "indexing_time_ms": round((time.time() - start) * 1000, 1),
        }

    async def query(self, request: QueryRequest) -> QueryResponse:
        """Run full RAG query: retrieve, rerank, generate."""
        start = time.time()

        # Check semantic cache
        if self.cache:
            cached = await self.cache.get(request.query)
            if cached:
                cached["timing"]["cache_hit"] = True
                return QueryResponse(**cached)

        # Stage 1: Embed query
        embed_start = time.time()
        query_embedding = await self.embedder.embed(request.query)
        embed_time = (time.time() - embed_start) * 1000

        # Stage 2: Retrieve (hybrid or dense-only)
        retrieve_start = time.time()
        fetch_k = request.top_k * self.over_fetch_multiplier

        if request.use_hybrid:
            dense_results = await self.vdb.search(
                collection_name="knowledge_base",
                vector=query_embedding,
                limit=fetch_k,
            )
            sparse_results = await self.vdb.search(
                collection_name="knowledge_base",
                query=request.query,
                limit=fetch_k,
            )
            candidates = self._rrf_fusion(
                dense_results, sparse_results,
                self.bm25_weight, self.rrf_k)
        else:
            candidates = await self.vdb.search(
                collection_name="knowledge_base",
                vector=query_embedding,
                limit=fetch_k,
            )
        retrieve_time = (time.time() - retrieve_start) * 1000

        # Stage 3: Rerank
        rerank_start = time.time()
        if request.use_reranking and self.reranker:
            pairs = [
                (request.query, c["payload"]["content"])
                for c in candidates
            ]
            scores = await self.reranker.predict(pairs)
            for c, score in zip(candidates, scores):
                c["rerank_score"] = float(score)
            candidates = sorted(
                candidates,
                key=lambda x: x.get("rerank_score", 0),
                reverse=True,
            )
        rerank_time = (time.time() - rerank_start) * 1000

        # Select top_k
        top_results = candidates[:request.top_k]

        # Stage 4: Assemble context with citations
        context_parts = []
        sources = []
        for i, result in enumerate(top_results):
            content = result["payload"]["content"]
            source = result["payload"].get("source", "unknown")
            context_parts.append(f"[{i+1}] ({source})\n{content}")
            sources.append({
                "citation": i + 1,
                "source": source,
                "title": result["payload"].get("title", ""),
            })
        context = "\n\n".join(context_parts)

        # Stage 5: Generate with grounded prompt
        generate_start = time.time()
        prompt = self._build_prompt(request.query, context)
        answer = await self.llm.generate(prompt)
        generate_time = (time.time() - generate_start) * 1000

        total_time = (time.time() - start) * 1000

        response = {
            "answer": answer,
            "sources": sources,
            "retrieved_chunks": len(top_results),
            "timing": {
                "embed_ms": round(embed_time, 1),
                "retrieve_ms": round(retrieve_time, 1),
                "rerank_ms": round(rerank_time, 1),
                "generate_ms": round(generate_time, 1),
                "total_ms": round(total_time, 1),
                "cache_hit": False,
            },
        }

        # Cache the response
        if self.cache:
            await self.cache.set(request.query, response, ttl=3600)

        return QueryResponse(**response)

    async def stream_query(self, request: QueryRequest):
        """Stream RAG response via SSE."""
        # Retrieve and rerank (non-streaming)
        query_embedding = await self.embedder.embed(request.query)
        fetch_k = request.top_k * self.over_fetch_multiplier

        if request.use_hybrid:
            dense = await self.vdb.search(
                collection_name="knowledge_base",
                vector=query_embedding, limit=fetch_k)
            sparse = await self.vdb.search(
                collection_name="knowledge_base",
                query=request.query, limit=fetch_k)
            candidates = self._rrf_fusion(dense, sparse)
        else:
            candidates = await self.vdb.search(
                collection_name="knowledge_base",
                vector=query_embedding, limit=fetch_k)

        if request.use_reranking and self.reranker:
            pairs = [(request.query, c["payload"]["content"])
                     for c in candidates]
            scores = await self.reranker.predict(pairs)
            for c, s in zip(candidates, scores):
                c["rerank_score"] = float(s)
            candidates = sorted(
                candidates, key=lambda x: x["rerank_score"],
                reverse=True)

        top = candidates[:request.top_k]
        context = "\n\n".join(
            f"[{i+1}] {r['payload']['content']}"
            for i, r in enumerate(top))
        prompt = self._build_prompt(request.query, context)

        # Stream LLM tokens
        async for token in self.llm.stream(prompt):
            yield f"data: {json.dumps({'token': token})}\n\n"

        sources = [
            {"citation": i+1, "source": r["payload"].get("source", "")}
            for i, r in enumerate(top)
        ]
        yield f"data: {json.dumps({'sources': sources, 'done': True})}\n\n"

    def _chunk_text(self, text: str, size: int, overlap: int) -> list:
        words = text.split()
        chunks = []
        start = 0
        while start < len(words):
            end = min(start + size, len(words))
            chunks.append(" ".join(words[start:end]))
            if end == len(words):
                break
            start = end - overlap
        return chunks

    def _rrf_fusion(self, dense: list, sparse: list,
                    bm25_weight: float = 0.3, k: int = 60) -> list:
        scores = {}
        results_map = {}
        for rank, r in enumerate(dense):
            scores[r["id"]] = scores.get(r["id"], 0) + \
                (1 - bm25_weight) / (k + rank + 1)
            results_map[r["id"]] = r
        for rank, r in enumerate(sparse):
            scores[r["id"]] = scores.get(r["id"], 0) + \
                bm25_weight / (k + rank + 1)
            results_map[r["id"]] = r
        return [results_map[i]
                for i in sorted(scores, key=scores.get, reverse=True)]

    def _build_prompt(self, query: str, context: str) -> str:
        return (
            "Answer the question based ONLY on the provided context. "
            "If the context does not contain enough information, say "
            "'I don't have enough information to answer this question.'\n\n"
            f"Context:\n{context}\n\n"
            f"Question: {query}\n\n"
            "Answer (cite sources using [1], [2], etc.):"
        )

# ─── FastAPI Endpoints ───

pipeline: Optional[RAGPipeline] = None

@app.post("/documents")
async def upload_document(doc: DocumentUpload):
    if not pipeline:
        raise HTTPException(503, "RAG pipeline not initialized")
    return await pipeline.index_document(doc)

@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    if not pipeline:
        raise HTTPException(503, "RAG pipeline not initialized")
    return await pipeline.query(request)

@app.get("/stream")
async def stream(query: str, top_k: int = 5):
    if not pipeline:
        raise HTTPException(503, "RAG pipeline not initialized")
    request = QueryRequest(query=query, top_k=top_k, stream=True)
    return StreamingResponse(
        pipeline.stream_query(request),
        media_type="text/event-stream",
    )

@app.get("/health")
async def health():
    return {"status": "healthy", "pipeline": pipeline is not None}

Build RAG Pipeline Checklist

[ ] Set up Python 3.12+ with FastAPI, Qdrant client, and embedding library
[ ] Install dependencies: fastapi, uvicorn, qdrant-client, sentence-transformers, cohere
[ ] Configure Qdrant (self-hosted Docker or managed cloud)
[ ] Create collection with HNSW index and metadata payload schema
[ ] Choose embedding model (text-embedding-3-small for production, MiniLM for local)
[ ] Set chunk size to 512 tokens with 64 token overlap
[ ] Implement document upload endpoint (POST /documents)
[ ] Normalize text: clean whitespace, remove boilerplate, deduplicate
[ ] Chunk documents with configurable size and overlap
[ ] Generate embeddings in batch for efficiency
[ ] Store vectors with metadata (title, source, chunk_index, timestamp)
[ ] Implement query endpoint (POST /query)
[ ] Embed query using the same model as indexing
[ ] Implement hybrid search: dense vector + BM25 + RRF fusion
[ ] Set BM25 weight (0.3 default) and RRF k (60 default)
[ ] Over-fetch candidates (3x top_k) for reranking
[ ] Add cross-encoder reranking (Cohere rerank-english-v3.0 or ms-marco-MiniLM)
[ ] Create (query, document) pairs for cross-encoder scoring
[ ] Sort by rerank score and select top_k
[ ] Assemble context with citation numbers [1], [2], etc.
[ ] Build grounded prompt with refusal behavior
[ ] Generate response with LLM
[ ] Return answer with sources and timing metrics
[ ] Add SSE streaming endpoint (GET /stream) for progressive response
[ ] Add semantic cache with Redis for repeated queries
[ ] Add MMR (Maximum Marginal Relevance) for result diversity
[ ] Add health check endpoint (GET /health)
[ ] Add API key authentication
[ ] Add rate limiting
[ ] Add request/response logging
[ ] Set up RAGAS evaluation pipeline
[ ] Create test set of 20+ questions with ground truth answers
[ ] Compute RAGAS faithfulness (target: 0.8+)
[ ] Compute answer relevancy, context precision, context recall
[ ] Track recall@k and precision@k continuously
[ ] Evaluate before and after every significant change
[ ] Add monitoring: latency, cost, cache hit rate, error rate
[ ] Add access controls — filter chunks by user permissions
[ ] Consider LangChain for orchestration (optional)
[ ] Consider Ollama for fully local RAG (self-hosted Mistral)
[ ] Consider pgvector if already on PostgreSQL (<10M vectors)
[ ] Deploy with Docker Compose (FastAPI + Qdrant + Redis)
[ ] Read what is RAG for architecture
[ ] Read how RAG works for pipeline stages
[ ] Optimize chunking strategies
[ ] Understand embeddings
[ ] Choose embedding model
[ ] Add hybrid search for production
[ ] Add reranking for precision
[ ] Prevent hallucinations with grounding
[ ] Implement citations for trust
[ ] Evaluate with RAG metrics
[ ] Apply zero-trust to RAG access
[ ] Log all RAG operations in audit logs
[ ] Test: indexing produces correct chunk count and embeddings
[ ] Test: hybrid search returns both semantic and keyword matches
[ ] Test: reranking improves top-k precision
[ ] Test: grounded prompt refuses when context is insufficient
[ ] Test: streaming endpoint produces progressive tokens
[ ] Test: RAGAS faithfulness meets 0.8+ threshold
[ ] Document API endpoints, configuration, and deployment instructions

FAQ

How do you build a RAG pipeline from scratch?

Build a RAG pipeline in 6 steps. DataCamp: (1) Set up FastAPI backend with async endpoints. (2) Load and chunk documents (512 tokens, 64 overlap). (3) Generate embeddings with text-embedding-3-small. (4) Store in Qdrant vector database with metadata. (5) Implement retrieval (hybrid: vector + BM25 + RRF fusion). (6) Generate with grounded prompt and citations. Qdrant: "Qdrant for vector storage, SentenceTransformers for embeddings, FastAPI for the web framework." Eduonix: "For production systems, tools like Pinecone, Qdrant, or Weaviate offer scalability, filtering, and better performance. Accuracy depends on retrieval quality — with proper chunking, reranking, and good embeddings, RAG systems can significantly reduce hallucinations." AhmadWKhan: "FastAPI for lightning-fast API, Qdrant for vector database, LangChain for RAG orchestration, Sentence Transformers for embeddings."

What tech stack should I use for a RAG pipeline?

The recommended 2026 production RAG stack is Python + FastAPI + Qdrant + text-embedding-3-small + cross-encoder reranker + RAGAS. Inventiple: "Stack: Python + Qdrant + text-embedding-3-small + BM25 + Cohere reranker + RAGAS. Qdrant is the recommended default for production RAG in 2026: it supports both dense and sparse (BM25) vectors natively, has strong filtering and metadata support, ships as a single binary or managed cloud service, and has an excellent Python client." DevTo: "Backend: Python 3.12 + FastAPI (100% async). Database: PostgreSQL 16 + pgvector + tsvector. Embeddings: paraphrase-multilingual-MiniLM-L12-v2 (384d). Reranker: cross-encoder/ms-marco-MiniLM-L-6-v2. LLM: Groq (llama-3.3-70b-versatile). Cache: Redis + PostgreSQL (semantic cache). Streaming: Server-Sent Events (SSE)." HamLuk: "The goal is not to explain how to generate embeddings, but how to integrate a vector-based RAG component cleanly into a FastAPI backend."

How do you implement hybrid search in RAG?

Implement hybrid search by combining dense vector search with sparse BM25 keyword search using Reciprocal Rank Fusion. DevTo: "Stage 1: Hybrid Search — pgvector (semantic) + BM25 (lexical) with Reciprocal Rank Fusion. Vector search uses pgvector HNSW with cosine distance. Lexical search uses PostgreSQL tsvector + ts_rank_cd. Fusion with RRF merges results: score[doc] = sum of weight / (k + rank + 1) for each result list." Inventiple: "Pure dense vector search is good but misses exact keyword matches — acronyms, product names, version numbers. Sparse BM25 retrieval catches these. Combining both with Reciprocal Rank Fusion (RRF) gives you the best of both worlds." Qdrant: Qdrant supports both dense and sparse vectors natively, making hybrid search straightforward. Steps: (1) Generate dense embedding for query. (2) Run BM25 keyword search. (3) Fuse with RRF: score = (1-bm25_weight)/(k+rank_dense+1) + bm25_weight/(k+rank_sparse+1). (4) Return fused top-K candidates.

How do you add reranking to a RAG pipeline?

Add reranking by using a cross-encoder model to re-score retrieved candidates. DevTo: "Stage 2: Cross-Encoder Reranking — a cross-encoder processes query + chunk together as a single input, capturing cross-attention relationships. It is slower but significantly more accurate. Pairs = [(query, chunk_content) for chunk in candidates]. Scores = reranker.predict(pairs). Sort by rerank_score descending." Inventiple: "Retrieval returns candidates; reranking picks the best ones. A cross-encoder reranker like Cohere rerank-english-v3.0 reads both the query and each passage together and scores relevance more accurately than embedding-based similarity. Our benchmark finding: 256-token chunks with Cohere reranking consistently outperformed 1024-token chunks without reranking." Glukhov: "Reranking with Embedding Models, Qwen3 Embedding + Qwen3 Reranker on Ollama. Reranking with Ollama + Qwen3 Reranker in Go." Steps: (1) Over-fetch candidates (3x top_k). (2) Create (query, document) pairs. (3) Score with cross-encoder. (4) Sort by score. (5) Return top_k.

How do you evaluate a RAG pipeline in production?

Evaluate a RAG pipeline with RAGAS metrics and continuous monitoring. Inventiple: "Evaluate before shipping — RAGAS faithfulness below 0.8 means hallucination risk is unacceptable. 512-token chunks with Cohere reranker outperforms naive 1024-token retrieval consistently. Hybrid retrieval is the production default — pure dense retrieval misses keyword-specific queries." Glukhov: "Track retrieval metrics — recall@k and precision@k continuously. Log retrieval set, reranked set, final context, latency, cost. Evaluate online/offline. Mastering RAG architecture, reranking, vector databases, hybrid search, and evaluation will determine whether your AI system remains a demo or becomes production-ready." NerdLevelTech: "Add evaluation: create a test set of 20 questions and measure RAGAS scores. Optimize chunking, add hybrid search, add reranking, go to production. Add caching, monitoring, and access controls. Scale: move to a managed vector database and optimize costs." RAGAS metrics: faithfulness (0.8+), answer relevancy, context precision, context recall. Track in CI/CD pipeline.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →