Módulo 1: RAG Pipeline Completo (Architecture Overview)

Componentes del RAG Pipeline

Siguiente →

Descripción de la cápsula

RAG no es un monolito mágico. Es un pipeline con 4 componentes distintos que trabajan en secuencia: Indexing (preparar documentos), Retrieval (buscar información relevante), Generation (crear respuesta), y Evaluation (medir calidad). Cada componente tiene decisiones técnicas específicas y puntos de optimización.

Entender estos componentes es crítico porque las técnicas avanzadas que aprenderás en módulos 2-8 se aplican en componentes específicos: chunking optimiza Indexing, re-ranking mejora Retrieval, prompt engineering afecta Generation, y RAGAS mide Evaluation. Sin entender dónde va cada técnica, estarías aplicándolas al azar.

Esta cápsula te da el mapa detallado de cada componente: qué hace, qué decisiones tomas, dónde se aplican técnicas avanzadas, y cómo interactúan entre sí. Al final, podrás dibujar el pipeline completo y explicar el flujo de datos desde documento crudo hasta respuesta final.

🏗️ Arquitectura RAG Completa

Vista de alto nivel:

┌─────────────┐
│  INDEXING   │  ← Preparación offline (una vez)
└──────┬──────┘
       │
       v
┌─────────────┐
│  RETRIEVAL  │  ← Query time (cada request)
└──────┬──────┘
       │
       v
┌─────────────┐
│ GENERATION  │  ← Query time (cada request)
└──────┬──────┘
       │
       v
┌─────────────┐
│ EVALUATION  │  ← Continuo (medir y mejorar)
└─────────────┘

Flujo de datos:

  1. Indexing: Documento → Chunks → Embeddings → Vector DB
  2. Retrieval: User query → Query embedding → Similarity search → Top-K docs
  3. Generation: Top-K docs + Query → Context injection → LLM → Response
  4. Evaluation: Response → Metrics (faithfulness, relevancy) → Feedback loop

📥 Componente 1: Indexing

¿Qué hace?

Prepara documentos para búsqueda semántica: divide en chunks, crea embeddings, almacena en vector DB.

Subcomponentes de Indexing:

1.1 Chunking (División de documentos)

Decisión: ¿Cómo dividir documentos largos?

# Opción A: Fixed-size chunking (naive)
chunks = [document[i:i+500] for i in range(0, len(document), 500)]

# Opción B: Semantic chunking (preserva coherencia)
chunks = semantic_chunker.split(document)  # Divide por temas

# Opción C: Recursive chunking (LangChain)
chunks = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
).split_text(document)

Trade-offs:

StrategyProsConsCuándo usar
Fixed-sizeRápido, simplePierde contexto semánticoPrototipos, docs estructurados
SemanticPreserva coherenciaLento, chunks variablesNarrativo, artículos
RecursiveBalance, overlapComplejidad mediaProduction default

Técnicas avanzadas (Módulo 2):

  • Semantic chunking basado en embeddings
  • Recursive con separators custom
  • Chunking por estructura (HTML, Markdown, JSON)

1.2 Embedding Model (Vectorización)

Decisión: ¿Qué modelo de embeddings usar?

# Opción A: OpenAI (managed, calidad alta)
from openai import OpenAI
client = OpenAI()

embedding = client.embeddings.create(
    model="text-embedding-ada-002",  # 1536 dimensiones
    input="Tu texto aquí"
).data[0].embedding

# Opción B: Local (gratuito, privado)
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')  # 384 dimensiones
embedding = model.encode("Tu texto aquí")

# Opción C: Cohere (multilingüe, calidad)
import cohere
co = cohere.Client("api_key")

embedding = co.embed(
    texts=["Tu texto aquí"],
    model="embed-multilingual-v3.0"  # 1024 dimensiones
).embeddings[0]

Comparación de modelos:

ModeloDimensionesCostoCalidadLatencyMultilingüe
OpenAI ada-0021536$0.0001/1K tokensAlta~50msSí
Cohere multilingual-v31024$0.0001/1K tokensMuy alta~60msExcelente
SentenceTransformers local384-768GratisMedia~20msLimitado

Decisión típica:

  • Prototipo: Local (gratis, rápido)
  • Producción inglés: OpenAI (calidad/costo)
  • Producción multilingüe: Cohere (mejor soporte)

1.3 Vector Storage (Base de datos)

Decisión: ¿Dónde almacenar embeddings?

# Opción A: ChromaDB (local, gratuito)
import chromadb

client = chromadb.Client()
collection = client.create_collection("docs")

collection.add(
    documents=["Texto del chunk"],
    embeddings=[[0.1, 0.2, ...]],  # Vector 1536D
    ids=["chunk_001"]
)

# Opción B: Pinecone (managed, production)
import pinecone

index = pinecone.Index("my-index")

index.upsert(
    vectors=[
        ("chunk_001", [0.1, 0.2, ...], {"text": "Texto del chunk"})
    ]
)

# Opción C: Numpy (in-memory, simple)
import numpy as np

embeddings_matrix = np.array([
    [0.1, 0.2, ...],  # Chunk 1
    [0.3, 0.4, ...],  # Chunk 2
])

Comparación de Vector DBs:

Vector DBSetupCostoPerformanceEscalabilidadCuándo usar
ChromaDBLocal, fácilGratis10K docs: rápido100K+ docs: lentoDev, prototipos
PineconeManaged cloud$70/mes+Siempre rápidoMillones de docsProduction
NumpyIn-memoryGratis<1K docs: okNo escalaTesting

Progresión típica:

  1. Desarrollo: ChromaDB local (módulos 1-6)
  2. Production: Pinecone managed (módulo 7)

Pipeline completo de Indexing:

# indexing_pipeline.py
from openai import OpenAI
import chromadb

def index_documents(documents: list[str]):
    """
    Pipeline de indexing completo.
    
    Input: Lista de documentos crudos
    Output: Vector DB poblado con embeddings
    """
    
    # 1. Chunking (naive fixed-size)
    chunks = []
    for doc in documents:
        doc_chunks = [doc[i:i+500] for i in range(0, len(doc), 500)]
        chunks.extend(doc_chunks)
    
    print(f"✅ Creados {len(chunks)} chunks de {len(documents)} documentos")
    
    # 2. Embeddings (OpenAI)
    client = OpenAI()
    embeddings = []
    
    for chunk in chunks:
        response = client.embeddings.create(
            model="text-embedding-ada-002",
            input=chunk
        )
        embeddings.append(response.data[0].embedding)
    
    print(f"✅ Creados {len(embeddings)} embeddings (1536D cada uno)")
    
    # 3. Storage (ChromaDB)
    chroma_client = chromadb.Client()
    collection = chroma_client.create_collection("my_docs")
    
    collection.add(
        documents=chunks,
        embeddings=embeddings,
        ids=[f"chunk_{i}" for i in range(len(chunks))]
    )
    
    print(f"✅ Almacenados en ChromaDB")
    
    return collection

# Uso
documents = [
    "FastAPI es un framework web moderno para Python...",
    "LangChain es una librería para construir aplicaciones con LLMs...",
    # ... más documentos
]

collection = index_documents(documents)

Output esperado:

✅ Creados 12 chunks de 3 documentos
✅ Creados 12 embeddings (1536D cada uno)
✅ Almacenados en ChromaDB

🔍 Componente 2: Retrieval

¿Qué hace?

Busca los chunks más relevantes para la query del usuario usando similarity search.

Subcomponentes de Retrieval:

2.1 Query Processing (Procesamiento de query)

Sin optimización (baseline):

# Query directa sin procesamiento
user_query = "¿Qué es FastAPI?"

# Crear embedding de la query
query_embedding = client.embeddings.create(
    model="text-embedding-ada-002",
    input=user_query
).data[0].embedding

Con optimización (Módulo 3):

# Query expansion (generar queries similares)
expanded_queries = [
    "¿Qué es FastAPI?",
    "Características de FastAPI",
    "FastAPI framework explicación"
]

# Query rewriting (reformular para claridad)
rewritten_query = llm.invoke(
    f"Reformula esta query para búsqueda: {user_query}"
)

Técnicas avanzadas (Módulo 3):

  • Query expansion (aumentar recall)
  • Query rewriting (claridad)
  • Query decomposition (multi-hop reasoning)
  • HyDE (Hypothetical Document Embeddings)

2.2 Similarity Search (Búsqueda vectorial)

Baseline: Cosine similarity

# ChromaDB similarity search
results = collection.query(
    query_embeddings=[query_embedding],
    n_results=5  # Top-5 documentos
)

# Resultado
{
    'ids': [['chunk_3', 'chunk_7', 'chunk_1', 'chunk_9', 'chunk_5']],
    'distances': [[0.15, 0.18, 0.21, 0.23, 0.25]],  # Cosine distance
    'documents': [['FastAPI es...', 'FastAPI tiene...', ...]]
}

Métricas de distancia:

MétricaFórmulaRangoCuándo usar
Cosine1 - cos(θ)[0, 2]Default (normalizado)
Euclidean||a - b||[0, ∞]Embeddings no normalizados
Dot producta · b[-∞, ∞]Magnitud importa

Decisión típica: Cosine similarity (default en 95% de casos)

2.3 Re-ranking (Mejora de precisión)

Sin re-ranking (baseline):

# Devolver top-5 directamente
top_k_docs = results['documents'][0][:5]

Con re-ranking (Módulo 4):

# Re-rankear con cross-encoder
from sentence_transformers import CrossEncoder

model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

# Re-rankear top-20 → top-5
scores = model.predict([
    (user_query, doc) for doc in results['documents'][0][:20]
])

# Ordenar por score de cross-encoder
reranked_indices = np.argsort(scores)[::-1][:5]
top_k_docs = [results['documents'][0][i] for i in reranked_indices]

Mejora típica con re-ranking:

  • Precision@5: 65% → 85% (+20%)
  • Latency: +150-200ms
  • Costo: +complejidad

Trade-off: Vale la pena en producción si precisión > latencia.

Pipeline completo de Retrieval:

# retrieval_pipeline.py

def retrieve_relevant_docs(
    user_query: str,
    collection,
    top_k: int = 5
) -> list[str]:
    """
    Pipeline de retrieval completo.
    
    Input: Query del usuario
    Output: Top-K documentos más relevantes
    """
    
    # 1. Query embedding
    client = OpenAI()
    query_embedding = client.embeddings.create(
        model="text-embedding-ada-002",
        input=user_query
    ).data[0].embedding
    
    print(f"✅ Query embedding creado (1536D)")
    
    # 2. Similarity search
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k
    )
    
    docs = results['documents'][0]
    distances = results['distances'][0]
    
    print(f"✅ Encontrados {len(docs)} documentos relevantes")
    print(f"   Distancias: {[f'{d:.3f}' for d in distances]}")
    
    return docs

# Uso
docs = retrieve_relevant_docs(
    user_query="¿Qué es FastAPI?",
    collection=collection,
    top_k=5
)

for i, doc in enumerate(docs, 1):
    print(f"{i}. {doc[:100]}...")

Output esperado:

✅ Query embedding creado (1536D)
✅ Encontrados 5 documentos relevantes
   Distancias: ['0.150', '0.180', '0.210', '0.230', '0.250']

1. FastAPI es un framework web moderno y rápido para Python...
2. FastAPI tiene validación automática de datos con Pydantic...
3. FastAPI genera documentación automática con Swagger UI...
4. FastAPI es async-first, soporta async/await nativo...
5. FastAPI es usado por Microsoft, Netflix, y Uber...

🤖 Componente 3: Generation

¿Qué hace?

Genera respuesta usando LLM con contexto de documentos recuperados.

Subcomponentes de Generation:

3.1 Context Injection (Inyectar contexto)

Pattern básico:

# Construir contexto desde documentos recuperados
context = "\n\n".join([
    f"Documento {i+1}: {doc}"
    for i, doc in enumerate(docs)
])

# Template de prompt
prompt_template = f"""
Usa el siguiente contexto para responder la pregunta.

Contexto:
{context}

Pregunta: {user_query}

Respuesta:
"""

Pattern avanzado (con metadata):

# Incluir metadata en contexto
context_with_metadata = "\n\n".join([
    f"[Fuente: {doc['source']}, Fecha: {doc['date']}]\n{doc['text']}"
    for doc in docs_with_metadata
])

3.2 LLM Prompting (Generación)

Baseline prompt:

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-3.5-turbo",
    messages=[
        {"role": "system", "content": "Eres un asistente útil que responde preguntas basándote en el contexto proporcionado."},
        {"role": "user", "content": prompt_template}
    ],
    temperature=0.0  # Determinístico
)

answer = response.choices[0].message.content

Advanced prompt (con instrucciones):

system_prompt = """
Eres un asistente técnico experto.

Instrucciones:
1. Responde SOLO basándote en el contexto proporcionado
2. Si el contexto no contiene la información, di "No tengo suficiente información"
3. Cita los documentos que usaste (Documento 1, Documento 2, etc.)
4. Sé conciso pero completo
5. Si hay información contradictoria, menciona ambas versiones
"""

response = client.chat.completions.create(
    model="gpt-4-turbo",  # Mejor calidad
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": prompt_template}
    ],
    temperature=0.0
)

3.3 Response Formatting (Formato de respuesta)

Response structure:

# Estructura de respuesta completa
response_obj = {
    "answer": answer,
    "sources": [doc['id'] for doc in docs],
    "confidence": calculate_confidence(answer, docs),
    "model_used": "gpt-3.5-turbo",
    "tokens_used": response.usage.total_tokens,
    "latency_ms": latency
}

Pipeline completo de Generation:

# generation_pipeline.py

def generate_answer(
    user_query: str,
    retrieved_docs: list[str]
) -> dict:
    """
    Pipeline de generation completo.
    
    Input: Query + documentos recuperados
    Output: Respuesta generada por LLM
    """
    
    # 1. Context injection
    context = "\n\n".join([
        f"Documento {i+1}: {doc}"
        for i, doc in enumerate(retrieved_docs)
    ])
    
    prompt = f"""
Usa el siguiente contexto para responder la pregunta.

Contexto:
{context}

Pregunta: {user_query}

Respuesta:
"""
    
    # 2. LLM generation
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "system", "content": "Eres un asistente útil. Responde basándote solo en el contexto proporcionado."},
            {"role": "user", "content": prompt}
        ],
        temperature=0.0
    )
    
    answer = response.choices[0].message.content
    tokens = response.usage.total_tokens
    
    print(f"✅ Respuesta generada ({tokens} tokens)")
    
    # 3. Response formatting
    return {
        "answer": answer,
        "sources": [f"Doc {i+1}" for i in range(len(retrieved_docs))],
        "model": "gpt-3.5-turbo",
        "tokens": tokens
    }

# Uso
result = generate_answer(
    user_query="¿Qué es FastAPI?",
    retrieved_docs=docs
)

print(f"Respuesta: {result['answer']}")
print(f"Fuentes: {result['sources']}")
print(f"Tokens: {result['tokens']}")

Output esperado:

✅ Respuesta generada (234 tokens)

Respuesta: FastAPI es un framework web moderno y rápido para Python, diseñado para construir APIs con alta performance. Sus características principales incluyen validación automática de datos con Pydantic, generación automática de documentación con Swagger UI, y soporte nativo para async/await. Es usado por empresas como Microsoft, Netflix y Uber en producción.

Fuentes: ['Doc 1', 'Doc 2', 'Doc 3']
Tokens: 234

📊 Componente 4: Evaluation

¿Qué hace?

Mide calidad del sistema RAG con métricas objetivas para identificar mejoras.

Métricas de Evaluation:

4.1 Retrieval Metrics (Calidad de búsqueda)

# Precision: ¿Cuántos de los recuperados son relevantes?
precision = relevant_retrieved / total_retrieved

# Recall: ¿Cuántos de los relevantes fueron recuperados?
recall = relevant_retrieved / total_relevant

# Ejemplo
# Total relevant docs en DB: 10
# Retrieved docs: 5
# Relevant en retrieved: 4

precision = 4 / 5  # 0.80 (80% de lo recuperado es relevante)
recall = 4 / 10    # 0.40 (solo encontramos 40% de los relevantes)

4.2 Generation Metrics (Calidad de respuesta)

Faithfulness (Groundedness):

# ¿La respuesta está basada en el contexto?
# Score: 0.0 (inventada) a 1.0 (completamente grounded)

# Ejemplo
context = "FastAPI es un framework web."
answer_grounded = "FastAPI es un framework web."  # Faithfulness: 1.0
answer_hallucinated = "FastAPI fue creado en 2015."  # Faithfulness: 0.0

Answer Relevancy:

# ¿La respuesta contesta la pregunta?
# Score: 0.0 (irrelevante) a 1.0 (perfectamente relevante)

# Ejemplo
question = "¿Qué es FastAPI?"
answer_relevant = "FastAPI es un framework web."  # Relevancy: 1.0
answer_irrelevant = "Python es un lenguaje."  # Relevancy: 0.3

Evaluation con RAGAS (Módulo 8):

# Evaluation automática con RAGAS
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy

# Dataset de evaluación
dataset = {
    "question": ["¿Qué es FastAPI?"],
    "answer": ["FastAPI es un framework web..."],
    "contexts": [["FastAPI es un framework web moderno..."]],
    "ground_truth": ["FastAPI es un framework web de Python"]
}

# Evaluar
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy]
)

print(results)
# Output:
# {
#   'faithfulness': 0.95,
#   'answer_relevancy': 0.92
# }

Interpretación de scores:

ScoreInterpretaciónAcción
0.90+ExcelenteProduction-ready
0.80-0.89BuenoOptimizaciones menores
0.70-0.79AceptableRevisar chunking/prompts
<0.70Necesita mejoraRediseñar pipeline

🔄 Feedback Loop (Mejora continua)

┌──────────────┐
│   Query      │
└──────┬───────┘
       │
       v
┌──────────────┐
│   Retrieval  │
└──────┬───────┘
       │
       v
┌──────────────┐
│  Generation  │
└──────┬───────┘
       │
       v
┌──────────────┐
│  Evaluation  │ ← Medir faithfulness, relevancy
└──────┬───────┘
       │
       │ (Si score <0.80)
       v
┌──────────────┐
│  Mejoras     │ ← Ajustar chunking, prompts, re-ranking
└──────┬───────┘
       │
       └───────> Loop back

🎯 Resumen

Conceptos clave:

  • ✅ RAG tiene 4 componentes: Indexing, Retrieval, Generation, Evaluation
  • ✅ Indexing: Chunking → Embeddings → Vector DB (offline, una vez)
  • ✅ Retrieval: Query embedding → Similarity search → Top-K docs (query time)
  • ✅ Generation: Context injection → LLM prompting → Response (query time)
  • ✅ Evaluation: Retrieval metrics + Generation metrics → Feedback loop (continuo)
  • ✅ Técnicas avanzadas se aplican en componentes específicos (chunking en Indexing, re-ranking en Retrieval)
  • ✅ Pipeline completo: documento → respuesta en 4 pasos claros

Qué sigue:

Cápsula 03 te enseña decisiones de arquitectura: qué chunking strategy usar, qué embeddings elegir, qué vector DB seleccionar, y cuándo aplicar re-ranking. Decisiones técnicas basadas en trade-offs reales.

📚 Recursos Adicionales

  1. LangChain Components Docs - Documentación oficial de componentes
  2. ChromaDB Architecture - Arquitectura de vector DB
  3. OpenAI Embeddings Best Practices - Guía oficial
  4. RAGAS Evaluation Framework - Evaluation metrics para RAG
  5. Building Production RAG (Video) - Arquitectura en producción
  6. RAG Components Deep Dive - Artículo técnico de Pinecone

Creado: Febrero 6, 2026
Versión: 1.0

← AnteriorIntroducción al Módulo: RAG Pipeline CompletoSiguiente →Decisiones de Arquitectura RAG