Module 1: The Complete RAG Pipeline (Architecture Overview)
Project: Baseline RAG System
Project overview
This is the Module 1 mini-project: building a complete baseline RAG system with ChromaDB + OpenAI. This baseline is the comparison point for every optimization you'll implement in modules 2-8.
The goal is NOT to build the best possible RAG (you'll get there gradually in the following modules), but to establish a baseline that works, that's measured, and that's documented. You need concrete numbers (precision: 68%, latency: 800ms) to validate that the advanced techniques actually improve the system.
This project teaches you the complete RAG flow: (1) Indexing (chunking + embeddings + storage), (2) Retrieval (query + search), (3) Generation (context + LLM), (4) Evaluation (manual metrics). By the end you'll have a working system and documented baseline metrics.
🎯 Project objectives
The main objective:
Build a baseline RAG system with measurable performance to compare against the improvements in modules 2-8.
The specific objectives:
- ✅ Implement the complete pipeline (Indexing → Retrieval → Generation)
- ✅ Use simple components (fixed-size chunking, direct query, cosine similarity)
- ✅ Measure baseline performance (latency, precision, recall)
- ✅ Document your architecture decisions
- ✅ Set up a production-ready environment (.env, .gitignore, requirements.txt)
📋 Technical requirements
The tech stack:
- Python: 3.10+
- Vector DB: ChromaDB 0.4.22 (local, free)
- Embeddings: OpenAI text-embedding-ada-002
- LLM: OpenAI GPT-3.5-turbo
- Chunking: Fixed-size (a simple baseline)
- Environment: python-dotenv for the API keys
The project's dataset:
You'll use the FastAPI documentation (50-100 pages) as your test corpus.
🛠️ Setting up the project
Step 1: The directory structure
mkdir rag_baseline_project
cd rag_baseline_project
# The structure
rag_baseline_project/
├── .env # API keys (DO NOT commit)
├── .gitignore # Ignore .env, venv, etc.
├── requirements.txt # Pinned dependencies
├── data/
│ └── fastapi_docs.txt # The document corpus
├── src/
│ ├── indexing.py # The indexing pipeline
│ ├── retrieval.py # The retrieval pipeline
│ ├── generation.py # The generation pipeline
│ └── evaluation.py # Metrics and benchmarking
├── notebooks/
│ └── rag_baseline_demo.ipynb # An interactive demo
└── README.md # The project documentationStep 2: Install the dependencies
# Create a virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install the dependencies
pip install chromadb==0.4.22 openai==1.12.0 python-dotenv==1.0.0
# Save the requirements
pip freeze > requirements.txtrequirements.txt:
chromadb==0.4.22
openai==1.12.0
python-dotenv==1.0.0Step 3: Configure the API keys
.env:
OPENAI_API_KEY=sk-proj-...your-key-here....gitignore:
# Environment
.env
venv/
__pycache__/
*.pyc
# Data (if it contains sensitive info)
# data/
# ChromaDB persistence
chroma_db/
# Jupyter notebook checkpoints
.ipynb_checkpoints/Step 4: Verify the setup
# test_setup.py
import chromadb
from openai import OpenAI
from dotenv import load_dotenv
import os
load_dotenv()
# Test the OpenAI API key
openai_key = os.getenv("OPENAI_API_KEY")
if not openai_key:
print("❌ OPENAI_API_KEY not found in .env")
else:
print("✅ OPENAI_API_KEY configured")
# Test ChromaDB
try:
client = chromadb.Client()
print("✅ ChromaDB works correctly")
except Exception as e:
print(f"❌ ChromaDB error: {e}")
# Test the OpenAI embeddings
try:
client = OpenAI()
response = client.embeddings.create(
model="text-embedding-ada-002",
input="Test"
)
print(f"✅ OpenAI embeddings works (dimensions: {len(response.data[0].embedding)})")
except Exception as e:
print(f"❌ OpenAI error: {e}")
print("\n🎉 Setup complete! Ready for the baseline project.")Run it:
python test_setup.pyExpected output:
✅ OPENAI_API_KEY configured
✅ ChromaDB works correctly
✅ OpenAI embeddings works (dimensions: 1536)
🎉 Setup complete! Ready for the baseline project.
📦 Implementing the pipeline
Step 1: The indexing pipeline
src/indexing.py:
"""
The indexing pipeline: Chunking → Embeddings → Storage
"""
import chromadb
from openai import OpenAI
from typing import List
import os
from dotenv import load_dotenv
load_dotenv()
class BaselineIndexingPipeline:
"""The baseline indexing pipeline with fixed-size chunking"""
def __init__(self, chunk_size: int = 500):
self.chunk_size = chunk_size
self.openai_client = OpenAI()
self.chroma_client = chromadb.PersistentClient(path="./chroma_db")
def chunk_document(self, document: str) -> List[str]:
"""
Fixed-size chunking (a simple baseline).
Args:
document: the document's full text
Returns:
A list of fixed-size chunks
"""
chunks = []
for i in range(0, len(document), self.chunk_size):
chunk = document[i:i + self.chunk_size]
chunks.append(chunk)
return chunks
def create_embeddings(self, chunks: List[str]) -> List[List[float]]:
"""
Create the embeddings with OpenAI text-embedding-ada-002.
Args:
chunks: a list of text chunks
Returns:
A list of embeddings (1536D vectors)
"""
embeddings = []
for chunk in chunks:
response = self.openai_client.embeddings.create(
model="text-embedding-ada-002",
input=chunk
)
embeddings.append(response.data[0].embedding)
return embeddings
def store_in_chromadb(
self,
chunks: List[str],
embeddings: List[List[float]],
collection_name: str = "fastapi_docs"
):
"""
Store the chunks and embeddings in ChromaDB.
Args:
chunks: a list of text chunks
embeddings: a list of embeddings
collection_name: the collection's name
"""
# Create or fetch the collection
try:
collection = self.chroma_client.get_collection(collection_name)
print(f"Collection '{collection_name}' already exists. Clearing it...")
self.chroma_client.delete_collection(collection_name)
except:
pass
collection = self.chroma_client.create_collection(collection_name)
# Add the documents
collection.add(
documents=chunks,
embeddings=embeddings,
ids=[f"chunk_{i}" for i in range(len(chunks))]
)
print(f"✅ Stored {len(chunks)} chunks in the ChromaDB collection '{collection_name}'")
def index_document(self, document: str, collection_name: str = "fastapi_docs"):
"""
The complete indexing pipeline.
Args:
document: the full document to index
collection_name: the collection's name in ChromaDB
"""
print("🔄 Starting the indexing pipeline...")
# 1. Chunking
print(f" 1/3 Chunking the document (chunk_size={self.chunk_size})...")
chunks = self.chunk_document(document)
print(f" ✅ Created {len(chunks)} chunks")
# 2. Embeddings
print(f" 2/3 Creating the embeddings...")
embeddings = self.create_embeddings(chunks)
print(f" ✅ Created {len(embeddings)} embeddings (1536D each)")
# 3. Storage
print(f" 3/3 Storing them in ChromaDB...")
self.store_in_chromadb(chunks, embeddings, collection_name)
print(f"✅ Indexing pipeline complete!\n")
# Example usage
if __name__ == "__main__":
# Read the sample document
with open("data/fastapi_docs.txt", "r", encoding="utf-8") as f:
document = f.read()
print(f"Document loaded: {len(document)} characters\n")
# Index it
pipeline = BaselineIndexingPipeline(chunk_size=500)
pipeline.index_document(document)Step 2: The retrieval pipeline
src/retrieval.py:
"""
The retrieval pipeline: Query → Embeddings → Search → Top-K
"""
import chromadb
from openai import OpenAI
from typing import List, Dict
from dotenv import load_dotenv
load_dotenv()
class BaselineRetrievalPipeline:
"""The baseline retrieval pipeline with semantic search"""
def __init__(self, collection_name: str = "fastapi_docs"):
self.openai_client = OpenAI()
self.chroma_client = chromadb.PersistentClient(path="./chroma_db")
self.collection = self.chroma_client.get_collection(collection_name)
def create_query_embedding(self, query: str) -> List[float]:
"""
Create the embedding for the user's query.
Args:
query: the query in natural language
Returns:
The embedding (a 1536D vector)
"""
response = self.openai_client.embeddings.create(
model="text-embedding-ada-002",
input=query
)
return response.data[0].embedding
def search(self, query: str, top_k: int = 5) -> Dict:
"""
Semantic search in ChromaDB.
Args:
query: the user's query
top_k: how many documents to retrieve
Returns:
A dict with the documents, IDs and distances
"""
# Create the query's embedding
query_embedding = self.create_query_embedding(query)
# Search in ChromaDB
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k
)
return {
"documents": results['documents'][0],
"ids": results['ids'][0],
"distances": results['distances'][0]
}
def retrieve(self, query: str, top_k: int = 5) -> Dict:
"""
The complete retrieval pipeline.
Args:
query: the user's query
top_k: how many documents to retrieve
Returns:
A dict with the retrieved documents and their metadata
"""
print(f"🔍 Retrieving the top-{top_k} documents for the query: '{query}'\n")
results = self.search(query, top_k)
print(f"✅ Found {len(results['documents'])} relevant documents:")
for i, (doc, distance) in enumerate(zip(results['documents'], results['distances']), 1):
print(f" {i}. (distance: {distance:.3f}) {doc[:100]}...")
print()
return results
# Example usage
if __name__ == "__main__":
retrieval = BaselineRetrievalPipeline()
# Test queries
queries = [
"What is FastAPI?",
"How to create a POST endpoint in FastAPI?",
"FastAPI authentication with OAuth2"
]
for query in queries:
results = retrieval.retrieve(query, top_k=3)
print("-" * 80 + "\n")Step 3: The generation pipeline
src/generation.py:
"""
The generation pipeline: Context + Query → LLM → Response
"""
from openai import OpenAI
from typing import Dict, List
from dotenv import load_dotenv
load_dotenv()
class BaselineGenerationPipeline:
"""The baseline generation pipeline with GPT-3.5-turbo"""
def __init__(self, model: str = "gpt-3.5-turbo"):
self.openai_client = OpenAI()
self.model = model
def create_context(self, documents: List[str]) -> str:
"""
Build the context from the retrieved documents.
Args:
documents: a list of documents (chunks)
Returns:
The formatted context
"""
context = "\n\n".join([
f"Document {i+1}:\n{doc}"
for i, doc in enumerate(documents)
])
return context
def create_prompt(self, query: str, context: str) -> str:
"""
Build the LLM prompt with the context and the query.
Args:
query: the user's query
context: the context from the retrieved documents
Returns:
The complete prompt
"""
prompt = f"""Use the following context to answer the user's question.
Context:
{context}
Question: {query}
Instructions:
- Answer based ONLY on the provided context
- If the context doesn't contain the information, say "I don't have enough information in the context"
- Be concise but complete
- Mention the documents you used (Document 1, Document 2, etc.)
Answer:"""
return prompt
def generate(self, query: str, documents: List[str]) -> Dict:
"""
The complete generation pipeline.
Args:
query: the user's query
documents: the retrieved documents
Returns:
A dict with the answer, its metadata and the token usage
"""
print(f"🤖 Generating the answer with {self.model}...\n")
# Build the context and the prompt
context = self.create_context(documents)
prompt = self.create_prompt(query, context)
# Call the LLM
response = self.openai_client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are a helpful assistant that answers questions based on the provided context."},
{"role": "user", "content": prompt}
],
temperature=0.0 # Deterministic
)
answer = response.choices[0].message.content
tokens_used = response.usage.total_tokens
print(f"✅ Answer generated ({tokens_used} tokens):\n")
print(f"{answer}\n")
return {
"answer": answer,
"model": self.model,
"tokens": tokens_used,
"query": query,
"num_docs": len(documents)
}
# Example usage
if __name__ == "__main__":
from retrieval import BaselineRetrievalPipeline
# Retrieval
retrieval = BaselineRetrievalPipeline()
query = "What is FastAPI?"
results = retrieval.retrieve(query, top_k=3)
# Generation
generation = BaselineGenerationPipeline()
response = generation.generate(query, results['documents'])Step 4: The complete RAG system
src/rag_system.py:
"""
The complete RAG system: Retrieval + Generation, integrated
"""
from retrieval import BaselineRetrievalPipeline
from generation import BaselineGenerationPipeline
import time
from typing import Dict
class BaselineRAGSystem:
"""The complete baseline RAG system"""
def __init__(self):
self.retrieval = BaselineRetrievalPipeline()
self.generation = BaselineGenerationPipeline()
def query(self, user_query: str, top_k: int = 5, verbose: bool = True) -> Dict:
"""
A complete query against the RAG system.
Args:
user_query: the user's question
top_k: how many documents to retrieve
verbose: print the logs
Returns:
A dict with the answer and its metadata
"""
start_time = time.time()
if verbose:
print("=" * 80)
print(f"RAG BASELINE SYSTEM")
print("=" * 80)
print(f"Query: {user_query}\n")
# 1. Retrieval
retrieval_start = time.time()
retrieval_results = self.retrieval.retrieve(user_query, top_k)
retrieval_time = (time.time() - retrieval_start) * 1000 # ms
# 2. Generation
generation_start = time.time()
generation_results = self.generation.generate(
user_query,
retrieval_results['documents']
)
generation_time = (time.time() - generation_start) * 1000 # ms
total_time = (time.time() - start_time) * 1000 # ms
# The complete result
result = {
"query": user_query,
"answer": generation_results['answer'],
"sources": [
{"id": id, "text": doc[:100] + "..."}
for id, doc in zip(retrieval_results['ids'], retrieval_results['documents'])
],
"performance": {
"retrieval_ms": round(retrieval_time, 2),
"generation_ms": round(generation_time, 2),
"total_ms": round(total_time, 2)
},
"metadata": {
"model": generation_results['model'],
"tokens": generation_results['tokens'],
"top_k": top_k
}
}
if verbose:
print("=" * 80)
print(f"Performance:")
print(f" - Retrieval: {result['performance']['retrieval_ms']}ms")
print(f" - Generation: {result['performance']['generation_ms']}ms")
print(f" - Total: {result['performance']['total_ms']}ms")
print("=" * 80 + "\n")
return result
# Example usage
if __name__ == "__main__":
rag = BaselineRAGSystem()
# Test queries
queries = [
"What is FastAPI?",
"How to create a POST endpoint?",
"What is dependency injection in FastAPI?"
]
for query in queries:
result = rag.query(query, top_k=3)
print("\n" + "="*80 + "\n")📊 Evaluation and benchmarking
src/evaluation.py:
"""
Evaluation: measuring the baseline RAG system's performance
"""
from rag_system import BaselineRAGSystem
import time
import statistics
from typing import List, Dict
class RAGEvaluator:
"""A RAG system evaluator"""
def __init__(self, rag_system: BaselineRAGSystem):
self.rag_system = rag_system
def benchmark_latency(self, queries: List[str], top_k: int = 5) -> Dict:
"""
Measure the system's latency across several queries.
Args:
queries: a list of test queries
top_k: how many documents to retrieve
Returns:
A dict with the latency metrics
"""
latencies = []
print(f"🔄 Benchmarking latency with {len(queries)} queries...\n")
for query in queries:
result = self.rag_system.query(query, top_k=top_k, verbose=False)
latencies.append(result['performance']['total_ms'])
# Compute the metrics
p50 = statistics.median(latencies)
p95 = statistics.quantiles(latencies, n=20)[18] if len(latencies) >= 20 else max(latencies)
avg = statistics.mean(latencies)
results = {
"num_queries": len(queries),
"latency_p50_ms": round(p50, 2),
"latency_p95_ms": round(p95, 2),
"latency_avg_ms": round(avg, 2),
"latency_min_ms": round(min(latencies), 2),
"latency_max_ms": round(max(latencies), 2)
}
print(f"✅ Latency Benchmark Results:")
print(f" - P50 (median): {results['latency_p50_ms']}ms")
print(f" - P95: {results['latency_p95_ms']}ms")
print(f" - Average: {results['latency_avg_ms']}ms")
print(f" - Min: {results['latency_min_ms']}ms")
print(f" - Max: {results['latency_max_ms']}ms\n")
return results
def manual_precision_evaluation(
self,
query: str,
top_k: int = 5
) -> float:
"""
Evaluate precision by hand (for the baseline, with no golden dataset).
Args:
query: the test query
top_k: how many documents were retrieved
Returns:
The precision score (0.0 - 1.0)
"""
result = self.rag_system.query(query, top_k=top_k, verbose=False)
print(f"\nQuery: {query}")
print(f"\nRetrieved documents (top-{top_k}):")
for i, source in enumerate(result['sources'], 1):
print(f"{i}. {source['text']}")
# Ask for the manual evaluation
print(f"\nHow many of the {top_k} documents are relevant?")
relevant = int(input("Number of relevant docs (0-5): "))
precision = relevant / top_k
print(f"Precision: {precision:.2%}\n")
return precision
# Example usage
if __name__ == "__main__":
rag = BaselineRAGSystem()
evaluator = RAGEvaluator(rag)
# Test queries
test_queries = [
"What is FastAPI?",
"How to create a POST endpoint?",
"What is dependency injection?",
"How to add authentication?",
"How to handle errors in FastAPI?",
"What is async/await in FastAPI?",
"How to validate request body?",
"What are path parameters?",
"How to return JSON response?",
"What is Pydantic in FastAPI?"
]
# Benchmark the latency
latency_results = evaluator.benchmark_latency(test_queries)
# Manual precision evaluation (3 random queries)
print("\n" + "="*80)
print("Manual Precision Evaluation (3 queries)")
print("="*80)
precisions = []
for query in test_queries[:3]:
precision = evaluator.manual_precision_evaluation(query, top_k=5)
precisions.append(precision)
avg_precision = statistics.mean(precisions)
print(f"\nAverage Precision@5: {avg_precision:.2%}")🎯 Running the project
Step 1: Prepare the dataset
# Download the FastAPI docs (example)
curl https://fastapi.tiangolo.com/ > data/fastapi_docs.txt
# Or build the dataset by hand with ~50-100 paragraphs of docsStep 2: Index the documents
python src/indexing.pyExpected output:
Document loaded: 125000 characters
🔄 Starting the indexing pipeline...
1/3 Chunking the document (chunk_size=500)...
✅ Created 250 chunks
2/3 Creating the embeddings...
✅ Created 250 embeddings (1536D each)
3/3 Storing them in ChromaDB...
✅ Stored 250 chunks in the ChromaDB collection 'fastapi_docs'
✅ Indexing pipeline complete!
Step 3: Test the retrieval
python src/retrieval.pyStep 4: Test the complete system
python src/rag_system.pyStep 5: Evaluate the performance
python src/evaluation.pyExpected output:
🔄 Benchmarking latency with 10 queries...
✅ Latency Benchmark Results:
- P50 (median): 720ms
- P95: 850ms
- Average: 735ms
- Min: 650ms
- Max: 920ms
================================================================================
Manual Precision Evaluation (3 queries)
================================================================================
Query: What is FastAPI?
...
Precision: 80%
Average Precision@5: 70%
📈 The baseline metrics you should expect
Performance targets (baseline):
| Metric | Target | Why |
|---|---|---|
| Latency P50 | 700-800ms | Retrieval (~50ms) + Embeddings (~50ms) + LLM (~600ms) |
| Latency P95 | 800-1,000ms | LLM outliers |
| Precision@5 | 65-75% | Fixed-size chunking + no re-ranking |
| Recall@50 | N/A (manual) | Requires a golden dataset |
| Cost per query | $0.002 | Embeddings ($0.0001) + LLM ($0.002) |
Document your results:
README.md:
# RAG Baseline System - Module 1
## Baseline Performance (my system)
| Metric | Result |
|---------|-----------|
| Latency P50 | 720ms |
| Latency P95 | 850ms |
| Precision@5 | 70% |
| Cost per query | $0.0021 |
## Architecture Decisions
- **Chunking:** Fixed-size (500 chars) - a simple baseline
- **Embeddings:** OpenAI ada-002 - high quality
- **Vector DB:** ChromaDB local - free, plenty for 250 chunks
- **LLM:** GPT-3.5-turbo - a good cost/quality balance
- **Re-ranking:** No (simple baseline)
## Next Steps (Modules 2-8)
1. **Module 2:** Improve the chunking (recursive) → Target: +10% precision
2. **Module 3:** Add query expansion → Target: +15% recall
3. **Module 4:** Implement re-ranking → Target: +20% precision🎯 Success criteria
✅ The project is complete if:
- The RAG system works end-to-end (query → answer)
- The baseline metrics are documented (latency, precision)
- The code is organized into separate modules (indexing, retrieval, generation)
- The environment setup is production-ready (.env, .gitignore, requirements.txt)
- The README covers your architecture decisions and next steps
🚀 Bonus (optional):
- An interactive notebook (Jupyter) for the demo
- Unit tests for each component
- A CLI for interactive queries
- A comparison across several chunk sizes (300, 500, 700)
📚 Project deliverables
- Source code (the complete src/ folder)
- README.md with your decisions and metrics
- requirements.txt with the exact dependencies
- Benchmark results, documented
- A demo notebook (optional)
🎯 Summary
The concepts you applied:
- ✅ The complete pipeline: Indexing → Retrieval → Generation
- ✅ Baseline components: fixed-size chunking, direct query, cosine similarity
- ✅ The stack: ChromaDB + OpenAI + Python
- ✅ Evaluation: latency benchmarking + manual precision
- ✅ Documentation: a README with the architecture and the metrics
The baseline metrics you should expect:
- Latency P50: ~700-800ms
- Precision@5: ~65-75%
- Enough to compare against the improvements in modules 2-8
What's next:
Module 2 optimizes chunking (recursive, semantic, structural) to improve precision by +10-15% over this baseline.
📚 Additional resources
- ChromaDB Getting Started - The official documentation
- OpenAI Embeddings Guide - Best practices
- LangChain RAG Tutorial - The official tutorial
- Building RAG from Scratch - Pinecone's guide
- RAG Evaluation Best Practices - The methodology
- Python Project Structure - Best practices
Created: February 6, 2026
Version: 1.0