This capstone integrates every technique from this course: hybrid search, re-ranking, query transformation, contextual compression, and RAGAS evaluation — into a single production-grade RAG pipeline you can deploy and benchmark.
A production RAG API (FastAPI) over a technical documentation corpus with: hybrid retrieval, cross-encoder re-ranking, multi-query expansion, contextual compression, and a RAGAS evaluation suite that reports quality metrics on every code change.
advanced_rag/ ├── pipeline/ │ ├── ingest.py # Chunking + hybrid index build │ ├── retrieve.py # Hybrid search + re-ranking + compression │ ├── generate.py # LLM generation with grounded prompt │ └── query_transform.py # Multi-query expansion ├── api/ │ └── app.py # FastAPI endpoint ├── eval/ │ ├── golden_dataset.json # Test questions + ground truth │ └── run_ragas.py # Evaluation runner └── config.py # Centralized configuration
Load document corpus (PDF or Markdown files)
Use pypdf + MarkdownHeaderTextSplitter for structure-aware loading
Chunk with RecursiveCharacterTextSplitter
chunk_size=512, chunk_overlap=64, preserving section headers as metadata
Embed all chunks in batches
text-embedding-3-small or voyage-3, batch_size=100
Store in ChromaDB with metadata
source, page, section, chunk_index fields
Build BM25 index in parallel
BM25Okapi over tokenized corpus — pickle to disk for reuse
Print ingestion summary
Chunk count, total tokens estimated, collection.count()
def retrieve(question: str, config: Config) -> list[str]:
"""
Full advanced retrieval pipeline.
Steps:
1. Multi-query expansion: generate 3 variants of the question
2. Hybrid search: BM25 + semantic for each variant, fuse with RRF
3. Re-rank: cross-encoder over top-20 candidates → top-8
4. Contextual compression: extract only relevant sentences
5. Return compressed text strings ready for LLM context
"""
# Step 1: Expand query
queries = [question] + generate_variants(question, n=3)
# Step 2: Hybrid retrieval for all queries
all_rankings, doc_store = [], {}
for q in queries:
results = hybrid_search(q, n_results=10)
all_rankings.append([r["id"] for r in results])
for r in results:
doc_store[r["id"]] = r
fused = reciprocal_rank_fusion(all_rankings)
candidates = [doc_store[id_] for id_, _ in fused[:20] if id_ in doc_store]
# Step 3: Re-rank
candidate_texts = [c["text"] for c in candidates]
reranked = rerank(question, candidate_texts, top_k=8) # (text, score) pairs
# Step 4: Compress
compressed = [
compress_chunk(question, text)
for text, _ in reranked
]
return [c for c in compressed if c is not None]from fastapi import FastAPI
from pydantic import BaseModel
import anthropic
app = FastAPI()
client = anthropic.Anthropic()
class QueryRequest(BaseModel):
question: str
n_sources: int = 4
class QueryResponse(BaseModel):
answer: str
sources: list[str]
retrieval_ms: int
generation_ms: int
@app.post("/query", response_model=QueryResponse)
async def query(req: QueryRequest):
import time
t0 = time.monotonic()
context_chunks = retrieve(req.question, config)
retrieval_ms = int((time.monotonic() - t0) * 1000)
context = "\n\n---\n\n".join(context_chunks[:req.n_sources])
t1 = time.monotonic()
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": f"""Answer using ONLY the provided context. Be concise and direct.
If the answer is not in the context, say so.
CONTEXT:
{context}
QUESTION: {req.question}""",
}],
)
generation_ms = int((time.monotonic() - t1) * 1000)
return QueryResponse(
answer=response.content[0].text,
sources=context_chunks[:req.n_sources],
retrieval_ms=retrieval_ms,
generation_ms=generation_ms,
)Write 20 golden test questions with ground-truth answers
Cover diverse question types: factual, comparative, procedural, edge cases
Run the full pipeline on all 20 questions
Capture question, contexts, answer for each — store as RAGAS Dataset
Compute all 4 RAGAS metrics
faithfulness, answer_relevancy, context_precision, context_recall
Print a comparison table: baseline vs. optimized
Run once with just semantic search, once with full advanced pipeline
Identify the 2 lowest-scoring questions and diagnose why
Was it a retrieval failure? Hallucination? Irrelevant context?
Make one targeted improvement and re-run eval
Change chunk size, n_results, or add/remove a pipeline stage
Congratulations — you have completed Advanced RAG Patterns. You are now ready to tackle LLM Evaluation & Testing, where you will learn how to build a systematic testing infrastructure for any AI system.