Módulo 7: Casos de Uso

2. Document Q&A

Descripción

Document Q&A es el patrón más demandado en IA multimodal: un usuario sube un documento (PDF, imagen de documento, Word) y hace preguntas sobre su contenido. El sistema extrae texto e imágenes, indexa el contenido, recupera los fragmentos relevantes, y genera una respuesta citando las fuentes.

Por qué importa: Este patrón es la base de asistentes legales, sistemas de soporte técnico, plataformas educativas, y herramientas de compliance. Es el caso de uso que más justifica inversiones en IA multimodal porque resuelve un problema real: "tengo 200 páginas de documentación y necesito una respuesta en 5 segundos".

Conexión con el módulo: Document Q&A combina Módulo 3 (extracción de documentos), Módulo 6 (RAG con embeddings), y Módulo 1-2 (generación con LLM). La cápsula 08 (Use Case Selector) usará este pipeline como uno de sus destinos de routing.

Pipeline Visual

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  Documento   │────▶│   Extraer    │────▶│   Chunking   │
│  (PDF/img)   │     │ texto+imgs   │     │              │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  │
                                                  ▼
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  Respuesta   │◀────│     LLM      │◀────│  Retrieval   │
│  + fuentes   │     │  (generar)   │     │ (buscar)     │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  ▲
                                                  │
                                           ┌──────────────┐
                                           │   Pregunta   │
                                           │  del usuario │
                                           └──────────────┘

Etapas del pipeline:

Cargar documento — Abrir PDF, imagen escaneada, o texto
Extraer contenido — Texto con PyMuPDF, imágenes embebidas, tablas
Chunking — Dividir en fragmentos de tamaño manejable
Indexar — Crear embeddings y almacenar en vector store
Query — Recibir pregunta del usuario
Retrieval — Buscar chunks más relevantes por similitud
Generar — LLM produce respuesta basada en chunks recuperados
Citar fuentes — Incluir referencias a los chunks usados

Paso 1: Cargar y Extraer Contenido

Extracción de texto

import fitz

def extract_text_from_pdf(pdf_path: str) -> list[dict]:
    doc = fitz.open(pdf_path)
    pages = []
    for page_num in range(len(doc)):
        page = doc[page_num]
        text = page.get_text()
        if text.strip():
            pages.append({
                "page": page_num + 1,
                "text": text.strip()
            })
    doc.close()
    return pages

Extracción de imágenes

import base64
from pathlib import Path

def extract_images_from_pdf(pdf_path: str, output_dir: str = "/tmp/doc_images") -> list[dict]:
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    doc = fitz.open(pdf_path)
    images = []

    for page_num in range(len(doc)):
        page = doc[page_num]
        image_list = page.get_images(full=True)

        for img_idx, img_info in enumerate(image_list):
            xref = img_info[0]
            base_image = doc.extract_image(xref)
            image_bytes = base_image["image"]
            ext = base_image["ext"]

            img_path = f"{output_dir}/page{page_num + 1}_img{img_idx + 1}.{ext}"
            with open(img_path, "wb") as f:
                f.write(image_bytes)

            images.append({
                "page": page_num + 1,
                "path": img_path,
                "size_bytes": len(image_bytes)
            })

    doc.close()
    return images

Extracción combinada

def extract_document(pdf_path: str) -> dict:
    pages = extract_text_from_pdf(pdf_path)
    images = extract_images_from_pdf(pdf_path)
    return {
        "pages": pages,
        "images": images,
        "total_pages": len(pages),
        "total_images": len(images)
    }

Paso 2: Chunking Inteligente

El chunking es crítico. Chunks muy grandes diluyen la relevancia; chunks muy pequeños pierden contexto.

Chunking por tamaño con overlap

def chunk_text(text: str, chunk_size: int = 1500, overlap: int = 200) -> list[str]:
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if chunk.strip():
            chunks.append(chunk.strip())
        start = end - overlap
    return chunks

Chunking por páginas con metadatos

def chunk_pages(pages: list[dict], chunk_size: int = 1500, overlap: int = 200) -> list[dict]:
    chunks = []
    for page in pages:
        page_chunks = chunk_text(page["text"], chunk_size, overlap)
        for i, chunk in enumerate(page_chunks):
            chunks.append({
                "text": chunk,
                "page": page["page"],
                "chunk_index": i,
                "source": f"Página {page['page']}, fragmento {i + 1}"
            })
    return chunks

Chunking semántico (por párrafos)

def chunk_by_paragraphs(text: str, max_chunk_size: int = 1500) -> list[str]:
    paragraphs = text.split("\n\n")
    chunks = []
    current_chunk = ""

    for para in paragraphs:
        if len(current_chunk) + len(para) + 2 <= max_chunk_size:
            current_chunk += para + "\n\n"
        else:
            if current_chunk.strip():
                chunks.append(current_chunk.strip())
            current_chunk = para + "\n\n"

    if current_chunk.strip():
        chunks.append(current_chunk.strip())

    return chunks

Paso 3: Indexar con Embeddings

Con ChromaDB

import chromadb
from openai import OpenAI

client = OpenAI()

def create_document_index(chunks: list[dict], collection_name: str = "document_qa") -> chromadb.Collection:
    chroma_client = chromadb.Client()

    try:
        chroma_client.delete_collection(collection_name)
    except Exception:
        pass

    collection = chroma_client.create_collection(
        name=collection_name,
        metadata={"hnsw:space": "cosine"}
    )

    texts = [c["text"] for c in chunks]
    embeddings = get_embeddings_batch(texts)

    collection.add(
        documents=texts,
        embeddings=embeddings,
        metadatas=[{"page": c["page"], "source": c["source"]} for c in chunks],
        ids=[f"chunk_{i}" for i in range(len(chunks))]
    )

    return collection


def get_embeddings_batch(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    response = client.embeddings.create(
        model=model,
        input=texts
    )
    return [item.embedding for item in response.data]

Indexar descripciones de imágenes (RAG multimodal)

def index_document_images(images: list[dict], collection: chromadb.Collection, start_id: int = 10000) -> None:
    for i, img in enumerate(images):
        description = describe_image_for_indexing(img["path"])

        embedding = get_embeddings_batch([description])[0]

        collection.add(
            documents=[description],
            embeddings=[embedding],
            metadatas=[{
                "page": img["page"],
                "source": f"Imagen en página {img['page']}",
                "type": "image"
            }],
            ids=[f"img_{start_id + i}"]
        )


def describe_image_for_indexing(image_path: str) -> str:
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe esta imagen en detalle para indexación. Incluye todos los datos visibles, textos, y estructura."},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
            ]
        }],
        max_tokens=300
    )
    return response.choices[0].message.content

Paso 4: Retrieval

def retrieve_relevant_chunks(
    collection: chromadb.Collection,
    question: str,
    n_results: int = 5
) -> list[dict]:
    query_embedding = get_embeddings_batch([question])[0]

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=n_results,
        include=["documents", "metadatas", "distances"]
    )

    chunks = []
    for i in range(len(results["documents"][0])):
        chunks.append({
            "text": results["documents"][0][i],
            "metadata": results["metadatas"][0][i],
            "distance": results["distances"][0][i]
        })

    return chunks

Paso 5: Generar Respuesta con Fuentes

def generate_answer(question: str, chunks: list[dict]) -> dict:
    context_parts = []
    for i, chunk in enumerate(chunks):
        source = chunk["metadata"].get("source", f"Fragmento {i + 1}")
        context_parts.append(f"[{source}]\n{chunk['text']}")

    context = "\n\n---\n\n".join(context_parts)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Eres un asistente de Q&A documental. Responde SOLO con base en el contexto proporcionado. Si no encuentras la respuesta, dilo explícitamente. Cita las fuentes entre corchetes [Página X, fragmento Y]."
            },
            {
                "role": "user",
                "content": f"Contexto:\n\n{context}\n\n---\n\nPregunta: {question}"
            }
        ],
        max_tokens=500,
        temperature=0.1
    )

    return {
        "answer": response.choices[0].message.content,
        "sources": [c["metadata"] for c in chunks],
        "model": "gpt-4o-mini",
        "chunks_used": len(chunks)
    }

Pipeline Completo

def document_qa_pipeline(pdf_path: str, question: str) -> dict:
    doc_content = extract_document(pdf_path)

    chunks = chunk_pages(doc_content["pages"])

    collection = create_document_index(chunks)

    if doc_content["images"]:
        index_document_images(doc_content["images"], collection)

    relevant = retrieve_relevant_chunks(collection, question)

    answer = generate_answer(question, relevant)
    answer["document"] = pdf_path
    answer["total_pages"] = doc_content["total_pages"]

    return answer

Uso:

result = document_qa_pipeline(
    "contrato_servicio.pdf",
    "¿Cuál es la cláusula de penalización por incumplimiento?"
)
print(result["answer"])
print(f"Fuentes: {result['sources']}")

Multi-Turn Conversation

Un sistema de Q&A útil mantiene contexto entre preguntas. El usuario pregunta "¿Cuál es la cláusula de penalización?" y después "¿Y cuánto es el monto?", y el sistema entiende que "el monto" se refiere a la penalización.

class DocumentQASession:
    def __init__(self, pdf_path: str):
        self.pdf_path = pdf_path
        self.history: list[dict] = []
        self.collection = None
        self._setup()

    def _setup(self):
        doc_content = extract_document(self.pdf_path)
        chunks = chunk_pages(doc_content["pages"])
        self.collection = create_document_index(chunks)
        if doc_content["images"]:
            index_document_images(doc_content["images"], self.collection)

    def ask(self, question: str) -> dict:
        contextualized_question = self._contextualize(question)

        relevant = retrieve_relevant_chunks(self.collection, contextualized_question)

        context_parts = []
        for i, chunk in enumerate(relevant):
            source = chunk["metadata"].get("source", f"Fragmento {i + 1}")
            context_parts.append(f"[{source}]\n{chunk['text']}")
        context = "\n\n---\n\n".join(context_parts)

        messages = [
            {
                "role": "system",
                "content": "Eres un asistente de Q&A documental. Responde SOLO con base en el contexto proporcionado. Cita fuentes entre corchetes. Mantén coherencia con la conversación previa."
            }
        ]

        for h in self.history[-6:]:
            messages.append({"role": "user", "content": h["question"]})
            messages.append({"role": "assistant", "content": h["answer"]})

        messages.append({
            "role": "user",
            "content": f"Contexto del documento:\n\n{context}\n\n---\n\nPregunta: {question}"
        })

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            max_tokens=500,
            temperature=0.1
        )

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

        self.history.append({
            "question": question,
            "answer": answer,
            "sources": [c["metadata"] for c in relevant]
        })

        return {
            "answer": answer,
            "sources": [c["metadata"] for c in relevant],
            "turn": len(self.history)
        }

    def _contextualize(self, question: str) -> str:
        if not self.history:
            return question

        recent = self.history[-3:]
        history_text = "\n".join(
            f"P: {h['question']}\nR: {h['answer'][:200]}"
            for h in recent
        )

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"Historial:\n{history_text}\n\nNueva pregunta: {question}\n\nReescribe la pregunta para que sea autocontenida (sin pronombres ambiguos). Si ya es clara, repítela tal cual."
            }],
            max_tokens=100,
            temperature=0
        )
        return response.choices[0].message.content

Uso:

session = DocumentQASession("manual_tecnico.pdf")

r1 = session.ask("¿Qué requisitos de hardware se mencionan?")
print(r1["answer"])

r2 = session.ask("¿Y los de software?")
print(r2["answer"])

r3 = session.ask("¿Son compatibles entre sí?")
print(r3["answer"])

Variaciones del Pipeline

Con reranking

Después del retrieval, reordena por relevancia usando un segundo modelo:

def rerank_chunks(question: str, chunks: list[dict], top_k: int = 3) -> list[dict]:
    prompt_parts = []
    for i, chunk in enumerate(chunks):
        prompt_parts.append(f"[{i}] {chunk['text'][:300]}")

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Pregunta: {question}\n\nFragmentos:\n" + "\n\n".join(prompt_parts) +
                       f"\n\nOrdena los índices del más al menos relevante. Solo los números separados por comas."
        }],
        max_tokens=50,
        temperature=0
    )

    try:
        indices = [int(x.strip()) for x in response.choices[0].message.content.split(",")]
        return [chunks[i] for i in indices[:top_k] if i < len(chunks)]
    except (ValueError, IndexError):
        return chunks[:top_k]

Con respuesta estructurada

def generate_structured_answer(question: str, chunks: list[dict]) -> dict:
    context = "\n\n".join(c["text"] for c in chunks)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Responde en formato JSON con: answer (string), confidence (low/medium/high), key_points (array de strings), sources_used (array de strings)."
            },
            {
                "role": "user",
                "content": f"Contexto:\n{context}\n\nPregunta: {question}"
            }
        ],
        max_tokens=500,
        temperature=0,
        response_format={"type": "json_object"}
    )

    import json
    return json.loads(response.choices[0].message.content)

Troubleshooting

Problema 1: Respuestas inventadas (hallucinations)

Síntoma: El modelo responde con información que no está en el documento.

Causa: Contexto insuficiente o prompt que no restringe al modelo.

Solución:

system_prompt = (
    "Responde EXCLUSIVAMENTE con información del contexto proporcionado. "
    "Si la respuesta no está en el contexto, responde: "
    "'No encontré información sobre esto en el documento.' "
    "NUNCA inventes datos."
)

Problema 2: Retrieval pobre (chunks irrelevantes)

Síntoma: Los chunks recuperados no contienen la respuesta aunque el documento sí la tiene.

Causa: Chunks muy grandes, overlap insuficiente, o embeddings que no capturan la semántica.

Solución:

Reducir chunk_size a 800-1000 caracteres
Aumentar overlap a 200-300
Usar text-embedding-3-large en lugar de small
Implementar reranking (ver variación anterior)

Problema 3: PDFs escaneados sin texto

Síntoma: extract_text_from_pdf retorna páginas vacías.

Causa: El PDF es una imagen escaneada, no tiene texto embebido.

Solución:

def extract_with_ocr_fallback(pdf_path: str) -> list[dict]:
    pages = extract_text_from_pdf(pdf_path)

    empty_pages = [p for p in pages if len(p["text"].strip()) < 50]
    if len(empty_pages) > len(pages) * 0.5:
        return extract_via_vision(pdf_path)

    return pages


def extract_via_vision(pdf_path: str) -> list[dict]:
    doc = fitz.open(pdf_path)
    pages = []
    for page_num in range(len(doc)):
        page = doc[page_num]
        pix = page.get_pixmap(dpi=200)
        img_bytes = pix.tobytes("png")
        b64 = base64.b64encode(img_bytes).decode()

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": "Extrae todo el texto visible en esta página. Mantén la estructura."},
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
                ]
            }],
            max_tokens=2000
        )
        pages.append({
            "page": page_num + 1,
            "text": response.choices[0].message.content
        })
    doc.close()
    return pages

Problema 4: Documentos muy largos (100+ páginas)

Síntoma: El indexado tarda mucho o excede límites de la API de embeddings.

Solución:

def index_large_document(chunks: list[dict], batch_size: int = 100) -> chromadb.Collection:
    chroma_client = chromadb.Client()
    collection = chroma_client.create_collection("large_doc")

    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        texts = [c["text"] for c in batch]
        embeddings = get_embeddings_batch(texts)

        collection.add(
            documents=texts,
            embeddings=embeddings,
            metadatas=[{"page": c["page"], "source": c["source"]} for c in batch],
            ids=[f"chunk_{i + j}" for j in range(len(batch))]
        )

    return collection

Problema 5: Contexto excede token limit del LLM

Síntoma: Error maximum context length exceeded.

Solución:

def trim_context_to_limit(chunks: list[dict], max_chars: int = 12000) -> list[dict]:
    trimmed = []
    total = 0
    for chunk in chunks:
        if total + len(chunk["text"]) > max_chars:
            break
        trimmed.append(chunk)
        total += len(chunk["text"])
    return trimmed

Ejercicios

Ejercicio 1: Document Q&A con score de confianza

Modifica generate_answer para que incluya un score de confianza (low/medium/high) basado en la distancia de los chunks recuperados.

Pista: Si la distancia promedio es < 0.3, confianza alta; < 0.5, media; > 0.5, baja.

Ver solución

def generate_answer_with_confidence(question: str, chunks: list[dict]) -> dict:
    avg_distance = sum(c["distance"] for c in chunks) / len(chunks) if chunks else 1.0

    if avg_distance < 0.3:
        confidence = "high"
    elif avg_distance < 0.5:
        confidence = "medium"
    else:
        confidence = "low"

    context_parts = []
    for i, chunk in enumerate(chunks):
        source = chunk["metadata"].get("source", f"Fragmento {i + 1}")
        context_parts.append(f"[{source}]\n{chunk['text']}")
    context = "\n\n---\n\n".join(context_parts)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Responde SOLO con base en el contexto. Cita fuentes. Si la información es insuficiente, dilo."
            },
            {
                "role": "user",
                "content": f"Contexto:\n\n{context}\n\n---\n\nPregunta: {question}"
            }
        ],
        max_tokens=500,
        temperature=0.1
    )

    return {
        "answer": response.choices[0].message.content,
        "confidence": confidence,
        "avg_distance": round(avg_distance, 3),
        "sources": [c["metadata"] for c in chunks]
    }

Ejercicio 2: Retrieval con filtro por página

Implementa una función que permita al usuario restringir la búsqueda a páginas específicas del documento.

Pista: Usa el parámetro where de ChromaDB para filtrar por metadatos.

Ver solución

def retrieve_from_pages(
    collection: chromadb.Collection,
    question: str,
    pages: list[int] = None,
    n_results: int = 5
) -> list[dict]:
    query_embedding = get_embeddings_batch([question])[0]

    query_params = {
        "query_embeddings": [query_embedding],
        "n_results": n_results,
        "include": ["documents", "metadatas", "distances"]
    }

    if pages:
        query_params["where"] = {"page": {"$in": pages}}

    results = collection.query(**query_params)

    chunks = []
    for i in range(len(results["documents"][0])):
        chunks.append({
            "text": results["documents"][0][i],
            "metadata": results["metadatas"][0][i],
            "distance": results["distances"][0][i]
        })

    return chunks

Ejercicio 3: Comparar dos documentos

Crea un pipeline que reciba dos PDFs y una pregunta, indexe ambos, y genere una respuesta comparativa.

Pista: Usa prefijos en los IDs y metadatos para distinguir de qué documento viene cada chunk.

Ver solución

def compare_documents_qa(pdf_path_a: str, pdf_path_b: str, question: str) -> dict:
    chroma_client = chromadb.Client()
    collection = chroma_client.create_collection("compare_docs")

    for label, pdf_path in [("Documento A", pdf_path_a), ("Documento B", pdf_path_b)]:
        pages = extract_text_from_pdf(pdf_path)
        chunks = chunk_pages(pages)

        texts = [c["text"] for c in chunks]
        embeddings = get_embeddings_batch(texts)

        collection.add(
            documents=texts,
            embeddings=embeddings,
            metadatas=[{
                "page": c["page"],
                "source": f"{label}, {c['source']}",
                "document": label
            } for c in chunks],
            ids=[f"{label.lower().replace(' ', '_')}_{i}" for i in range(len(chunks))]
        )

    relevant = retrieve_relevant_chunks(collection, question, n_results=8)

    context_parts = []
    for chunk in relevant:
        source = chunk["metadata"]["source"]
        context_parts.append(f"[{source}]\n{chunk['text']}")
    context = "\n\n---\n\n".join(context_parts)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Compara la información de Documento A y Documento B para responder. Cita qué documento dice qué."
            },
            {
                "role": "user",
                "content": f"Contexto:\n\n{context}\n\n---\n\nPregunta comparativa: {question}"
            }
        ],
        max_tokens=600,
        temperature=0.1
    )

    return {
        "answer": response.choices[0].message.content,
        "sources": [c["metadata"] for c in relevant]
    }

Recursos Adicionales

RAG Guide — LangChain — Implementación de referencia
ChromaDB Documentation — Vector store usado en los ejemplos
OpenAI Embeddings — Modelos de embeddings
PyMuPDF Documentation — Extracción de PDFs

← Anterior1. Introducción: Casos de Uso Siguiente →3. Image Analysis Automatizado