Módulo 1: RAG Pipeline Completo (Architecture Overview)
Proyecto: Baseline RAG System
Descripción del proyecto
Este es el mini-proyecto de Módulo 1: construir un sistema RAG baseline completo con ChromaDB + OpenAI. Este baseline sirve como punto de comparación para las optimizaciones que implementarás en módulos 2-8.
El objetivo NO es construir el mejor RAG posible (eso lo harás gradualmente en módulos siguientes), sino establecer un baseline funcional, medido, y documentado. Necesitas números concretos (precision: 68%, latency: 800ms) para validar que técnicas avanzadas realmente mejoran el sistema.
Este proyecto te enseña el flujo completo RAG: (1) Indexing (chunking + embeddings + storage), (2) Retrieval (query + search), (3) Generation (context + LLM), (4) Evaluation (métricas manuales). Al final tendrás un sistema funcional y métricas baseline documentadas.
🎯 Objetivos del Proyecto
Objetivo principal:
Construir sistema RAG baseline con performance medible para comparar con mejoras en módulos 2-8.
Objetivos específicos:
- ✅ Implementar pipeline completo (Indexing → Retrieval → Generation)
- ✅ Usar componentes simples (fixed-size chunking, direct query, cosine similarity)
- ✅ Medir performance baseline (latency, precision, recall)
- ✅ Documentar decisiones de arquitectura
- ✅ Setup environment production-ready (.env, .gitignore, requirements.txt)
📋 Requisitos Técnicos
Stack tecnológico:
- Python: 3.10+
- Vector DB: ChromaDB 0.4.22 (local, gratuito)
- Embeddings: OpenAI text-embedding-ada-002
- LLM: OpenAI GPT-3.5-turbo
- Chunking: Fixed-size (baseline simple)
- Environment: python-dotenv para API keys
Dataset para el proyecto:
Usarás documentación de FastAPI (50-100 páginas) como corpus de prueba.
🛠️ Setup del Proyecto
Paso 1: Estructura de directorios
mkdir rag_baseline_project
cd rag_baseline_project
# Estructura
rag_baseline_project/
├── .env # API keys (NO commitear)
├── .gitignore # Ignorar .env, venv, etc.
├── requirements.txt # Dependencias con versiones
├── data/
│ └── fastapi_docs.txt # Corpus de documentos
├── src/
│ ├── indexing.py # Pipeline de indexing
│ ├── retrieval.py # Pipeline de retrieval
│ ├── generation.py # Pipeline de generation
│ └── evaluation.py # Métricas y benchmarking
├── notebooks/
│ └── rag_baseline_demo.ipynb # Demo interactivo
└── README.md # Documentación del proyectoPaso 2: Instalar dependencias
# Crear virtual environment
python -m venv venv
source venv/bin/activate # En Windows: venv\Scripts\activate
# Instalar dependencias
pip install chromadb==0.4.22 openai==1.12.0 python-dotenv==1.0.0
# Guardar requirements
pip freeze > requirements.txtrequirements.txt:
chromadb==0.4.22
openai==1.12.0
python-dotenv==1.0.0Paso 3: Configurar API keys
.env:
OPENAI_API_KEY=sk-proj-...tu-key-aqui....gitignore:
# Environment
.env
venv/
__pycache__/
*.pyc
# Data (si contiene info sensible)
# data/
# ChromaDB persistence
chroma_db/
# Jupyter notebooks checkpoints
.ipynb_checkpoints/Paso 4: Verificar setup
# test_setup.py
import chromadb
from openai import OpenAI
from dotenv import load_dotenv
import os
load_dotenv()
# Test OpenAI API key
openai_key = os.getenv("OPENAI_API_KEY")
if not openai_key:
print("❌ OPENAI_API_KEY no encontrada en .env")
else:
print("✅ OPENAI_API_KEY configurada")
# Test ChromaDB
try:
client = chromadb.Client()
print("✅ ChromaDB funciona correctamente")
except Exception as e:
print(f"❌ Error con ChromaDB: {e}")
# Test OpenAI embeddings
try:
client = OpenAI()
response = client.embeddings.create(
model="text-embedding-ada-002",
input="Test"
)
print(f"✅ OpenAI embeddings funciona (dimensiones: {len(response.data[0].embedding)})")
except Exception as e:
print(f"❌ Error con OpenAI: {e}")
print("\n🎉 Setup completo! Listo para proyecto baseline.")Ejecutar:
python test_setup.pyOutput esperado:
✅ OPENAI_API_KEY configurada
✅ ChromaDB funciona correctamente
✅ OpenAI embeddings funciona (dimensiones: 1536)
🎉 Setup completo! Listo para proyecto baseline.
📦 Implementación del Pipeline
Paso 1: Indexing Pipeline
src/indexing.py:
"""
Pipeline de Indexing: Chunking → Embeddings → Storage
"""
import chromadb
from openai import OpenAI
from typing import List
import os
from dotenv import load_dotenv
load_dotenv()
class BaselineIndexingPipeline:
"""Pipeline de indexing baseline con 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 (baseline simple).
Args:
document: Texto completo del documento
Returns:
Lista de chunks de tamaño fijo
"""
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]]:
"""
Crear embeddings con OpenAI text-embedding-ada-002.
Args:
chunks: Lista de chunks de texto
Returns:
Lista de embeddings (vectores 1536D)
"""
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"
):
"""
Almacenar chunks y embeddings en ChromaDB.
Args:
chunks: Lista de chunks de texto
embeddings: Lista de embeddings
collection_name: Nombre de la collection
"""
# Crear o obtener collection
try:
collection = self.chroma_client.get_collection(collection_name)
print(f"Collection '{collection_name}' ya existe. Limpiando...")
self.chroma_client.delete_collection(collection_name)
except:
pass
collection = self.chroma_client.create_collection(collection_name)
# Agregar documentos
collection.add(
documents=chunks,
embeddings=embeddings,
ids=[f"chunk_{i}" for i in range(len(chunks))]
)
print(f"✅ Almacenados {len(chunks)} chunks en ChromaDB collection '{collection_name}'")
def index_document(self, document: str, collection_name: str = "fastapi_docs"):
"""
Pipeline completo de indexing.
Args:
document: Documento completo a indexar
collection_name: Nombre de la collection en ChromaDB
"""
print("🔄 Iniciando pipeline de indexing...")
# 1. Chunking
print(f" 1/3 Chunking documento (chunk_size={self.chunk_size})...")
chunks = self.chunk_document(document)
print(f" ✅ Creados {len(chunks)} chunks")
# 2. Embeddings
print(f" 2/3 Creando embeddings...")
embeddings = self.create_embeddings(chunks)
print(f" ✅ Creados {len(embeddings)} embeddings (1536D cada uno)")
# 3. Storage
print(f" 3/3 Almacenando en ChromaDB...")
self.store_in_chromadb(chunks, embeddings, collection_name)
print(f"✅ Pipeline de indexing completo!\n")
# Ejemplo de uso
if __name__ == "__main__":
# Leer documento de ejemplo
with open("data/fastapi_docs.txt", "r", encoding="utf-8") as f:
document = f.read()
print(f"Documento cargado: {len(document)} caracteres\n")
# Indexar
pipeline = BaselineIndexingPipeline(chunk_size=500)
pipeline.index_document(document)Paso 2: Retrieval Pipeline
src/retrieval.py:
"""
Pipeline de Retrieval: 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:
"""Pipeline de retrieval baseline con 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]:
"""
Crear embedding de la query del usuario.
Args:
query: Query en texto natural
Returns:
Embedding (vector 1536D)
"""
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:
"""
Búsqueda semántica en ChromaDB.
Args:
query: Query del usuario
top_k: Número de documentos a recuperar
Returns:
Dict con documentos, IDs, y distancias
"""
# Crear embedding de query
query_embedding = self.create_query_embedding(query)
# Buscar en 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:
"""
Pipeline completo de retrieval.
Args:
query: Query del usuario
top_k: Número de documentos a recuperar
Returns:
Dict con documentos recuperados y metadata
"""
print(f"🔍 Retrieving top-{top_k} documentos para query: '{query}'\n")
results = self.search(query, top_k)
print(f"✅ Encontrados {len(results['documents'])} documentos relevantes:")
for i, (doc, distance) in enumerate(zip(results['documents'], results['distances']), 1):
print(f" {i}. (distance: {distance:.3f}) {doc[:100]}...")
print()
return results
# Ejemplo de uso
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")Paso 3: Generation Pipeline
src/generation.py:
"""
Pipeline de Generation: Context + Query → LLM → Response
"""
from openai import OpenAI
from typing import Dict, List
from dotenv import load_dotenv
load_dotenv()
class BaselineGenerationPipeline:
"""Pipeline de generation baseline con 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:
"""
Crear contexto a partir de documentos recuperados.
Args:
documents: Lista de documentos (chunks)
Returns:
Contexto formateado
"""
context = "\n\n".join([
f"Documento {i+1}:\n{doc}"
for i, doc in enumerate(documents)
])
return context
def create_prompt(self, query: str, context: str) -> str:
"""
Crear prompt para LLM con contexto y query.
Args:
query: Query del usuario
context: Contexto de documentos recuperados
Returns:
Prompt completo
"""
prompt = f"""Usa el siguiente contexto para responder la pregunta del usuario.
Contexto:
{context}
Pregunta: {query}
Instrucciones:
- Responde basándote SOLO en el contexto proporcionado
- Si el contexto no contiene la información, di "No tengo suficiente información en el contexto"
- Sé conciso pero completo
- Menciona los documentos que usaste (Documento 1, Documento 2, etc.)
Respuesta:"""
return prompt
def generate(self, query: str, documents: List[str]) -> Dict:
"""
Pipeline completo de generation.
Args:
query: Query del usuario
documents: Documentos recuperados
Returns:
Dict con respuesta, metadata, y usage
"""
print(f"🤖 Generando respuesta con {self.model}...\n")
# Crear contexto y prompt
context = self.create_context(documents)
prompt = self.create_prompt(query, context)
# Llamar LLM
response = self.openai_client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "Eres un asistente útil que responde preguntas basándose en el contexto proporcionado."},
{"role": "user", "content": prompt}
],
temperature=0.0 # Determinístico
)
answer = response.choices[0].message.content
tokens_used = response.usage.total_tokens
print(f"✅ Respuesta generada ({tokens_used} tokens):\n")
print(f"{answer}\n")
return {
"answer": answer,
"model": self.model,
"tokens": tokens_used,
"query": query,
"num_docs": len(documents)
}
# Ejemplo de uso
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'])Paso 4: Sistema RAG Completo
src/rag_system.py:
"""
Sistema RAG completo: Retrieval + Generation integrado
"""
from retrieval import BaselineRetrievalPipeline
from generation import BaselineGenerationPipeline
import time
from typing import Dict
class BaselineRAGSystem:
"""Sistema RAG baseline completo"""
def __init__(self):
self.retrieval = BaselineRetrievalPipeline()
self.generation = BaselineGenerationPipeline()
def query(self, user_query: str, top_k: int = 5, verbose: bool = True) -> Dict:
"""
Query completa al sistema RAG.
Args:
user_query: Pregunta del usuario
top_k: Número de documentos a recuperar
verbose: Imprimir logs
Returns:
Dict con respuesta y 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
# Resultado completo
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
# Ejemplo de uso
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 y Benchmarking
src/evaluation.py:
"""
Evaluation: Medir performance del sistema RAG baseline
"""
from rag_system import BaselineRAGSystem
import time
import statistics
from typing import List, Dict
class RAGEvaluator:
"""Evaluador de sistema RAG"""
def __init__(self, rag_system: BaselineRAGSystem):
self.rag_system = rag_system
def benchmark_latency(self, queries: List[str], top_k: int = 5) -> Dict:
"""
Medir latency del sistema con múltiples queries.
Args:
queries: Lista de queries de prueba
top_k: Número de documentos a recuperar
Returns:
Dict con métricas de latency
"""
latencies = []
print(f"🔄 Benchmarking latency con {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'])
# Calcular métricas
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:
"""
Evaluar precision manualmente (para baseline, sin golden dataset).
Args:
query: Query de prueba
top_k: Número de documentos recuperados
Returns:
Precision score (0.0 - 1.0)
"""
result = self.rag_system.query(query, top_k=top_k, verbose=False)
print(f"\nQuery: {query}")
print(f"\nDocumentos recuperados (top-{top_k}):")
for i, source in enumerate(result['sources'], 1):
print(f"{i}. {source['text']}")
# Pedir evaluación manual
print(f"\n¿Cuántos de los {top_k} documentos son relevantes?")
relevant = int(input("Número de relevantes (0-5): "))
precision = relevant / top_k
print(f"Precision: {precision:.2%}\n")
return precision
# Ejemplo de uso
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 latency
latency_results = evaluator.benchmark_latency(test_queries)
# Manual precision evaluation (3 queries aleatorias)
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%}")🎯 Ejecución del Proyecto
Paso 1: Preparar dataset
# Descargar FastAPI docs (ejemplo)
curl https://fastapi.tiangolo.com/ > data/fastapi_docs.txt
# O crear dataset manualmente con ~50-100 párrafos de docsPaso 2: Indexar documentos
python src/indexing.pyOutput esperado:
Documento cargado: 125000 caracteres
🔄 Iniciando pipeline de indexing...
1/3 Chunking documento (chunk_size=500)...
✅ Creados 250 chunks
2/3 Creando embeddings...
✅ Creados 250 embeddings (1536D cada uno)
3/3 Almacenando en ChromaDB...
✅ Almacenados 250 chunks en ChromaDB collection 'fastapi_docs'
✅ Pipeline de indexing completo!
Paso 3: Probar retrieval
python src/retrieval.pyPaso 4: Probar sistema completo
python src/rag_system.pyPaso 5: Evaluar performance
python src/evaluation.pyOutput esperado:
🔄 Benchmarking latency con 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%
📈 Métricas Baseline Esperadas
Performance targets (baseline):
| Métrica | Target | Justificación |
|---|---|---|
| Latency P50 | 700-800ms | Retrieval (~50ms) + Embeddings (~50ms) + LLM (~600ms) |
| Latency P95 | 800-1,000ms | Outliers de LLM |
| Precision@5 | 65-75% | Fixed-size chunking + no re-ranking |
| Recall@50 | N/A (manual) | Requiere golden dataset |
| Cost per query | $0.002 | Embeddings ($0.0001) + LLM ($0.002) |
Documentar tus resultados:
README.md:
# RAG Baseline System - Módulo 1
## Performance Baseline (Mi sistema)
| Métrica | Resultado |
|---------|-----------|
| Latency P50 | 720ms |
| Latency P95 | 850ms |
| Precision@5 | 70% |
| Cost per query | $0.0021 |
## Decisiones de Arquitectura
- **Chunking:** Fixed-size (500 chars) - Simple baseline
- **Embeddings:** OpenAI ada-002 - Calidad alta
- **Vector DB:** ChromaDB local - Gratis, suficiente para 250 chunks
- **LLM:** GPT-3.5-turbo - Balance costo/calidad
- **Re-ranking:** No (baseline simple)
## Próximos Pasos (Módulos 2-8)
1. **Módulo 2:** Mejorar chunking (recursive) → Target: +10% precision
2. **Módulo 3:** Agregar query expansion → Target: +15% recall
3. **Módulo 4:** Implementar re-ranking → Target: +20% precision🎯 Criterios de Éxito
✅ Proyecto completo si:
- Sistema RAG funciona end-to-end (query → respuesta)
- Métricas baseline documentadas (latency, precision)
- Código organizado con módulos separados (indexing, retrieval, generation)
- Environment setup production-ready (.env, .gitignore, requirements.txt)
- README con decisiones de arquitectura y próximos pasos
🚀 Bonus (opcional):
- Notebook interactivo (Jupyter) para demo
- Tests unitarios para cada componente
- CLI para queries interactivas
- Comparación con múltiples chunk sizes (300, 500, 700)
📚 Entregables del Proyecto
- Código fuente (src/ folder completo)
- README.md con decisiones y métricas
- requirements.txt con dependencias exactas
- Benchmark results documentados
- Notebook demo (opcional)
🎯 Resumen
Conceptos aplicados:
- ✅ Pipeline completo: Indexing → Retrieval → Generation
- ✅ Componentes baseline: Fixed-size chunking, direct query, cosine similarity
- ✅ Stack: ChromaDB + OpenAI + Python
- ✅ Evaluation: Latency benchmarking + manual precision
- ✅ Documentation: README con arquitectura y métricas
Métricas baseline esperadas:
- Latency P50: ~700-800ms
- Precision@5: ~65-75%
- Suficiente para comparar con mejoras en módulos 2-8
Qué sigue:
Módulo 2 optimiza chunking (recursive, semantic, structural) para mejorar precision +10-15% vs este baseline.
📚 Recursos Adicionales
- ChromaDB Getting Started - Documentación oficial
- OpenAI Embeddings Guide - Best practices
- LangChain RAG Tutorial - Tutorial oficial
- Building RAG from Scratch - Pinecone guide
- RAG Evaluation Best Practices - Metodología
- Python Project Structure - Best practices
Creado: Febrero 6, 2026
Versión: 1.0