8. Proyecto: Multi-Stage Analysis Pipeline

Descripción del Proyecto

En este proyecto construirás un pipeline de análisis de documentos completo y listo para producción. El sistema recibe un documento de cualquier longitud, lo procesa en cuatro etapas especializadas (Extract → Analyze → Synthesize → Format), con gestión de contexto inteligente para documentos largos, routing por complejidad, manejo de errores robusto con retry, y salida estructurada.

Este proyecto integra todas las técnicas del Módulo 06:

• Prompt Chaining (etapas secuenciales)
• Task Decomposition (cada etapa tiene responsabilidad única)
• Multi-Stage Pipelines (estado, retry, fallbacks)
• Context Window Management (chunking y summarización)
• Routing por Complejidad (selección de modelo según tamaño)

Especificaciones del Sistema

Capacidades requeridas

Arquitectura

Documento Input
      │
      ▼
┌─────────────────────────────────────────────┐
│  PRE-PROCESAMIENTO                           │
│  - Contar tokens con tiktoken                │
│  - Si > 8K tokens: Summarization Chain       │
│  - Preparar metadata del documento           │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│  ETAPA 1: EXTRACT                            │
│  - Extraer secciones clave                   │
│  - Identificar datos, cifras, fechas         │
│  - Detectar entidades importantes            │
│  Modelo: gpt-4o-mini                         │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│  ETAPA 2: ANALYZE                            │
│  - Análisis de patrones y tendencias         │
│  - Identificar causas y efectos              │
│  - Evaluar riesgos y oportunidades           │
│  Modelo: gpt-4o (mayor capacidad)            │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│  ETAPA 3: SYNTHESIZE                         │
│  - Generar insights accionables              │
│  - Priorizar hallazgos por impacto           │
│  - Crear narrativa coherente                 │
│  Modelo: gpt-4o                              │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│  ETAPA 4: FORMAT                             │
│  - Estructurar en JSON con Pydantic          │
│  - Validar coherencia y completitud          │
│  - Generar resumen ejecutivo                 │
│  Modelo: gpt-4o-mini                         │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
         Output estructurado (JSON)

Implementación Completa

Modelos de datos

from openai import OpenAI
from pydantic import BaseModel, Field
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Any
import time
import json
import tiktoken
import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)

client = OpenAI()


# ============================================================
# MODELOS DE DATOS
# ============================================================

class Hallazgo(BaseModel):
    categoria: str = Field(..., description="Categoria del hallazgo: dato_clave, riesgo, oportunidad, etc.")
    descripcion: str = Field(..., description="Descripción detallada del hallazgo")
    impacto: str = Field(..., description="bajo, medio, alto")
    evidencia: str = Field(..., description="Cita o dato del documento que sustenta el hallazgo")

class Recomendacion(BaseModel):
    accion: str = Field(..., description="Acción concreta a tomar")
    prioridad: str = Field(..., description="urgente, importante, nice-to-have")
    responsable: Optional[str] = Field(None, description="Área o rol responsable")
    plazo: Optional[str] = Field(None, description="Plazo sugerido: inmediato, 30 dias, 90 dias, etc.")

class ResultadoAnalisis(BaseModel):
    titulo: str = Field(..., description="Título descriptivo del documento analizado")
    resumen_ejecutivo: str = Field(..., description="Resumen de 3-5 oraciones para un ejecutivo")
    tipo_documento: str = Field(..., description="informe, contrato, articulo, datos, otro")
    hallazgos: list[Hallazgo] = Field(..., description="Lista de hallazgos principales (máximo 10)")
    recomendaciones: list[Recomendacion] = Field(..., description="Acciones recomendadas (máximo 5)")
    score_confianza: float = Field(..., description="Confianza en el análisis (0-1)")
    metadata: dict = Field(default_factory=dict, description="Métricas del pipeline")

@dataclass
class EstadoPipeline:
    """Estado compartido durante la ejecución del pipeline."""
    documento_original: str
    documento_procesado: str  # Puede ser resumido si es muy largo
    
    # Outputs de cada etapa
    output_extract: str = ""
    output_analyze: str = ""
    output_synthesize: str = ""
    output_final: Optional[ResultadoAnalisis] = None
    
    # Métricas
    tokens_input: int = 0
    tokens_totales_usados: int = 0
    tiempo_inicio: float = field(default_factory=time.time)
    tiempos_por_etapa: dict = field(default_factory=dict)
    modelos_usados: dict = field(default_factory=dict)
    retries_por_etapa: dict = field(default_factory=dict)
    errores: list[str] = field(default_factory=list)
    fue_resumido: bool = False

Utilidades de contexto

# ============================================================
# UTILIDADES DE CONTEXTO
# ============================================================

def contar_tokens(texto: str, modelo: str = "gpt-4o-mini") -> int:
    """Cuenta los tokens de un texto usando tiktoken."""
    try:
        enc = tiktoken.encoding_for_model(modelo)
    except KeyError:
        enc = tiktoken.get_encoding("cl100k_base")
    return len(enc.encode(texto))


def resumir_chunk(chunk: str, n_palabras: int = 200) -> str:
    """Resume un fragmento de texto en n_palabras."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"""Resume el siguiente texto en aproximadamente {n_palabras} palabras.
Mantén los datos más importantes, cifras, fechas y conclusiones clave.
No omitas información crítica.

Texto:
{chunk}"""
        }],
        temperature=0,
        max_tokens=400
    )
    return response.choices[0].message.content


def preprocesar_documento(
    documento: str,
    max_tokens: int = 8000,
    verbose: bool = True
) -> tuple[str, bool, int]:
    """
    Preprocesa el documento para que quepa en el contexto del pipeline.
    
    Returns:
        Tupla (documento_procesado, fue_resumido, tokens_originales)
    """
    tokens_originales = contar_tokens(documento)
    
    if verbose:
        logger.info(f"Documento original: {tokens_originales:,} tokens ({len(documento):,} chars)")
    
    if tokens_originales <= max_tokens:
        return documento, False, tokens_originales
    
    # El documento es demasiado largo: chunking + summarización
    if verbose:
        logger.info(f"Documento excede {max_tokens:,} tokens. Aplicando summarización...")
    
    # Dividir en chunks de ~3000 tokens con overlap
    encoder = tiktoken.encoding_for_model("gpt-4o-mini")
    tokens = encoder.encode(documento)
    
    chunk_size = 3000
    overlap = 200
    chunks_tokens = []
    
    i = 0
    while i < len(tokens):
        chunk_tokens = tokens[i:i + chunk_size]
        chunks_tokens.append(encoder.decode(chunk_tokens))
        i += chunk_size - overlap
    
    if verbose:
        logger.info(f"Dividido en {len(chunks_tokens)} chunks")
    
    # Resumir cada chunk
    resumenes = []
    for idx, chunk in enumerate(chunks_tokens):
        if verbose:
            logger.info(f"  Resumiendo chunk {idx + 1}/{len(chunks_tokens)}...")
        resumen = resumir_chunk(chunk, n_palabras=150)
        resumenes.append(f"[Fragmento {idx + 1}]\n{resumen}")
    
    documento_resumido = "\n\n".join(resumenes)
    tokens_finales = contar_tokens(documento_resumido)
    
    if verbose:
        logger.info(f"Documento resumido: {tokens_finales:,} tokens ({tokens_originales/tokens_finales:.1f}x reducción)")
    
    return documento_resumido, True, tokens_originales

Las cuatro etapas del pipeline

# ============================================================
# ETAPAS DEL PIPELINE
# ============================================================

PROMPT_EXTRACT = """Eres un experto en análisis de documentos. Tu tarea es extraer información estructurada.

Documento a analizar:
{documento}

Extrae y organiza:
1. **Tema principal**: ¿De qué trata este documento?
2. **Tipo de documento**: informe, contrato, artículo, datos, presentación, otro
3. **Datos clave**: cifras, porcentajes, fechas, nombres, lugares importantes
4. **Secciones principales**: Enumera las secciones o temas tratados
5. **Argumentos o conclusiones**: Principales afirmaciones del documento
6. **Entidades mencionadas**: Personas, organizaciones, productos

Sé exhaustivo pero conciso. Prioriza los datos cuantitativos."""


PROMPT_ANALYZE = """Eres un analista senior. Basándote en la siguiente extracción de un documento, realiza un análisis profundo.

Extracción del documento:
{extracted}

Analiza:
1. **Patrones y tendencias**: ¿Qué patrones emergen de los datos y argumentos?
2. **Fortalezas**: ¿Qué aspectos positivos o fortalezas identifica el documento?
3. **Riesgos o problemas**: ¿Qué riesgos, debilidades o problemas se mencionan?
4. **Oportunidades**: ¿Qué oportunidades de mejora o crecimiento hay?
5. **Causas y efectos**: ¿Qué relaciones causa-efecto son evidentes?
6. **Contexto**: ¿Qué contexto externo es relevante para interpretar esto?
7. **Gaps de información**: ¿Qué información importante falta o es ambigua?

Sé crítico y objetivo. Distingue entre hechos y opiniones."""


PROMPT_SYNTHESIZE = """Eres un consultor estratégico. Basándote en el análisis siguiente, sintetiza los hallazgos clave y genera recomendaciones accionables.

Análisis:
{analyzed}

Sintetiza:
1. **Los 5 hallazgos más importantes** (ordenados por impacto): Cada uno con categoría (dato_clave, riesgo, oportunidad, tendencia), descripción clara, nivel de impacto (alto/medio/bajo), y evidencia del documento.
2. **Las 3-5 recomendaciones más importantes**: Cada una con acción específica, prioridad (urgente/importante/nice-to-have), responsable sugerido, y plazo.
3. **Resumen ejecutivo**: 3-5 oraciones que un ejecutivo puede leer en 30 segundos.
4. **Score de confianza** (0.0-1.0): ¿Qué tan confiable es este análisis dado el documento original?

Sé específico y accionable. Evita generalidades."""


PROMPT_FORMAT = """Convierte la siguiente síntesis en un JSON estructurado.

Síntesis:
{synthesized}

JSON requerido:
{{
  "titulo": "título descriptivo del documento",
  "resumen_ejecutivo": "3-5 oraciones para un ejecutivo",
  "tipo_documento": "informe|contrato|articulo|datos|otro",
  "hallazgos": [
    {{
      "categoria": "dato_clave|riesgo|oportunidad|tendencia",
      "descripcion": "descripción del hallazgo",
      "impacto": "alto|medio|bajo",
      "evidencia": "cita o dato que lo sustenta"
    }}
  ],
  "recomendaciones": [
    {{
      "accion": "acción específica a tomar",
      "prioridad": "urgente|importante|nice-to-have",
      "responsable": "área o rol (puede ser null)",
      "plazo": "inmediato|30 días|90 días (puede ser null)"
    }}
  ],
  "score_confianza": 0.0
}}

Máximo 10 hallazgos y 5 recomendaciones. Mantén las descripciones concisas pero completas."""


def ejecutar_etapa(
    nombre: str,
    prompt: str,
    modelo: str,
    max_tokens: int,
    max_retries: int = 2,
    json_mode: bool = False,
    verbose: bool = True
) -> tuple[str, int, int]:
    """
    Ejecuta una etapa del pipeline con retry automático.
    
    Returns:
        Tupla (respuesta, retries_realizados, tokens_usados)
    """
    ultimo_error = None
    
    for intento in range(max_retries + 1):
        try:
            kwargs = {
                "model": modelo,
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0,
                "max_tokens": max_tokens
            }
            if json_mode:
                kwargs["response_format"] = {"type": "json_object"}
            
            response = client.chat.completions.create(**kwargs)
            respuesta = response.choices[0].message.content
            tokens = response.usage.total_tokens
            
            if verbose and intento > 0:
                logger.info(f"  {nombre}: Éxito en intento {intento + 1}")
            
            return respuesta, intento, tokens
            
        except Exception as e:
            ultimo_error = str(e)
            if verbose:
                logger.warning(f"  {nombre}: Error en intento {intento + 1}: {e}")
            
            if intento < max_retries:
                time.sleep(2 ** intento)  # Exponential backoff
    
    raise RuntimeError(f"Etapa '{nombre}' falló tras {max_retries + 1} intentos. Último error: {ultimo_error}")


def etapa_extract(state: EstadoPipeline, verbose: bool = True) -> None:
    """Etapa 1: Extracción de información estructurada."""
    if verbose:
        logger.info("▶ Etapa 1/4: EXTRACT")
    
    t0 = time.time()
    prompt = PROMPT_EXTRACT.format(documento=state.documento_procesado)
    
    respuesta, retries, tokens = ejecutar_etapa(
        nombre="extract",
        prompt=prompt,
        modelo="gpt-4o-mini",
        max_tokens=1500,
        verbose=verbose
    )
    
    state.output_extract = respuesta
    state.tiempos_por_etapa["extract"] = time.time() - t0
    state.modelos_usados["extract"] = "gpt-4o-mini"
    state.retries_por_etapa["extract"] = retries
    state.tokens_totales_usados += tokens
    
    if verbose:
        logger.info(f"  ✓ Extract: {len(respuesta)} chars, {time.time()-t0:.1f}s")


def etapa_analyze(state: EstadoPipeline, verbose: bool = True) -> None:
    """Etapa 2: Análisis profundo con gpt-4o."""
    if verbose:
        logger.info("▶ Etapa 2/4: ANALYZE")
    
    t0 = time.time()
    
    # Si el documento era muy largo, usar gpt-4o para mayor capacidad de síntesis
    modelo = "gpt-4o" if state.fue_resumido or state.tokens_input > 3000 else "gpt-4o-mini"
    prompt = PROMPT_ANALYZE.format(extracted=state.output_extract)
    
    respuesta, retries, tokens = ejecutar_etapa(
        nombre="analyze",
        prompt=prompt,
        modelo=modelo,
        max_tokens=2000,
        verbose=verbose
    )
    
    state.output_analyze = respuesta
    state.tiempos_por_etapa["analyze"] = time.time() - t0
    state.modelos_usados["analyze"] = modelo
    state.retries_por_etapa["analyze"] = retries
    state.tokens_totales_usados += tokens
    
    if verbose:
        logger.info(f"  ✓ Analyze ({modelo}): {len(respuesta)} chars, {time.time()-t0:.1f}s")


def etapa_synthesize(state: EstadoPipeline, verbose: bool = True) -> None:
    """Etapa 3: Síntesis y generación de recomendaciones."""
    if verbose:
        logger.info("▶ Etapa 3/4: SYNTHESIZE")
    
    t0 = time.time()
    prompt = PROMPT_SYNTHESIZE.format(analyzed=state.output_analyze)
    
    respuesta, retries, tokens = ejecutar_etapa(
        nombre="synthesize",
        prompt=prompt,
        modelo="gpt-4o",
        max_tokens=2000,
        verbose=verbose
    )
    
    state.output_synthesize = respuesta
    state.tiempos_por_etapa["synthesize"] = time.time() - t0
    state.modelos_usados["synthesize"] = "gpt-4o"
    state.retries_por_etapa["synthesize"] = retries
    state.tokens_totales_usados += tokens
    
    if verbose:
        logger.info(f"  ✓ Synthesize: {len(respuesta)} chars, {time.time()-t0:.1f}s")


def etapa_format(state: EstadoPipeline, verbose: bool = True) -> None:
    """Etapa 4: Formateo en JSON estructurado con validación Pydantic."""
    if verbose:
        logger.info("▶ Etapa 4/4: FORMAT")
    
    t0 = time.time()
    prompt = PROMPT_FORMAT.format(synthesized=state.output_synthesize)
    
    respuesta, retries, tokens = ejecutar_etapa(
        nombre="format",
        prompt=prompt,
        modelo="gpt-4o-mini",
        max_tokens=2000,
        json_mode=True,
        verbose=verbose
    )
    
    # Parsear y validar con Pydantic
    try:
        datos = json.loads(respuesta)
        
        # Añadir metadata del pipeline
        tiempo_total = time.time() - state.tiempo_inicio
        datos["metadata"] = {
            "tokens_originales": state.tokens_input,
            "tokens_procesados": contar_tokens(state.documento_procesado),
            "tokens_api_total": state.tokens_totales_usados + tokens,
            "fue_resumido": state.fue_resumido,
            "tiempo_total_segundos": round(tiempo_total, 2),
            "tiempos_por_etapa": {k: round(v, 2) for k, v in state.tiempos_por_etapa.items()},
            "modelos_usados": state.modelos_usados,
            "retries_totales": sum(state.retries_por_etapa.values())
        }
        
        state.output_final = ResultadoAnalisis(**datos)
        
    except Exception as e:
        logger.error(f"Error validando JSON: {e}")
        logger.error(f"JSON recibido: {respuesta[:200]}")
        # Fallback: crear un ResultadoAnalisis mínimo
        state.output_final = ResultadoAnalisis(
            titulo="Análisis de documento",
            resumen_ejecutivo=state.output_synthesize[:300] if state.output_synthesize else "No disponible",
            tipo_documento="otro",
            hallazgos=[Hallazgo(
                categoria="dato_clave",
                descripcion="Ver análisis en texto plano",
                impacto="medio",
                evidencia=state.output_analyze[:200] if state.output_analyze else ""
            )],
            recomendaciones=[Recomendacion(
                accion="Revisar análisis manualmente",
                prioridad="importante"
            )],
            score_confianza=0.5,
            metadata={"error": str(e), "fallback": True}
        )
    
    state.tiempos_por_etapa["format"] = time.time() - t0
    state.modelos_usados["format"] = "gpt-4o-mini"
    state.retries_por_etapa["format"] = retries
    state.tokens_totales_usados += tokens
    
    if verbose:
        logger.info(f"  ✓ Format: validado con Pydantic, {time.time()-t0:.1f}s")

Pipeline principal

# ============================================================
# PIPELINE PRINCIPAL
# ============================================================

def analizar_documento(
    documento: str,
    max_tokens_contexto: int = 8000,
    verbose: bool = True
) -> ResultadoAnalisis:
    """
    Pipeline completo de análisis de documento.
    
    Args:
        documento: Texto del documento a analizar
        max_tokens_contexto: Umbral para activar summarización
        verbose: Mostrar logs detallados
    
    Returns:
        ResultadoAnalisis con hallazgos y recomendaciones estructurados
    """
    if verbose:
        logger.info("=" * 60)
        logger.info("INICIANDO MULTI-STAGE ANALYSIS PIPELINE")
        logger.info("=" * 60)
    
    # Pre-procesamiento: gestión de contexto
    documento_procesado, fue_resumido, tokens_originales = preprocesar_documento(
        documento,
        max_tokens=max_tokens_contexto,
        verbose=verbose
    )
    
    # Inicializar estado del pipeline
    state = EstadoPipeline(
        documento_original=documento,
        documento_procesado=documento_procesado,
        tokens_input=tokens_originales,
        fue_resumido=fue_resumido
    )
    
    # Ejecutar las cuatro etapas
    try:
        etapa_extract(state, verbose=verbose)
        etapa_analyze(state, verbose=verbose)
        etapa_synthesize(state, verbose=verbose)
        etapa_format(state, verbose=verbose)
        
        if verbose:
            tiempo_total = time.time() - state.tiempo_inicio
            logger.info("=" * 60)
            logger.info(f"✅ PIPELINE COMPLETADO en {tiempo_total:.1f}s")
            logger.info(f"   Tokens API usados: {state.tokens_totales_usados:,}")
            logger.info(f"   Hallazgos: {len(state.output_final.hallazgos)}")
            logger.info(f"   Recomendaciones: {len(state.output_final.recomendaciones)}")
            logger.info("=" * 60)
        
        return state.output_final
        
    except Exception as e:
        logger.error(f"Error en pipeline: {e}")
        state.errores.append(str(e))
        raise

Función de presentación de resultados

def imprimir_resultado(resultado: ResultadoAnalisis) -> None:
    """Imprime el resultado del análisis de forma legible."""
    print("\n" + "=" * 70)
    print(f"📄 {resultado.titulo}")
    print(f"Tipo: {resultado.tipo_documento} | Confianza: {resultado.score_confianza:.0%}")
    print("=" * 70)
    
    print(f"\n📋 RESUMEN EJECUTIVO\n{resultado.resumen_ejecutivo}")
    
    print(f"\n🔍 HALLAZGOS CLAVE ({len(resultado.hallazgos)})")
    for i, h in enumerate(resultado.hallazgos, 1):
        impacto_emoji = {"alto": "🔴", "medio": "🟡", "bajo": "🟢"}.get(h.impacto, "⚪")
        print(f"\n  {i}. {impacto_emoji} [{h.categoria.upper()}] {h.descripcion}")
        print(f"     Evidencia: {h.evidencia[:100]}...")
    
    print(f"\n💡 RECOMENDACIONES ({len(resultado.recomendaciones)})")
    for i, r in enumerate(resultado.recomendaciones, 1):
        prio_emoji = {"urgente": "🚨", "importante": "⚡", "nice-to-have": "💭"}.get(r.prioridad, "•")
        print(f"\n  {i}. {prio_emoji} {r.accion}")
        if r.responsable:
            print(f"     Responsable: {r.responsable}", end="")
        if r.plazo:
            print(f" | Plazo: {r.plazo}", end="")
        print()
    
    if resultado.metadata:
        meta = resultado.metadata
        print(f"\n⚙️ METADATA DEL PIPELINE")
        print(f"  Tiempo total: {meta.get('tiempo_total_segundos', 'N/A')}s")
        print(f"  Tokens usados: {meta.get('tokens_api_total', 'N/A'):,}")
        print(f"  Fue resumido: {'Sí' if meta.get('fue_resumido') else 'No'}")
        if meta.get("retries_totales", 0) > 0:
            print(f"  Retries: {meta['retries_totales']}")

Prueba del Pipeline

# ============================================================
# DOCUMENTOS DE PRUEBA
# ============================================================

DOCUMENTO_CORTO = """
INFORME TRIMESTRAL Q3 2024 - TECH STARTUP XYZ

Resumen ejecutivo:
En el tercer trimestre de 2024, XYZ alcanzó un crecimiento del 45% en ingresos respecto al trimestre anterior, 
con ingresos totales de $2.3M. Sin embargo, los gastos operativos aumentaron un 62%, principalmente debido a 
la expansión del equipo de ingeniería de 12 a 28 personas y la apertura de nuestra oficina en México.

Métricas clave:
- Usuarios activos mensuales (MAU): 45,000 (+38% vs Q2)
- Tasa de retención a 30 días: 72% (objetivo: 75%)
- Costo de adquisición de cliente (CAC): $85 (vs $120 en Q2, mejora del 29%)
- Lifetime Value promedio (LTV): $340 (ratio LTV/CAC: 4.0x)
- Net Promoter Score: 67 (industry benchmark: 45)

Logros del trimestre:
1. Lanzamiento de la feature de "Smart Analytics" que incrementó engagement en 23%
2. Firma de contrato enterprise con RetailMega Corp ($400K ARR)
3. Publicación en TechCrunch que generó 1,200 leads orgánicos
4. Reducción del tiempo de onboarding de 14 días a 5 días

Desafíos:
- La expansión del equipo está presionando los márgenes operativos (margen bruto bajó de 68% a 61%)
- Tres clientes enterprise pausaron contratos por "presupuesto Q4"
- El pipeline de ventas enterprise tiene retrasos promedio de 45 días

Próximos pasos para Q4:
- Optimizar proceso de ventas enterprise para reducir ciclo a 30 días
- Lanzar programa de partners para expansión en LATAM
- Implementar pricing tier para SMBs ($49/mes) para diversificar base de clientes
- Meta de ingresos Q4: $3.2M (+39% vs Q3)
"""

DOCUMENTO_LARGO = """
ANÁLISIS DE MERCADO: INTELIGENCIA ARTIFICIAL EN LATINOAMÉRICA 2024-2026

SECCIÓN 1: PANORAMA GENERAL

La adopción de inteligencia artificial en Latinoamérica ha experimentado un crecimiento sin precedentes 
durante 2023-2024. Según el informe de IDC Latinoamérica, el gasto en tecnologías de IA en la región 
alcanzó $4.2B USD en 2023, con una proyección de $9.8B para 2026, representando un CAGR del 32.5%.

Brasil lidera con el 48% del mercado regional, seguido por México (22%), Colombia (10%), Argentina (8%) 
y Chile (6%). Los sectores con mayor adopción son: Servicios Financieros (35%), Retail (22%), 
Telecomunicaciones (18%) y Salud (12%).

Los principales catalizadores de crecimiento son: reducción del 40% en costos de computación en la nube 
en los últimos 3 años, disponibilidad de APIs de LLMs en español, marcos regulatorios más claros 
(especialmente en Brasil y Chile), y talento técnico creciente (la región graduó 180,000 ingenieros en 2023).

SECCIÓN 2: BARRERAS DE ADOPCIÓN

La principal barrera identificada es la brecha de talento: existe un déficit estimado de 85,000 
profesionales especializados en IA en toda la región. Las empresas reportan que el 67% de sus proyectos 
de IA fallan o se retrasan por falta de talento calificado.

La segunda barrera es la infraestructura de datos. El 72% de las empresas latinoamericanas aún mantiene 
sus datos en sistemas legacy, y solo el 23% tiene una estrategia de datos estructurada necesaria para 
proyectos de IA. Esto contrasta con el 64% de empresas norteamericanas que ya tienen data lakes en la nube.

Los costos de implementación también representan una barrera. El proyecto de IA promedio en LATAM 
requiere $500K-$2M de inversión inicial, inaccesible para las PyMEs que representan el 99% del tejido 
empresarial de la región.

SECCIÓN 3: CASOS DE ÉXITO

Nubank (Brasil): Implementó modelos de credit scoring con ML que redujeron la tasa de morosidad en 31% 
mientras expandían el acceso al crédito a 8M de nuevos clientes sin historial bancario. 
ROI reportado: 340% en 18 meses.

Rappi (Colombia/México): Su sistema de recomendación impulsado por IA incrementó el valor promedio 
de pedido en 22% y redujo el tiempo de entrega predicho en un 15%, mejorando el NPS en 12 puntos.

MercadoLibre: Su sistema antifraude basado en ML bloquea 98.7% de transacciones fraudulentas 
en tiempo real, ahorrando estimados $800M anuales en fraudes evitados.

SECCIÓN 4: PERSPECTIVAS 2025-2026

Las tendencias que dominarán el mercado LATAM en los próximos dos años:

1. IA Generativa para Contenido en Español: El mercado de herramientas de generación de contenido 
en español crecerá 180% en 2025, impulsado por la demanda de marketing personalizado.

2. IA en Salud: Se esperan 50 nuevas startups de HealthTech IA en Brasil y México, con especial 
énfasis en diagnóstico por imágenes y telemedicina rural.

3. Regulación: Chile promulgará su Ley de IA en Q1 2025. Brasil está finalizando su marco regulatorio.
México aún está en etapas tempranas de definición regulatoria.

4. Talento: Los salarios de ML Engineers en LATAM crecieron 45% en 2024 y se espera un crecimiento 
adicional del 25-30% en 2025-2026. La región verá la creación de 45,000 nuevos empleos en IA.

CONCLUSIONES Y RECOMENDACIONES

Para empresas que buscan implementar IA en LATAM, las principales recomendaciones son:
- Priorizar la construcción de capacidades internas de datos antes de implementar IA
- Considerar partnerships con universidades locales para desarrollo de talento
- Evaluar soluciones de IA como servicio (AIaaS) para reducir inversión inicial
- Monitorear el desarrollo regulatorio, especialmente en Chile y Brasil
- Explorar casos de uso de alto impacto y menor complejidad como punto de partida
""" * 2  # Multiplicar para simular un documento más largo


if __name__ == "__main__":
    print("=" * 70)
    print("PRUEBA 1: Documento corto (informe trimestral)")
    print("=" * 70)
    
    resultado = analizar_documento(DOCUMENTO_CORTO, verbose=True)
    imprimir_resultado(resultado)
    
    # Exportar a JSON
    with open("resultado_analisis.json", "w", encoding="utf-8") as f:
        json.dump(resultado.model_dump(), f, ensure_ascii=False, indent=2)
    print("\n✅ Resultado exportado a resultado_analisis.json")
    
    print("\n" + "=" * 70)
    print("PRUEBA 2: Documento largo (análisis de mercado)")
    print("=" * 70)
    
    resultado_largo = analizar_documento(DOCUMENTO_LARGO, max_tokens_contexto=6000, verbose=True)
    imprimir_resultado(resultado_largo)

Extensiones Opcionales

Extensión 1: Soporte para múltiples idiomas

IDIOMAS_SOPORTADOS = {
    "es": "Spanish",
    "en": "English",
    "pt": "Portuguese"
}

def analizar_documento_multilingue(
    documento: str,
    idioma_output: str = "es"
) -> ResultadoAnalisis:
    """
    Pipeline con soporte multilingüe.
    Detecta el idioma del input y genera output en el idioma especificado.
    """
    # Detectar idioma
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Detecta el idioma de este texto (responde solo con el código ISO 639-1 de 2 letras): {documento[:200]}"
        }],
        temperature=0, max_tokens=5
    )
    idioma_input = response.choices[0].message.content.strip().lower()[:2]
    
    instruccion_idioma = f"\n\nIMPORTANTE: Genera toda tu respuesta en {IDIOMAS_SOPORTADOS.get(idioma_output, 'Spanish')}."
    
    # Inyectar instrucción de idioma en los prompts
    prompts_modificados = {
        "extract": PROMPT_EXTRACT + instruccion_idioma,
        "analyze": PROMPT_ANALYZE + instruccion_idioma,
        "synthesize": PROMPT_SYNTHESIZE + instruccion_idioma,
    }
    
    # Ejecutar pipeline con prompts modificados...
    # (implementación completa similar a analizar_documento)
    return analizar_documento(documento)

Extensión 2: Pipeline asíncrono para procesar múltiples documentos

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def analizar_documento_async(
    documento: str,
    nombre: str = "documento"
) -> tuple[str, ResultadoAnalisis]:
    """Versión asíncrona del pipeline para procesamiento paralelo."""
    # Implementación asíncrona del pipeline
    # Cada etapa usa async_client.chat.completions.create(...)
    resultado = analizar_documento(documento, verbose=False)
    return nombre, resultado


async def analizar_multiples_documentos(
    documentos: dict[str, str]  # nombre -> contenido
) -> dict[str, ResultadoAnalisis]:
    """Analiza múltiples documentos en paralelo."""
    tareas = [
        analizar_documento_async(contenido, nombre)
        for nombre, contenido in documentos.items()
    ]
    resultados_lista = await asyncio.gather(*tareas, return_exceptions=True)
    
    resultados = {}
    for item in resultados_lista:
        if isinstance(item, Exception):
            print(f"Error: {item}")
        else:
            nombre, resultado = item
            resultados[nombre] = resultado
    
    return resultados


# Uso:
documentos_batch = {
    "informe_q3": DOCUMENTO_CORTO,
    "analisis_mercado": DOCUMENTO_LARGO[:3000]
}
# resultados = asyncio.run(analizar_multiples_documentos(documentos_batch))

Criterios de Éxito del Proyecto

Evalúa tu implementación contra estos criterios:

Total: 100 puntos. Mínimo para aprobar: 70 pts.

Checklist de validación

def validar_pipeline_completo() -> bool:
    """Suite de validación del pipeline."""
    print("=== VALIDACIÓN DEL PIPELINE ===\n")
    checks = []
    
    # Check 1: Documento corto
    print("1. Procesando documento corto...")
    try:
        res = analizar_documento(DOCUMENTO_CORTO[:500], verbose=False)
        assert res.titulo, "Falta título"
        assert len(res.hallazgos) >= 1, "Sin hallazgos"
        assert len(res.recomendaciones) >= 1, "Sin recomendaciones"
        assert 0 <= res.score_confianza <= 1, "Score inválido"
        checks.append(("Documento corto", True))
        print("   ✅ OK")
    except Exception as e:
        checks.append(("Documento corto", False))
        print(f"   ❌ FALLO: {e}")
    
    # Check 2: Context management
    print("\n2. Probando context management...")
    texto_largo = "Este es un documento. " * 1000  # ~4000 palabras
    doc_proc, fue_resumido, tokens = preprocesar_documento(texto_largo, max_tokens=500, verbose=False)
    if fue_resumido:
        checks.append(("Context management", True))
        print("   ✅ Summarización activada correctamente")
    else:
        checks.append(("Context management", False))
        print("   ❌ No se activó la summarización")
    
    # Check 3: Formato JSON
    print("\n3. Validando output Pydantic...")
    try:
        fake_data = {
            "titulo": "Test",
            "resumen_ejecutivo": "Resumen de prueba.",
            "tipo_documento": "informe",
            "hallazgos": [{"categoria": "dato_clave", "descripcion": "Test", "impacto": "alto", "evidencia": "Test"}],
            "recomendaciones": [{"accion": "Test", "prioridad": "urgente"}],
            "score_confianza": 0.8
        }
        resultado = ResultadoAnalisis(**fake_data)
        checks.append(("Validación Pydantic", True))
        print("   ✅ Pydantic funciona correctamente")
    except Exception as e:
        checks.append(("Validación Pydantic", False))
        print(f"   ❌ Error Pydantic: {e}")
    
    # Resumen
    total = len(checks)
    exitosos = sum(1 for _, ok in checks if ok)
    print(f"\n=== RESULTADO: {exitosos}/{total} checks pasados ===")
    
    return exitosos == total


if __name__ == "__main__":
    validar_pipeline_completo()

Ejercicios

Ejercicio 1: Agregar una etapa de "Fact-Checking"

Añade una quinta etapa entre Analyze y Synthesize que verifique si los hallazgos del análisis son coherentes con el documento original.

PROMPT_FACTCHECK = """Eres un verificador de hechos riguroso. 
Revisa si los siguientes hallazgos del análisis están respaldados por el documento original.

Documento original (resumido):
{documento}

Análisis a verificar:
{analyzed}

Para cada afirmación importante:
1. ¿Está directamente respaldada por el documento? (sí/no/parcialmente)
2. ¿Hay exageración o interpretación excesiva?
3. ¿Hay afirmaciones que contradicen el documento?

Genera una versión corregida del análisis que:
- Elimine afirmaciones no respaldadas
- Modere afirmaciones exageradas
- Mantenga solo lo que el documento sustenta
- Indique el nivel de certeza (alta/media/baja) para cada punto"""

def etapa_factcheck(state: EstadoPipeline, verbose: bool = True) -> None:
    """Etapa adicional de verificación de hechos."""
    if verbose:
        logger.info("▶ Etapa FACTCHECK")
    
    t0 = time.time()
    prompt = PROMPT_FACTCHECK.format(
        documento=state.documento_procesado[:2000],
        analyzed=state.output_analyze
    )
    
    respuesta, retries, tokens = ejecutar_etapa(
        nombre="factcheck",
        prompt=prompt,
        modelo="gpt-4o-mini",
        max_tokens=1500,
        verbose=verbose
    )
    
    # Reemplazar el analyze con la versión verificada
    state.output_analyze = respuesta
    state.tiempos_por_etapa["factcheck"] = time.time() - t0
    state.tokens_totales_usados += tokens
    
    if verbose:
        logger.info(f"  ✓ Factcheck: {time.time()-t0:.1f}s")

Ejercicio 2: Exportar a múltiples formatos

Implementa funciones para exportar el ResultadoAnalisis a Markdown y HTML además de JSON.

def exportar_markdown(resultado: ResultadoAnalisis) -> str:
    """Genera un reporte en formato Markdown."""
    md = f"# {resultado.titulo}\n\n"
    md += f"**Tipo:** {resultado.tipo_documento} | **Confianza:** {resultado.score_confianza:.0%}\n\n"
    md += f"## Resumen Ejecutivo\n{resultado.resumen_ejecutivo}\n\n"
    md += "## Hallazgos Clave\n\n"
    
    for i, h in enumerate(resultado.hallazgos, 1):
        impacto_badge = f"![{h.impacto}]()"
        md += f"### {i}. {h.descripcion}\n"
        md += f"- **Categoría:** {h.categoria}\n"
        md += f"- **Impacto:** {h.impacto}\n"
        md += f"- **Evidencia:** _{h.evidencia}_\n\n"
    
    md += "## Recomendaciones\n\n"
    for i, r in enumerate(resultado.recomendaciones, 1):
        md += f"### {i}. {r.accion}\n"
        md += f"- **Prioridad:** {r.prioridad}\n"
        if r.responsable:
            md += f"- **Responsable:** {r.responsable}\n"
        if r.plazo:
            md += f"- **Plazo:** {r.plazo}\n"
        md += "\n"
    
    return md

def exportar_html(resultado: ResultadoAnalisis) -> str:
    """Genera un reporte en formato HTML básico."""
    html = f"<html><body><h1>{resultado.titulo}</h1>"
    html += f"<p><b>Tipo:</b> {resultado.tipo_documento}</p>"
    html += f"<h2>Resumen</h2><p>{resultado.resumen_ejecutivo}</p>"
    html += "<h2>Hallazgos</h2><ul>"
    for h in resultado.hallazgos:
        html += f"<li><b>{h.descripcion}</b> ({h.impacto}): {h.evidencia[:80]}...</li>"
    html += "</ul>"
    html += "<h2>Recomendaciones</h2><ol>"
    for r in resultado.recomendaciones:
        html += f"<li>{r.accion} [{r.prioridad}]</li>"
    html += "</ol></body></html>"
    return html

# Uso:
# resultado = analizar_documento(DOCUMENTO_CORTO)
# with open("informe.md", "w") as f:
#     f.write(exportar_markdown(resultado))
# with open("informe.html", "w") as f:
#     f.write(exportar_html(resultado))

Resumen del Módulo

Has completado el Módulo 06: Prompt Composition. Este proyecto ha demostrado cómo integrar todas las técnicas del módulo en un sistema de producción real.

Lo que construiste:

• Prompt Chaining (Cápsula 02): Las 4 etapas secuenciales, cada una recibiendo el output de la anterior
• Task Decomposition (Cápsula 03): Extract, Analyze, Synthesize, Format como sub-tareas especializadas
• Multi-Stage Pipelines (Cápsula 04): EstadoPipeline, retry, fallbacks, logging de métricas
• Context Window Management (Cápsula 05): Chunking con tiktoken, summarización automática
• Routing por Complejidad (Cápsula 07): Mini para extracción/formato, gpt-4o para análisis/síntesis

Próximos pasos sugeridos:

1. Añadir persistencia de resultados en base de datos
2. Crear una API con FastAPI que reciba documentos y retorne análisis
3. Implementar procesamiento asíncrono con cola de trabajos
4. Añadir cache para documentos ya analizados
5. Construir dashboard de monitoreo con métricas de uso

Recursos adicionales

1. tiktoken GitHub
2. Pydantic v2 documentation
3. OpenAI Structured Outputs
4. LangChain Document Loaders
5. Building LLM-Powered Apps - Patterns and Architectures
6. Python asyncio documentation