Módulo 6: RAG Multimodal
8. Proyecto: Multimodal RAG Básico
Descripción
Este proyecto integra todo lo que aprendiste en el módulo: embeddings multimodales, indexación en vector stores, retrieval híbrido, procesamiento de documentos con imágenes, y generación de respuestas con LLM. Vas a construir un sistema completo de RAG multimodal que toma PDFs con texto e imágenes, los procesa, los indexa en ChromaDB, recupera fragmentos relevantes ante una pregunta, y genera respuestas citando las fuentes.
El sistema tiene cuatro componentes: un procesador de documentos que extrae texto e imágenes de PDFs, un indexador que genera embeddings y los almacena en ChromaDB, un retriever híbrido que busca por similitud semántica con filtros de metadata, y un generador de respuestas que usa un LLM con el contexto recuperado.
Lo que construyes: Un pipeline end-to-end que recibe PDFs, los procesa preservando la relación texto-imagen, indexa el contenido, y responde preguntas basándose en lo indexado.
Conexión con el módulo: Usa directamente los embeddings de la cápsula 02, la indexación de la cápsula 03, el retrieval de la cápsula 04, el procesamiento de documentos de la cápsula 05, y opcionalmente LangChain de la cápsula 06. Las optimizaciones de la cápsula 07 se aplican como extensiones.
Especificaciones
Input
- Uno o más documentos PDF con texto e imágenes
- Pregunta del usuario en lenguaje natural
Output
- Respuesta generada por el LLM basada exclusivamente en el contenido recuperado
- Lista de fuentes utilizadas (documento, página, tipo de contenido)
- Métricas del pipeline (chunks procesados, imágenes descritas, tiempo)
Requisitos
Funcionales:
1. Procesar PDFs extrayendo texto e imágenes por página
2. Filtrar imágenes irrelevantes (logos, iconos, duplicados)
3. Describir imágenes relevantes con Vision API
4. Generar chunks con contexto de imagen asociado
5. Indexar chunks en ChromaDB con metadata
6. Recuperar chunks relevantes ante una query
7. Generar respuesta con LLM usando los chunks como contexto
8. Retornar fuentes y métricas junto con la respuesta
No funcionales:
- Cache de descripciones de imágenes
- Manejo de errores sin detener el pipeline completo
- Métricas de costo y tiempo por operación
- Soporte para múltiples documentos en la misma colección
Setup
pip install openai chromadb pymupdf pillow python-dotenvimport os
from dotenv import load_dotenv
load_dotenv()
assert os.getenv("OPENAI_API_KEY"), "Falta OPENAI_API_KEY en .env"Paso 1: Cache de Descripciones
import json
import hashlib
from pathlib import Path
from datetime import datetime
class DescriptionCache:
def __init__(self, cache_dir: str = "./description_cache"):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.hits = 0
self.misses = 0
def _hash(self, image_data: bytes) -> str:
return hashlib.sha256(image_data).hexdigest()
def get(self, image_data: bytes) -> str | None:
cache_file = self.cache_dir / f"{self._hash(image_data)}.json"
if cache_file.exists():
self.hits += 1
return json.loads(cache_file.read_text()).get("description")
self.misses += 1
return None
def set(self, image_data: bytes, description: str, model: str = "") -> None:
img_hash = self._hash(image_data)
data = {
"description": description,
"model": model,
"hash": img_hash,
"created_at": datetime.now().isoformat(),
}
(self.cache_dir / f"{img_hash}.json").write_text(
json.dumps(data, ensure_ascii=False, indent=2)
)
def stats(self) -> dict:
total = self.hits + self.misses
return {
"hits": self.hits,
"misses": self.misses,
"hit_rate": f"{self.hits / max(total, 1):.1%}",
"cached_files": len(list(self.cache_dir.glob("*.json"))),
}Paso 2: Procesador de Documentos
El procesador toma un PDF y produce una lista de chunks. Cada chunk puede ser texto puro o una descripción de imagen.
import fitz
import base64
from openai import OpenAI
oai_client = OpenAI()
MIN_IMAGE_WIDTH = 100
MIN_IMAGE_HEIGHT = 100
MIN_IMAGE_BYTES = 5000
class DocumentProcessor:
def __init__(
self,
cache: DescriptionCache,
output_dir: str = "./extracted_images",
max_chunk_words: int = 500,
vision_model: str = "gpt-4o-mini",
):
self.cache = cache
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.max_chunk_words = max_chunk_words
self.vision_model = vision_model
self.stats = {
"pages": 0, "images_extracted": 0,
"images_filtered": 0, "images_described": 0,
"text_chunks": 0, "image_chunks": 0,
}
def process(self, pdf_path: str) -> list[dict]:
doc = fitz.open(pdf_path)
pdf_name = Path(pdf_path).stem
all_chunks = []
for page_num in range(len(doc)):
page = doc[page_num]
self.stats["pages"] += 1
text = page.get_text().strip()
if text:
all_chunks.extend(self._split_text(text, page_num + 1, pdf_path))
all_chunks.extend(
self._process_images(doc, page, page_num, pdf_name, pdf_path)
)
doc.close()
for i, chunk in enumerate(all_chunks):
chunk["chunk_index"] = i
return all_chunks
def _split_text(self, text: str, page: int, source: str) -> list[dict]:
words = text.split()
chunks = []
start = 0
overlap = 50
while start < len(words):
end = start + self.max_chunk_words
chunk_text = " ".join(words[start:end])
chunks.append({
"text": chunk_text, "page": page,
"source": source, "type": "text",
"word_count": len(chunk_text.split()),
})
self.stats["text_chunks"] += 1
start += self.max_chunk_words - overlap
return chunks
def _process_images(
self, doc, page, page_num: int, pdf_name: str, source: str
) -> list[dict]:
chunks = []
for img_idx, img_ref in enumerate(page.get_images()):
xref = img_ref[0]
try:
base_image = doc.extract_image(xref)
if not base_image or not base_image.get("image"):
continue
image_data = base_image["image"]
width = base_image.get("width", 0)
height = base_image.get("height", 0)
if (width < MIN_IMAGE_WIDTH or height < MIN_IMAGE_HEIGHT
or len(image_data) < MIN_IMAGE_BYTES):
self.stats["images_filtered"] += 1
continue
self.stats["images_extracted"] += 1
ext = base_image.get("ext", "png")
filename = f"{pdf_name}_p{page_num + 1}_img{img_idx}.{ext}"
filepath = self.output_dir / filename
filepath.write_bytes(image_data)
description = self._describe_image(image_data, ext)
self.stats["images_described"] += 1
chunks.append({
"text": description, "page": page_num + 1,
"source": source, "type": "image",
"image_path": str(filepath),
})
except Exception:
continue
return chunks
def _describe_image(self, image_data: bytes, ext: str) -> str:
cached = self.cache.get(image_data)
if cached:
return cached
b64 = base64.b64encode(image_data).decode("utf-8")
mime = f"image/{ext}" if ext != "jpg" else "image/jpeg"
response = oai_client.chat.completions.create(
model=self.vision_model,
messages=[{
"role": "user",
"content": [
{
"type": "text",
"text": (
"Describe esta imagen en 2-3 oraciones para indexación "
"en búsqueda semántica. Incluye tipo de contenido, "
"elementos principales, y datos específicos visibles."
),
},
{"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}},
],
}],
max_tokens=150,
)
description = response.choices[0].message.content
self.cache.set(image_data, description, self.vision_model)
return descriptionPaso 3: Indexador con ChromaDB
import chromadb
from chromadb.utils import embedding_functions
class MultimodalIndexer:
def __init__(
self,
persist_directory: str = "./chroma_db",
collection_name: str = "multimodal_rag",
embedding_model: str = "text-embedding-3-small",
):
self.collection_name = collection_name
self.ef = embedding_functions.OpenAIEmbeddingFunction(
api_key=os.getenv("OPENAI_API_KEY"),
model_name=embedding_model,
)
self.chroma_client = chromadb.PersistentClient(path=persist_directory)
self.collection = self.chroma_client.get_or_create_collection(
name=collection_name, embedding_function=self.ef,
)
self.stats = {"indexed": 0, "errors": 0}
def index_chunks(self, chunks: list[dict], batch_size: int = 100) -> dict:
for i in range(0, len(chunks), batch_size):
batch = chunks[i:i + batch_size]
documents, ids, metadatas = [], [], []
for chunk in batch:
chunk_id = (
f"{Path(chunk.get('source', 'unknown')).stem}"
f"_{chunk['type']}_{chunk.get('chunk_index', 0)}"
)
documents.append(chunk["text"])
ids.append(chunk_id)
metadatas.append({
"type": chunk["type"],
"source": chunk.get("source", ""),
"page": chunk.get("page", 0),
"image_path": chunk.get("image_path", ""),
})
try:
self.collection.upsert(
documents=documents, ids=ids, metadatas=metadatas,
)
self.stats["indexed"] += len(batch)
except Exception as e:
print(f"Error indexando batch {i}: {e}")
self.stats["errors"] += 1
return self.stats
def get_info(self) -> dict:
count = self.collection.count()
if count == 0:
return {"total": 0, "by_type": {}}
sample = self.collection.get(limit=min(count, 1000), include=["metadatas"])
type_counts = {}
for meta in sample["metadatas"]:
t = meta.get("type", "unknown")
type_counts[t] = type_counts.get(t, 0) + 1
return {"total": count, "by_type": type_counts}
def clear(self) -> None:
self.chroma_client.delete_collection(self.collection_name)
self.collection = self.chroma_client.get_or_create_collection(
name=self.collection_name, embedding_function=self.ef,
)Paso 4: Retriever Híbrido
Busca chunks relevantes combinando similitud semántica con filtros de metadata.
class HybridRetriever:
def __init__(self, indexer: MultimodalIndexer, default_k: int = 5):
self.collection = indexer.collection
self.default_k = default_k
def retrieve(
self, query: str, k: int | None = None,
content_type: str | None = None,
) -> list[dict]:
k = k or self.default_k
params = {
"query_texts": [query], "n_results": k,
"include": ["documents", "metadatas", "distances"],
}
if content_type:
params["where"] = {"type": {"$eq": content_type}}
results = self.collection.query(**params)
retrieved = []
for i in range(len(results["ids"][0])):
similarity = round(1 - results["distances"][0][i], 4)
retrieved.append({
"id": results["ids"][0][i],
"text": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"similarity": similarity,
})
return retrieved
def retrieve_hybrid(
self, query: str, text_k: int = 4, image_k: int = 2,
) -> list[dict]:
text_results = self.retrieve(query, k=text_k, content_type="text")
image_results = self.retrieve(query, k=image_k, content_type="image")
combined = text_results + image_results
combined.sort(key=lambda x: x["similarity"], reverse=True)
seen = set()
return [r for r in combined if r["id"] not in seen and not seen.add(r["id"])]
def retrieve_smart(self, query: str, k: int | None = None) -> list[dict]:
visual_keywords = {
"diagrama", "imagen", "figura", "gráfico", "tabla",
"captura", "foto", "muestra", "visualiza",
}
if set(query.lower().split()) & visual_keywords:
return self.retrieve_hybrid(query, text_k=2, image_k=4)
return self.retrieve_hybrid(query, text_k=k or self.default_k, image_k=1)Paso 5: Generador de Respuestas
class AnswerGenerator:
def __init__(self, model: str = "gpt-4o", max_tokens: int = 800):
self.client = OpenAI()
self.model = model
self.max_tokens = max_tokens
self.stats = {"queries": 0, "total_tokens": 0}
def generate(self, query: str, chunks: list[dict]) -> dict:
if not chunks:
return {
"answer": "No encontré información relevante para responder.",
"sources": [], "tokens": {},
}
context = self._format_context(chunks)
sources = self._extract_sources(chunks)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": (
"Responde basándote EXCLUSIVAMENTE en el contexto proporcionado.\n"
"El contexto incluye texto y descripciones de imágenes de documentos.\n"
"Si una imagen contiene información relevante, menciónala.\n"
"Si no hay información suficiente, dilo claramente.\n"
"Al final, lista las fuentes: [Fuente: documento, página X, tipo]."
),
},
{"role": "user", "content": f"Contexto:\n{context}\n\nPregunta: {query}"},
],
max_tokens=self.max_tokens,
temperature=0,
)
usage = response.usage
self.stats["queries"] += 1
self.stats["total_tokens"] += usage.total_tokens if usage else 0
return {
"answer": response.choices[0].message.content,
"sources": sources,
"tokens": {
"prompt": usage.prompt_tokens if usage else 0,
"completion": usage.completion_tokens if usage else 0,
"total": usage.total_tokens if usage else 0,
},
}
def _format_context(self, chunks: list[dict]) -> str:
parts = []
for i, chunk in enumerate(chunks):
meta = chunk.get("metadata", {})
header = (
f"[FRAGMENTO {i+1} | {meta.get('type','text').upper()} | "
f"{Path(meta.get('source','')).name} p.{meta.get('page','?')} | "
f"sim: {chunk.get('similarity', 0)}]"
)
parts.append(f"{header}\n{chunk['text']}")
return "\n\n---\n\n".join(parts)
def _extract_sources(self, chunks: list[dict]) -> list[dict]:
return [{
"source": Path(c.get("metadata", {}).get("source", "")).name,
"page": c.get("metadata", {}).get("page", 0),
"type": c.get("metadata", {}).get("type", "text"),
"similarity": c.get("similarity", 0),
} for c in chunks]Paso 6: Pipeline Completo
import time
class MultimodalRAGPipeline:
def __init__(
self,
persist_dir: str = "./chroma_db",
collection_name: str = "multimodal_rag",
image_dir: str = "./extracted_images",
cache_dir: str = "./description_cache",
chunk_words: int = 500,
llm_model: str = "gpt-4o",
vision_model: str = "gpt-4o-mini",
):
self.cache = DescriptionCache(cache_dir)
self.processor = DocumentProcessor(
cache=self.cache, output_dir=image_dir,
max_chunk_words=chunk_words, vision_model=vision_model,
)
self.indexer = MultimodalIndexer(
persist_directory=persist_dir, collection_name=collection_name,
)
self.retriever = HybridRetriever(self.indexer)
self.generator = AnswerGenerator(model=llm_model)
self.metrics = {
"docs_processed": 0, "total_chunks": 0,
"indexing_time": 0.0, "queries": 0, "avg_query_time": 0.0,
}
def index_document(self, pdf_path: str) -> dict:
start = time.time()
chunks = self.processor.process(pdf_path)
index_stats = self.indexer.index_chunks(chunks)
elapsed = time.time() - start
self.metrics["docs_processed"] += 1
self.metrics["total_chunks"] += len(chunks)
self.metrics["indexing_time"] += elapsed
return {
"document": pdf_path, "chunks": len(chunks),
"stats": index_stats, "time_seconds": round(elapsed, 2),
}
def index_directory(self, dir_path: str, glob: str = "*.pdf") -> dict:
results = []
for pdf_path in Path(dir_path).glob(glob):
result = self.index_document(str(pdf_path))
results.append(result)
print(f" Indexado: {pdf_path.name} ({result['chunks']} chunks, {result['time_seconds']}s)")
return {
"documents": len(results),
"total_chunks": sum(r["chunks"] for r in results),
"details": results,
}
def query(self, question: str, k: int = 5, smart: bool = True) -> dict:
start = time.time()
retrieved = (
self.retriever.retrieve_smart(question, k=k)
if smart
else self.retriever.retrieve(question, k=k)
)
result = self.generator.generate(question, retrieved)
elapsed = time.time() - start
self.metrics["queries"] += 1
total_time = self.metrics.get("_total_q_time", 0) + elapsed
self.metrics["_total_q_time"] = total_time
self.metrics["avg_query_time"] = round(total_time / self.metrics["queries"], 2)
return {
"question": question,
"answer": result["answer"],
"sources": result["sources"],
"retrieved_chunks": len(retrieved),
"tokens": result["tokens"],
"time_seconds": round(elapsed, 2),
}
def interactive(self) -> None:
print("=== Multimodal RAG — Sesión Interactiva ===")
print(f"Colección: {self.indexer.get_info()}")
print("Escribe 'salir' para terminar.\n")
while True:
question = input("Pregunta: ").strip()
if question.lower() in ("salir", "exit", "quit"):
break
if not question:
continue
result = self.query(question)
print(f"\nRespuesta:\n{result['answer']}\n")
for src in result["sources"]:
print(f" - {src['source']} p.{src['page']} ({src['type']})")
print(f"Tiempo: {result['time_seconds']}s | Tokens: {result['tokens'].get('total', 0)}\n")
def get_metrics(self) -> dict:
return {
**{k: v for k, v in self.metrics.items() if not k.startswith("_")},
"collection": self.indexer.get_info(),
"cache": self.cache.stats(),
}Uso completo
pipeline = MultimodalRAGPipeline(chunk_words=500, llm_model="gpt-4o")
# --- Indexar ---
# result = pipeline.index_document("docs/manual_tecnico.pdf")
# print(f"Indexado: {result['chunks']} chunks en {result['time_seconds']}s")
# --- Consultar ---
# answer = pipeline.query("¿Cómo se conectan los microservicios?")
# print(answer["answer"])
# --- Sesión interactiva ---
# pipeline.interactive()
# --- Métricas ---
# print(pipeline.get_metrics())Extensión 1: Reranking de Resultados
El retriever por similitud coseno devuelve resultados razonables, pero un reranker mejora la precisión reordenando los candidatos con un segundo modelo.
class Reranker:
def __init__(self, model: str = "gpt-4o-mini"):
self.client = OpenAI()
self.model = model
def rerank(self, query: str, chunks: list[dict], top_k: int = 3) -> list[dict]:
if len(chunks) <= top_k:
return chunks
candidates = "\n".join(
f"[{i}] {c['text'][:200].replace(chr(10), ' ')}"
for i, c in enumerate(chunks)
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": (
"Dada una pregunta y fragmentos de documentos, devuelve los "
"índices de los fragmentos MÁS relevantes de mayor a menor. "
"Responde SOLO con índices separados por comas. Ejemplo: 2,0,4"
),
},
{
"role": "user",
"content": (
f"Pregunta: {query}\n\nFragmentos:\n{candidates}\n\n"
f"Top {top_k} (solo índices):"
),
},
],
max_tokens=50, temperature=0,
)
try:
indices = [int(x.strip()) for x in response.choices[0].message.content.strip().split(",")]
indices = [i for i in indices if 0 <= i < len(chunks)][:top_k]
except ValueError:
return chunks[:top_k]
return [chunks[i] for i in indices]
def query_with_reranking(pipeline, question: str, initial_k: int = 10, final_k: int = 3) -> dict:
retrieved = pipeline.retriever.retrieve_smart(question, k=initial_k)
reranked = Reranker().rerank(question, retrieved, top_k=final_k)
result = pipeline.generator.generate(question, reranked)
return {
"question": question, "answer": result["answer"],
"initial_candidates": len(retrieved), "after_reranking": len(reranked),
"sources": result["sources"],
}Extensión 2: Image Q&A Directo
Cuando el retriever encuentra una imagen relevante y el archivo original está disponible, puedes enviarlo directamente al LLM para análisis más profundo.
class ImageQA:
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
def answer_with_image(self, question: str, image_path: str, text_context: str = "") -> str:
image_data = Path(image_path).read_bytes()
ext = Path(image_path).suffix.lstrip(".")
mime = f"image/{ext}" if ext != "jpg" else "image/jpeg"
b64 = base64.b64encode(image_data).decode("utf-8")
user_text = (
f"Contexto:\n{text_context}\n\nPregunta: {question}"
if text_context
else f"Pregunta sobre esta imagen: {question}"
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "Responde basándote en la imagen y el contexto. Describe elementos relevantes.",
},
{
"role": "user",
"content": [
{"type": "text", "text": user_text},
{"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}},
],
},
],
max_tokens=500,
)
return response.choices[0].message.content
def query_with_image_qa(pipeline, question: str, k: int = 5) -> dict:
retrieved = pipeline.retriever.retrieve_smart(question, k=k)
image_chunks = [
r for r in retrieved
if r["metadata"].get("type") == "image"
and r["metadata"].get("image_path")
and Path(r["metadata"]["image_path"]).exists()
]
text_chunks = [r for r in retrieved if r["metadata"].get("type") == "text"]
text_context = "\n\n".join(c["text"] for c in text_chunks[:3])
base_result = pipeline.generator.generate(question, retrieved)
image_answers = []
qa = ImageQA()
for img_chunk in image_chunks[:2]:
answer = qa.answer_with_image(question, img_chunk["metadata"]["image_path"], text_context)
image_answers.append({
"image": img_chunk["metadata"]["image_path"],
"page": img_chunk["metadata"]["page"],
"answer": answer,
})
return {
"question": question,
"text_answer": base_result["answer"],
"image_answers": image_answers,
"sources": base_result["sources"],
}Troubleshooting
ChromaDB "Collection already exists" al re-ejecutar
def reset_pipeline(pipeline: MultimodalRAGPipeline) -> None:
pipeline.indexer.clear()
print("Colección limpiada. Puedes re-indexar.")Imágenes no se extraen de ciertos PDFs
Algunos PDFs tienen imágenes como vectores SVG o parte del renderizado. PyMuPDF solo extrae raster. Como alternativa, renderiza la página completa como imagen.
def extract_page_as_image_fallback(pdf_path: str, page_num: int, dpi: int = 150) -> bytes:
doc = fitz.open(pdf_path)
page = doc[page_num]
pix = page.get_pixmap(matrix=fitz.Matrix(dpi / 72, dpi / 72))
image_bytes = pix.tobytes("png")
doc.close()
return image_bytesEl LLM alucina a pesar del contexto
Verifica que los chunks recuperados son relevantes (problema de retrieval, no de generación). Baja la temperatura a 0 y usa un prompt más estricto. Si persiste, aumenta k para dar más contexto.
Queries lentas (>10 segundos)
1. Usar gpt-4o-mini en lugar de gpt-4o (3-5x más rápido)
2. Reducir max_tokens del generador
3. Reducir k (menos chunks en contexto)
4. Pre-computar respuestas para queries frecuentes
Descripciones de imagen genéricas
Usa gpt-4o en lugar de gpt-4o-mini para el vision_model del procesador. Es más caro pero produce descripciones más detalladas.
Error de memoria con muchos documentos
ChromaDB PersistentClient mantiene índices en memoria. Para >100K chunks, usa un vector store dedicado (Pinecone, Weaviate). Para el alcance de este proyecto, ChromaDB es suficiente para cientos de documentos.
Checklist
-
DescriptionCacheimplementado y persistiendo a disco -
DocumentProcessorextrae texto e imágenes de PDFs - Imágenes irrelevantes se filtran antes de describir
- Imágenes relevantes se describen con Vision API (con cache)
-
MultimodalIndexeralmacena chunks en ChromaDB con metadata -
HybridRetrieverbusca por texto, imagen, o ambos - Retrieval inteligente detecta queries visuales
-
AnswerGeneratorproduce respuestas con fuentes citadas -
MultimodalRAGPipelineorquesta indexación y consulta - Pipeline soporta múltiples documentos
- Métricas de costo, tiempo y cache disponibles
- Extensión de reranking funcional
- Extensión de Image Q&A funcional
Ejercicios
Ejercicio 1: Pipeline con reporte de calidad
Implementa una función que, dado un conjunto de preguntas de prueba con respuestas esperadas, ejecute el pipeline y genere un reporte midiendo si las respuestas contienen los puntos clave esperados.
Ver solución
def evaluate_pipeline(pipeline: MultimodalRAGPipeline, test_cases: list[dict]) -> dict:
results = []
for case in test_cases:
question = case["question"]
expected = case.get("expected_keywords", [])
response = pipeline.query(question)
answer_lower = response["answer"].lower()
found = [kw for kw in expected if kw.lower() in answer_lower]
missed = [kw for kw in expected if kw.lower() not in answer_lower]
coverage = len(found) / max(len(expected), 1)
results.append({
"question": question,
"coverage": round(coverage, 2),
"found": found, "missed": missed,
"sources": len(response["sources"]),
"time": response["time_seconds"],
})
avg_coverage = sum(r["coverage"] for r in results) / max(len(results), 1)
return {
"cases": len(test_cases),
"avg_coverage": round(avg_coverage, 2),
"details": results,
}
test_cases = [
{"question": "¿Qué protocolo usa el API Gateway?", "expected_keywords": ["http", "rest", "gateway"]},
{"question": "¿Qué base de datos usa el servicio de pagos?", "expected_keywords": ["postgresql", "pagos"]},
]
# report = evaluate_pipeline(pipeline, test_cases)
# print(f"Cobertura promedio: {report['avg_coverage']}")Ejercicio 2: Multi-document RAG con comparación
Implementa una función que indexe múltiples documentos y responda preguntas comparativas entre ellos, citando de qué documento proviene cada dato.
Ver solución
def comparative_query(
pipeline: MultimodalRAGPipeline,
question: str,
doc_names: list[str],
k_per_doc: int = 3,
) -> dict:
all_chunks = []
for doc_name in doc_names:
results = pipeline.retriever.retrieve(question, k=k_per_doc, content_type=None)
matching = [r for r in results if doc_name in r.get("metadata", {}).get("source", "")]
for r in matching:
r["from_doc"] = doc_name
all_chunks.extend(matching)
all_chunks.sort(key=lambda x: x["similarity"], reverse=True)
top = all_chunks[:k_per_doc * 2]
context = "\n\n---\n\n".join(
f"[{c.get('from_doc', '?')} | p.{c['metadata']['page']}] {c['text'][:300]}"
for c in top
)
response = OpenAI().chat.completions.create(
model=pipeline.generator.model,
messages=[
{
"role": "system",
"content": (
"Responde comparando información de DIFERENTES documentos. "
"Para cada punto, indica de qué documento proviene."
),
},
{"role": "user", "content": f"Contexto:\n{context}\n\nPregunta: {question}"},
],
max_tokens=800, temperature=0,
)
return {
"question": question,
"answer": response.choices[0].message.content,
"documents_compared": list(set(c.get("from_doc", "?") for c in top)),
}
# result = comparative_query(pipeline, "¿Cómo maneja cada sistema la autenticación?", ["manual_v1", "manual_v2"])
# print(result["answer"])Resumen
- El proyecto tiene cuatro componentes: procesador de documentos, indexador, retriever híbrido, y generador de respuestas.
- El procesador extrae texto e imágenes de PDFs, filtra imágenes irrelevantes, y describe las relevantes con Vision API usando cache persistente.
- El indexador almacena chunks con embeddings en ChromaDB, soportando upsert y procesamiento por batches.
- El retriever híbrido combina búsqueda de texto e imagen, con detección automática de queries visuales.
- El generador construye prompts con contexto formateado y retorna respuestas con fuentes.
- El pipeline orquesta todo: indexación de documentos, consultas con retrieval inteligente, y sesiones interactivas.
- Reranking mejora la precisión reordenando candidatos con un segundo modelo.
- Image Q&A permite analizar imágenes originales directamente cuando están disponibles.
Próximo módulo: Módulo 7 — Casos de Uso. Ya tienes vision, documentos, generación, audio y RAG; ahora aplicarás todo en patrones de diseño reales: Document Q&A, image analysis automatizado, análisis de video, y un Use Case Selector que enruta inputs al pipeline correcto.
Recursos Adicionales
- ChromaDB Documentation — API y configuración
- OpenAI Vision API — Uso de imágenes con GPT-4o
- PyMuPDF (fitz) — Extracción de PDFs
- RAG Best Practices — Patrones de RAG en producción
- Embedding Models Comparison — Benchmark de modelos