RAG Pipeline

🎯 Goal

After reading this chapter:

• You know the full RAG (Retrieval Augmented Generation) architecture
• You can build a production-grade RAG pipeline
• You understand chunking strategies and trade-offs
• You can apply advanced RAG techniques (HyDE, multi-query, re-ranking)
• You know how to measure and improve RAG quality

What to learn

• RAG architecture — Naive, Advanced, Modular
• Chunking strategies — fixed, semantic, sliding window, recursive
• Retrieval strategies — dense, sparse, hybrid, multi-query
• Reranking — Cross-encoder, LLM-based
• HyDE(Hypothetical Document Embeddings)
• Citation and source attribution
• Context window management
• RAG evaluation — RAGAS, custom metrics

What is RAG and why?

Problem

LLM hallucination — it can give incorrect info:

• Training data is old (up to 2024)
• Doesn’t know your private documents
• Incorrect answer on specific facts

Solution — RAG

1. User asks: "What is our company policy?"
2. Retrieval: Fetch 5 similar chunks from vector DB
3. Augment: Add chunks to the prompt
4. Generate: LLM answers based on context
5. Cite: Show which chunks the answer came from

RAG vs Fine-tuning

**Rule:**RAGfor knowledge, fine-tuningfor behavior/style.

RAG architecture

Naive RAG

Query → Embed → Vector DB Search → Top-K chunks → LLM prompt → Answer

Problems:

• Poor retrieval → poor answer
• Contradictions among chunks in context
• LLM hallucinates outside the context

Advanced RAG (modern)

Query
  ↓
Query Transformation:
  - Multi-query (3 ta variant)
  - HyDE (sintetik javob → embed)
  - Step-back (umumiyroq savol)
  ↓
Hybrid Retrieval:
  - Dense (semantic)
  - Sparse (BM25)
  - Metadata filter
  ↓
Reranking (Cross-encoder)
  ↓
Context Construction:
  - Deduplication
  - Sort by relevance
  - Compress (LLM summary)
  ↓
LLM Generation:
  - Structured prompt
  - Citation markers
  ↓
Post-processing:
  - Source attribution
  - Confidence score

Code examples

Production RAG pipeline

from dataclasses import dataclass
from openai import AsyncOpenAI
from anthropic import AsyncAnthropic
from qdrant_client import AsyncQdrantClient
from sentence_transformers import CrossEncoder

@dataclass
class RetrievedChunk:
    text: str
    source: str
    page: int
    score: float

@dataclass
class RAGAnswer:
    answer: str
    sources: list[RetrievedChunk]
    confidence: float

class RAGPipeline:
    def __init__(self):
        self.openai = AsyncOpenAI()
        self.anthropic = AsyncAnthropic()
        self.qdrant = AsyncQdrantClient(url="http://localhost:6333")
        self.reranker = CrossEncoder("BAAI/bge-reranker-base")
        self.collection = "docs"
    
    async def embed(self, text: str) -> list[float]:
        response = await self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=[text],
        )
        return response.data[0].embedding
    
    async def retrieve(self, query: str, top_k: int = 20) -> list[RetrievedChunk]:
        embedding = await self.embed(query)
        results = await self.qdrant.search(
            collection_name=self.collection,
            query_vector=embedding,
            limit=top_k,
        )
        return [
            RetrievedChunk(
                text=r.payload["text"],
                source=r.payload.get("source", ""),
                page=r.payload.get("page", 0),
                score=r.score,
            )
            for r in results
        ]
    
    def rerank(self, query: str, chunks: list[RetrievedChunk], top_k: int = 5):
        pairs = [(query, c.text) for c in chunks]
        scores = self.reranker.predict(pairs)
        ranked = sorted(zip(scores, chunks), key=lambda x: -x[0])
        # Save the new score
        for new_score, chunk in ranked[:top_k]:
            chunk.score = float(new_score)
        return [c for _, c in ranked[:top_k]]
    
    def build_prompt(self, query: str, chunks: list[RetrievedChunk]) -> str:
        context = "\n\n".join([
            f"[Source{i+1}:{c.source}, page{c.page}]\n{c.text}"
            for i, c in enumerate(chunks)
        ])
        
        return f"""You are an expert assistant. Answer the question precisely based on the following context.

RULES:
1. Answer ONLY based on the given context
2. If the answer is not in the context, reply "No answer found in the provided information"
3. For each fact, provide a [Source N] citation
4. Answer in Uzbek

CONTEXT:
{context}QUESTION:{query}ANSWER:"""
    
    async def generate(self, prompt: str) -> tuple[str, float]:
        response = await self.anthropic.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        )
        text = response.content[0].text
        # Confidence estimation (simple heuristic)
        confidence = 0.9 if "[Source" in text else 0.3
        return text, confidence
    
    async def query(self, query: str) -> RAGAnswer:
        # 1. Retrieve
        chunks = await self.retrieve(query, top_k=20)
        
        # 2. Rerank
        top_chunks = self.rerank(query, chunks, top_k=5)
        
        # 3. Build prompt
        prompt = self.build_prompt(query, top_chunks)
        
        # 4. Generate
        answer, confidence = await self.generate(prompt)
        
        return RAGAnswer(
            answer=answer,
            sources=top_chunks,
            confidence=confidence,
        )

# Usage
rag = RAGPipeline()
result = await rag.query("What are our work hours?")
print(result.answer)
for src in result.sources:
    print(f" -{src.source}(p.{src.page}):{src.score:.3f}")

Multi-query — split the question into 3 variants

async def multi_query_search(query: str, top_k: int = 5):
    """One query → 3 variants → combined result."""
    
    # 1. Generate query variants
    variant_prompt = f"""Rewrite the following question in 3 different ways: Question:{query}Variants (each on a new line):
1.
2.
3."""
    
    response = await openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": variant_prompt}],
    )
    variants = response.choices[0].message.content.strip().split("\n")
    variants = [v.split(". ", 1)[1] for v in variants if ". " in v]
    
    # 2. Retrieve for each
    all_chunks = []
    for q in [query] + variants:
        chunks = await retrieve(q, top_k=top_k)
        all_chunks.extend(chunks)
    
    # 3. Deduplicate (by id or content hash)
    seen = set()
    unique = []
    for c in all_chunks:
        key = hash(c.text[:100])
        if key not in seen:
            seen.add(key)
            unique.append(c)
    
    return unique

HyDE — Hypothetical Document Embeddings

async def hyde_search(query: str, top_k: int = 5):
    """Instead of searching directly from the query, generate a synthetic 'answer' and embed it."""
    
    # 1. Generate synthetic answer
    hypothesis = await openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": 
            f"Write a complete, detailed answer to the following question (even if not factual):\n{query}"}],
    )
    hypothetical_answer = hypothesis.choices[0].message.content
    
    # 2. Embed the hypothetical answer
    embedding = await openai.embeddings.create(
        model="text-embedding-3-small",
        input=[hypothetical_answer],
    )
    
    # 3. Search with this embedding (answer → answer similarity!)
    results = await qdrant.search(
        collection_name="docs",
        query_vector=embedding.data[0].embedding,
        limit=top_k,
    )
    
    return results

Smart chunking strategies

from langchain.text_splitter import RecursiveCharacterTextSplitter

# Strategy 1: Fixed-size (simplest)
fixed = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
)

# Strategy 2: Markdown-aware
from langchain.text_splitter import MarkdownHeaderTextSplitter

md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=[
    ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"),
])

# Strategy 3: Semantic (LangChain experimental)
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

semantic = SemanticChunker(
    OpenAIEmbeddings(model="text-embedding-3-small"),
    breakpoint_threshold_type="percentile",
)

# Strategy 4: Sliding window (overlap)
def sliding_window_chunks(text: str, window: int = 500, stride: int = 250):
    chunks = []
    for i in range(0, len(text) - window + 1, stride):
        chunks.append(text[i:i + window])
    return chunks

Context window management

def build_context_within_budget(
    chunks: list[RetrievedChunk],
    max_tokens: int = 8000,
    encoder=tiktoken.encoding_for_model("gpt-4o"),
) -> list[RetrievedChunk]:
    """Return only chunks that fit the budget."""
    included = []
    total = 0
    
    for chunk in chunks:  # already sorted by relevance
        tokens = len(encoder.encode(chunk.text))
        if total + tokens > max_tokens:
            break
        included.append(chunk)
        total += tokens
    
    return included

RAG evaluation — RAGAS

# pip install ragas

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# Test set
data = {
    "question": ["What are the work hours?", "Where is the address?"],
    "answer": ["From 8:00 to 18:00", "Toshkent, Yunusobod"],
    "contexts": [
        ["Our work hours are Monday to Friday 8:00-18:00"],
        ["Office: Toshkent, Yunusobod district"],
    ],
    "ground_truth": ["8:00-18:00", "Toshkent, Yunusobod"],
}

dataset = Dataset.from_dict(data)
result = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)
print(result)
# {faithfulness: 0.95, answer_relevancy: 0.88, ...}

Backend integration

Production RAG FastAPI endpoint

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    app.state.rag = RAGPipeline()
    yield

app = FastAPI(lifespan=lifespan)

class RAGRequest(BaseModel):
    query: str
    session_id: str = None
    top_k: int = 5
    rerank: bool = True
    multi_query: bool = False

class RAGResponse(BaseModel):
    answer: str
    sources: list[dict]
    confidence: float
    latency_ms: int

@app.post("/rag/query", response_model=RAGResponse)
async def rag_query(req: RAGRequest):
    start = time.time()
    
    result = await app.state.rag.query(req.query)
    
    # Log for monitoring
    await log_query(
        query=req.query,
        answer=result.answer,
        sources=[s.source for s in result.sources],
        confidence=result.confidence,
        session_id=req.session_id,
    )
    
    return RAGResponse(
        answer=result.answer,
        sources=[
            {"text": s.text[:200], "source": s.source, "page": s.page, "score": s.score}
            for s in result.sources
        ],
        confidence=result.confidence,
        latency_ms=int((time.time() - start) * 1000),
    )

Streaming RAG answer (SSE)

@app.post("/rag/stream")
async def rag_stream(req: RAGRequest):
    # 1. Retrieve (non-streaming)
    chunks = await app.state.rag.retrieve(req.query)
    top_chunks = app.state.rag.rerank(req.query, chunks)
    prompt = app.state.rag.build_prompt(req.query, top_chunks)
    
    async def event_stream():
        # Send sources first
        sources = [{"source": c.source, "score": c.score} for c in top_chunks]
        yield f"data:{json.dumps({'type': 'sources', 'data': sources})}\n\n"
        
        # Stream LLM response
        async with anthropic.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        ) as stream:
            async for text in stream.text_stream:
                yield f"data:{json.dumps({'type': 'token', 'text': text})}\n\n"
        
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(event_stream(), media_type="text/event-stream")

Resources

• “Advanced RAG Techniques” — IVAN Ilin (Medium series)
• LlamaIndex Advanced RAG cookbook
• RAGAS docs — docs.ragas.io
• “RAG vs Fine-tuning” — Anthropic guide
• HyDE paper — Gao et al.
• Cohere RAG guides — production patterns

🏋️ Exercises

🟢 Easy

1. Naive RAG: 10 documents — chunking → vector DB → query.
2. Citation: show source as [Source N] in the answer.
3. Compare chunking strategies: 500 vs 1000 vs 2000 tokens.

🟡 Medium

1. Multi-query RAG: query → 3 variants → merge.
2. HyDE: synthetic answer → embed → search.
3. Reranking: top 20 → top 5 with cross-encoder.

🔴 Hard

1. Production RAG service: FastAPI + Qdrant + Celery (ingestion) + Langfuse (observability).
2. RAG evaluation: Create a test set of 100 Q&A pairs, evaluate with RAGAS.
3. Domain-specific tuning: Custom RAG for Uzbek legal documents (chunking, prompts).

Capstone

notebooks/month-05/06_rag_pipeline.ipynb:

• **Project:**RAG chatbot for the Constitution of Uzbekistan or Criminal Code
• 100+ documents ingestion
• Multi-query + HyDE + reranking
• Citation
• Streamlit UI
• RAGAS evaluation

✅ Checklist

• I know the RAG architecture
• I can apply chunking strategies (fixed, semantic)
• Hybrid retrieval (dense + sparse)
• Reranking (cross-encoder)
• HyDE and Multi-query
• Citation and source attribution
• Streaming RAG
• RAG evaluation (RAGAS)

Moving on to AI Agents.