Módulo 8: Document Analyzer Multimodal

5. RAG y Q&A

Siguiente →

Descripción

El RAGModule es el componente que le da memoria al Document Analyzer. Mientras los otros módulos procesan un documento una vez y generan resultados inmediatos, RAG indexa el contenido para que puedas hacer preguntas después — incluso sobre múltiples documentos. Un usuario sube una factura hoy, un contrato mañana, y puede preguntar "¿cuánto pagamos a Tech Solutions en total?" y obtener respuesta con fuentes.

Por qué importa: Sin RAG, cada pregunta requiere reprocesar el documento completo. Con RAG, el documento se procesa una vez, se indexa en una base de datos vectorial, y las preguntas se responden en milisegundos consultando los chunks más relevantes. Esto es especialmente valioso cuando tienes decenas o cientos de documentos indexados.

Conexión con el módulo: En el Módulo 6 construiste un sistema RAG básico con ChromaDB y embeddings. Aquí lo integras en el Document Analyzer con mejoras: indexación de texto Y descripciones de imágenes (RAG multimodal), metadata por página y documento, retrieval híbrido con filtros, y generación de respuestas con fuentes citadas.

Arquitectura del RAGModule

Responsabilidades

RAGModule
├── index()          → Indexar documento procesado en ChromaDB
├── query()          → Buscar chunks relevantes y generar respuesta
├── delete()         → Eliminar documento del índice
├── list_documents() → Listar documentos indexados
└── get_stats()      → Estadísticas del índice

Flujo de indexación

ProcessedDocument
    │
    ├── Páginas de texto → TextChunker → Chunks con metadata
    │                                       │
    │                                       ▼
    │                              ChromaDB.add(documents, metadatas, ids)
    │
    └── Páginas de imagen → VisionAnalyzer.describe_for_rag()
                               │
                               ▼
                          Descripciones textuales
                               │
                               ▼
                     ChromaDB.add(descriptions, metadatas, ids)

Flujo de Q&A

Pregunta del usuario
    │
    ▼
ChromaDB.query(question, n_results=5)
    │
    ▼
Chunks relevantes + metadata (página, tipo, documento)
    │
    ▼
LLM: "Dado este contexto, responde la pregunta"
    │
    ▼
Respuesta + fuentes citadas

Modelo de Datos para RAG

Chunk indexado

from pydantic import BaseModel
from typing import Optional


class IndexedChunk(BaseModel):
    chunk_id: str
    doc_id: str
    text: str
    page_number: int
    content_type: str  # "text" | "image_description"
    metadata: dict = {}


class QAResult(BaseModel):
    question: str
    answer: str
    sources: list[dict] = []
    chunks_used: int = 0
    confidence: Optional[float] = None

Estructura de metadata en ChromaDB

Cada chunk se almacena con metadata que permite filtrar por documento, página y tipo:

{
    "doc_id": "a1b2c3d4",
    "page_number": 3,
    "content_type": "text",          # "text" | "image_description"
    "filename": "factura_marzo.pdf",
    "document_type": "factura",       # clasificación del VisionAnalyzer
    "chunk_index": 2,
    "char_count": 1423
}

Implementación: RAGModule

Clase completa

import logging
import os
from typing import Optional

import chromadb
from chromadb.utils import embedding_functions
from openai import OpenAI

logger = logging.getLogger(__name__)

DEFAULT_COLLECTION = "document_analyzer"
DEFAULT_N_RESULTS = 5
CHUNK_SIZE = 1500
CHUNK_OVERLAP = 200
MIN_CHUNK_SIZE = 100


class RAGModule:
    def __init__(
        self,
        collection_name: str = DEFAULT_COLLECTION,
        persist_directory: Optional[str] = None
    ):
        self.openai_client = OpenAI()

        self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
            api_key=os.getenv("OPENAI_API_KEY"),
            model_name="text-embedding-3-small"
        )

        if persist_directory:
            self.chroma_client = chromadb.PersistentClient(path=persist_directory)
        else:
            self.chroma_client = chromadb.Client()

        self.collection = self.chroma_client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedding_fn,
            metadata={"hnsw:space": "cosine"}
        )

        logger.info(
            f"RAGModule inicializado. Colección: {collection_name}, "
            f"documentos existentes: {self.collection.count()}"
        )

    def index(
        self,
        doc_id: str,
        content,
        filename: str = "",
        document_type: str = "otro",
        image_descriptions: Optional[list[str]] = None
    ) -> dict:
        chunks_indexed = 0

        if content.has_text_pages and content.full_text:
            text_chunks = self._chunk_text(content.full_text, doc_id)
            if text_chunks:
                self.collection.add(
                    documents=[c["text"] for c in text_chunks],
                    ids=[c["id"] for c in text_chunks],
                    metadatas=[{
                        "doc_id": doc_id,
                        "page_number": 0,
                        "content_type": "text",
                        "filename": filename,
                        "document_type": document_type,
                        "chunk_index": c["index"],
                        "char_count": len(c["text"])
                    } for c in text_chunks]
                )
                chunks_indexed += len(text_chunks)
                logger.info(f"Indexados {len(text_chunks)} chunks de texto para {doc_id}")

        if image_descriptions:
            desc_ids = [f"{doc_id}_imgdesc_{i}" for i in range(len(image_descriptions))]
            self.collection.add(
                documents=image_descriptions,
                ids=desc_ids,
                metadatas=[{
                    "doc_id": doc_id,
                    "page_number": i + 1,
                    "content_type": "image_description",
                    "filename": filename,
                    "document_type": document_type,
                    "chunk_index": i,
                    "char_count": len(desc)
                } for i, desc in enumerate(image_descriptions)]
            )
            chunks_indexed += len(image_descriptions)
            logger.info(f"Indexadas {len(image_descriptions)} descripciones de imagen para {doc_id}")

        return {
            "doc_id": doc_id,
            "chunks_indexed": chunks_indexed,
            "total_in_collection": self.collection.count()
        }

    def query(
        self,
        question: str,
        doc_id: Optional[str] = None,
        n_results: int = DEFAULT_N_RESULTS,
        content_type: Optional[str] = None
    ) -> QAResult:
        where_filter = self._build_filter(doc_id, content_type)

        query_kwargs = {
            "query_texts": [question],
            "n_results": min(n_results, self.collection.count() or 1),
        }
        if where_filter:
            query_kwargs["where"] = where_filter

        try:
            results = self.collection.query(**query_kwargs)
        except Exception as e:
            logger.error(f"Error en query ChromaDB: {e}")
            return QAResult(
                question=question,
                answer="Error al buscar en el índice.",
                sources=[], chunks_used=0
            )

        if not results["documents"] or not results["documents"][0]:
            return QAResult(
                question=question,
                answer="No se encontraron documentos relevantes para responder la pregunta.",
                sources=[], chunks_used=0
            )

        documents = results["documents"][0]
        metadatas = results["metadatas"][0] if results.get("metadatas") else [{}] * len(documents)
        distances = results["distances"][0] if results.get("distances") else [0] * len(documents)

        context = self._build_context(documents, metadatas)
        answer = self._generate_answer(question, context)

        sources = []
        for i, (doc_text, meta, dist) in enumerate(zip(documents, metadatas, distances)):
            sources.append({
                "text": doc_text[:200] + "..." if len(doc_text) > 200 else doc_text,
                "page": meta.get("page_number", "?"),
                "type": meta.get("content_type", "?"),
                "filename": meta.get("filename", "?"),
                "relevance": round(1 - dist, 3) if dist else None
            })

        avg_relevance = sum(s["relevance"] for s in sources if s["relevance"]) / len(sources) if sources else 0

        return QAResult(
            question=question,
            answer=answer,
            sources=sources,
            chunks_used=len(documents),
            confidence=round(avg_relevance, 2)
        )

    def delete(self, doc_id: str) -> dict:
        all_items = self.collection.get(where={"doc_id": doc_id})
        if all_items["ids"]:
            self.collection.delete(ids=all_items["ids"])
            logger.info(f"Eliminados {len(all_items['ids'])} chunks de {doc_id}")
            return {"deleted": len(all_items["ids"]), "doc_id": doc_id}
        return {"deleted": 0, "doc_id": doc_id}

    def list_documents(self) -> list[dict]:
        all_items = self.collection.get()
        doc_ids = set()
        docs = {}

        for meta in all_items.get("metadatas", []):
            did = meta.get("doc_id", "unknown")
            if did not in docs:
                docs[did] = {
                    "doc_id": did,
                    "filename": meta.get("filename", "?"),
                    "document_type": meta.get("document_type", "?"),
                    "chunks": 0
                }
            docs[did]["chunks"] += 1

        return list(docs.values())

    def get_stats(self) -> dict:
        total = self.collection.count()
        docs = self.list_documents()
        return {
            "total_chunks": total,
            "total_documents": len(docs),
            "documents": docs
        }

    def _chunk_text(self, text: str, doc_id: str) -> list[dict]:
        chunks = []
        paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
        current = ""
        idx = 0

        for para in paragraphs:
            if len(current) + len(para) + 2 <= CHUNK_SIZE:
                current += ("\n\n" + para if current else para)
            else:
                if len(current) >= MIN_CHUNK_SIZE:
                    chunks.append({
                        "id": f"{doc_id}_chunk_{idx}",
                        "text": current,
                        "index": idx
                    })
                    idx += 1
                overlap = current[-CHUNK_OVERLAP:] if current else ""
                current = overlap + "\n\n" + para if overlap else para

        if len(current) >= MIN_CHUNK_SIZE:
            chunks.append({
                "id": f"{doc_id}_chunk_{idx}",
                "text": current,
                "index": idx
            })

        return chunks

    def _build_filter(
        self, doc_id: Optional[str], content_type: Optional[str]
    ) -> Optional[dict]:
        conditions = []
        if doc_id:
            conditions.append({"doc_id": {"$eq": doc_id}})
        if content_type:
            conditions.append({"content_type": {"$eq": content_type}})

        if len(conditions) == 0:
            return None
        if len(conditions) == 1:
            return conditions[0]
        return {"$and": conditions}

    def _build_context(self, documents: list[str], metadatas: list[dict]) -> str:
        parts = []
        for i, (doc, meta) in enumerate(zip(documents, metadatas)):
            source_info = f"[Fuente {i+1}: {meta.get('filename', '?')}, página {meta.get('page_number', '?')}]"
            parts.append(f"{source_info}\n{doc}")
        return "\n\n---\n\n".join(parts)

    def _generate_answer(self, question: str, context: str) -> str:
        r = self.openai_client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "Eres un asistente que responde preguntas sobre documentos. "
                        "Responde SOLO con base en el contexto proporcionado. "
                        "Si la información no está en el contexto, di que no tienes información suficiente. "
                        "Cita las fuentes cuando sea relevante (e.g., '[Fuente 1]')."
                    )
                },
                {
                    "role": "user",
                    "content": f"Contexto:\n{context}\n\nPregunta: {question}"
                }
            ],
            max_tokens=500,
            temperature=0
        )
        return r.choices[0].message.content

Indexación Multimodal

El concepto

RAG multimodal indexa tanto texto directo como descripciones generadas de imágenes. Cuando un usuario pregunta sobre una tabla que estaba en una página escaneada, el sistema encuentra la descripción de esa imagen y la usa como contexto.

Documento de 10 páginas
├── Páginas 1-7 (texto) → 12 chunks de texto indexados
└── Páginas 8-10 (escaneadas) → 3 descripciones de imagen indexadas

Total en ChromaDB: 15 chunks

Integración con VisionAnalyzer

def index_full_document(
    rag: RAGModule,
    analyzer: VisionAnalyzer,
    doc_id: str,
    content,
    filename: str,
    document_type: str
) -> dict:
    image_descriptions = None

    if content.has_image_pages:
        images = content.get_images_for_vision()
        image_descriptions = analyzer.describe_for_rag(images)

    return rag.index(
        doc_id=doc_id,
        content=content,
        filename=filename,
        document_type=document_type,
        image_descriptions=image_descriptions
    )

Retrieval con Filtros

Casos de uso de filtros

EscenarioFiltro
Pregunta sobre un documento específicodoc_id="factura_001"
Pregunta sobre todas las facturasdocument_type="factura"
Solo buscar en texto (no imágenes)content_type="text"
Solo buscar en descripciones de imágenescontent_type="image_description"
Sin filtro (buscar en todo)None

Ejemplo de queries filtradas

rag = RAGModule(persist_directory="./chroma_data")

result = rag.query(
    question="¿Cuál es el total de la factura de marzo?",
    doc_id="factura_marzo_001"
)
print(f"Respuesta: {result.answer}")
print(f"Fuentes: {len(result.sources)}")

result_all = rag.query(
    question="¿Cuánto hemos pagado en total este trimestre?",
    doc_id=None  # busca en todos los documentos
)
print(f"Respuesta: {result_all.answer}")

result_images = rag.query(
    question="¿Qué muestran los gráficos del informe?",
    content_type="image_description"
)
print(f"Respuesta: {result_images.answer}")

Persistencia del Índice

Por qué persistir

Sin persistencia, el índice se pierde al reiniciar el servicio. Cada documento se re-indexaría en cada request.

# Sin persistencia (solo para desarrollo)
rag = RAGModule()

# Con persistencia (producción)
rag = RAGModule(persist_directory="./chroma_data")

Gestión del índice persistente

stats = rag.get_stats()
print(f"Total chunks: {stats['total_chunks']}")
print(f"Total documentos: {stats['total_documents']}")

for doc in stats["documents"]:
    print(f"  {doc['doc_id']}: {doc['filename']} ({doc['chunks']} chunks)")

rag.delete("documento_viejo_001")

Troubleshooting

"Las respuestas no son relevantes"

Causa probable: Chunks demasiado grandes diluyen la relevancia. O embeddings de baja calidad.

Solución: Reducir CHUNK_SIZE:

CHUNK_SIZE = 800      # más pequeño = más preciso
CHUNK_OVERLAP = 100   # ajustar overlap

O aumentar n_results para dar más contexto al LLM:

result = rag.query(question="...", n_results=10)

"ChromaDB lanza error de dimensión"

Causa probable: Mezclaste embeddings de diferentes modelos en la misma colección.

Solución: Crear una nueva colección o borrar la existente:

self.chroma_client.delete_collection("document_analyzer")
self.collection = self.chroma_client.create_collection(
    name="document_analyzer",
    embedding_function=self.embedding_fn
)

"La indexación es lenta"

Causa probable: Muchos chunks pequeños generan muchas llamadas a embeddings.

Solución: ChromaDB envía embeddings en batch automáticamente, pero puedes indexar chunks en lotes:

BATCH_SIZE = 100

for i in range(0, len(all_chunks), BATCH_SIZE):
    batch = all_chunks[i:i + BATCH_SIZE]
    self.collection.add(
        documents=[c["text"] for c in batch],
        ids=[c["id"] for c in batch],
        metadatas=[c["meta"] for c in batch]
    )

"No encuentra documentos recién indexados"

Causa probable: Usando ChromaDB in-memory y se reinició el proceso.

Solución: Usar PersistentClient:

self.chroma_client = chromadb.PersistentClient(path="./chroma_data")

Uso del RAGModule

Ejemplo completo end-to-end

processor = DocumentProcessor()
analyzer = VisionAnalyzer()
rag = RAGModule(persist_directory="./chroma_data")

doc = processor.process("contrato_servicios.pdf")
doc_type = analyzer.classify(doc)

image_descriptions = None
if doc.has_image_pages:
    image_descriptions = analyzer.describe_for_rag(doc.get_images_for_vision())

index_result = rag.index(
    doc_id="contrato_001",
    content=doc,
    filename="contrato_servicios.pdf",
    document_type=doc_type,
    image_descriptions=image_descriptions
)
print(f"Indexados: {index_result['chunks_indexed']} chunks")

qa = rag.query(
    question="¿Cuáles son las partes del contrato?",
    doc_id="contrato_001"
)
print(f"\nPregunta: {qa.question}")
print(f"Respuesta: {qa.answer}")
print(f"Confianza: {qa.confidence}")
print(f"Fuentes usadas: {qa.chunks_used}")
for source in qa.sources:
    print(f"  - [{source['type']}] Página {source['page']}: {source['text'][:80]}...")

Output esperado

Indexados: 8 chunks

Pregunta: ¿Cuáles son las partes del contrato?
Respuesta: Las partes del contrato son: (1) Tech Solutions S.A. como prestador
de servicios, y (2) Empresa ABC S.A. de C.V. como contratante [Fuente 1].
Confianza: 0.87
Fuentes usadas: 5
  - [text] Página ?: CONTRATO DE PRESTACIÓN DE SERVICIOS que celebran...
  - [text] Página ?: Las partes acuerdan los siguientes términos...

Ejercicios

Ejercicio 1: Respuesta con fuentes formateadas

Modifica el query() para que la respuesta incluya las fuentes citadas en formato legible. El LLM debe citar "[Fuente N]" en su respuesta, y el resultado debe incluir un mapeo de fuentes con su texto y metadata.

Ver solución 
def query_with_formatted_sources(
    self,
    question: str,
    doc_id: Optional[str] = None,
    n_results: int = 5
) -> dict:
    result = self.query(question, doc_id, n_results)

    source_map = {}
    for i, source in enumerate(result.sources):
        key = f"Fuente {i + 1}"
        source_map[key] = {
            "texto": source["text"],
            "archivo": source["filename"],
            "pagina": source["page"],
            "tipo": source["type"],
            "relevancia": source.get("relevance")
        }

    return {
        "pregunta": result.question,
        "respuesta": result.answer,
        "confianza": result.confidence,
        "fuentes_citadas": source_map,
        "total_fuentes": len(source_map)
    }


rag = RAGModule()
formatted = rag.query_with_formatted_sources(
    "¿Cuál es la fecha de firma?", doc_id="contrato_001"
)
print(f"Respuesta: {formatted['respuesta']}")
print(f"\nFuentes citadas:")
for key, src in formatted["fuentes_citadas"].items():
    print(f"  [{key}] {src['archivo']}, pág {src['pagina']}: {src['texto'][:60]}...")

Ejercicio 2: Búsqueda cross-document

Implementa una función que busque en todos los documentos indexados y agrupe las fuentes por documento. Útil para preguntas como "¿cuánto pagamos en total este trimestre?" que requieren información de múltiples facturas.

Ver solución 
def cross_document_query(
    self,
    question: str,
    n_results: int = 10
) -> dict:
    result = self.query(question, doc_id=None, n_results=n_results)

    by_document: dict[str, list] = {}
    for source in result.sources:
        doc_key = source.get("filename", "unknown")
        if doc_key not in by_document:
            by_document[doc_key] = []
        by_document[doc_key].append(source)

    return {
        "pregunta": result.question,
        "respuesta": result.answer,
        "confianza": result.confidence,
        "documentos_consultados": len(by_document),
        "fuentes_por_documento": {
            doc: {
                "cantidad_fuentes": len(sources),
                "fuentes": [s["text"][:100] for s in sources]
            }
            for doc, sources in by_document.items()
        }
    }


rag = RAGModule(persist_directory="./chroma_data")
cross = rag.cross_document_query("¿Cuánto pagamos en total este trimestre?")
print(f"Respuesta: {cross['respuesta']}")
print(f"Documentos consultados: {cross['documentos_consultados']}")
for doc_name, info in cross["fuentes_por_documento"].items():
    print(f"  {doc_name}: {info['cantidad_fuentes']} fuentes")

Resumen

  • El RAGModule indexa documentos para Q&A: texto como chunks, imágenes como descripciones textuales.
  • Usa ChromaDB con embeddings de OpenAI (text-embedding-3-small).
  • Retrieval con filtros: por documento, por tipo de contenido, o sin filtro (cross-document).
  • Generación de respuesta: LLM con contexto recuperado, instrucciones de citar fuentes.
  • Persistencia: PersistentClient para no perder el índice entre reinicios.
  • Gestión: listar documentos, eliminar, estadísticas del índice.
  • La indexación multimodal permite Q&A sobre contenido visual a través de descripciones.

Recursos Adicionales

  1. ChromaDB Documentation — Referencia completa
  2. OpenAI Embeddings — Modelos de embeddings
  3. RAG Best Practices — Patrones RAG con LangChain
  4. Módulo 6 de esta guía — Base de RAG multimodal
← Anterior4. Análisis con VisionSiguiente →6. Integración de Audio