Módulo 6: Prompt Composition y Chaining

3. Decomposition de Tareas Complejas

Siguiente →

Descripción

La decomposition es el proceso sistemático de dividir una tarea compleja en sub-tareas manejables, cada una con un input claro, un prompt especializado, y un output con schema validable. Es la habilidad fundamental para diseñar pipelines de LLM escalables.

En esta cápsula aprenderás el framework completo de decomposition: cómo identificar los límites naturales de las sub-tareas, cómo diseñar interfaces entre prompts, y cómo manejar dependencias para permitir paralelización cuando sea posible.

El Framework de Decomposition

Los 4 Pasos

1. IDENTIFICAR sub-tareas
   └── ¿Qué pasos son independientes? ¿Cuáles dependen de otros?

2. DEFINIR interfaces
   └── Input/output de cada etapa (schema con tipos de datos)

3. DISEÑAR prompts
   └── Un prompt especializado por sub-tarea

4. ORDENAR con dependencias
   └── Qué puede correr en paralelo vs qué debe ser secuencial

Cuándo dividir una tarea

Una tarea debe dividirse cuando:

  • Tiene múltiples objetivos distintos (extraer + analizar + reportar)
  • El output de una parte es necesario como input de otra
  • Partes de la tarea pueden optimizarse independientemente
  • El error handling debe ser granular (reintentar solo la parte fallida)
  • Partes independientes pueden ejecutarse en paralelo

Ejemplo Completo: Research → Analyze → Write

from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Optional
import json

client = OpenAI()

# ============================================================
# SCHEMAS DE INTERFACES (Pydantic)
# ============================================================

class HallazgoResearch(BaseModel):
    """Output de la sub-tarea de research."""
    hallazgos: list[str] = Field(description="Lista de hallazgos clave")
    fuentes_tipo: list[str] = Field(description="Tipo de fuente de cada hallazgo")
    confianza: float = Field(ge=0, le=1, description="Confianza general en los hallazgos")

class AnalisisOutput(BaseModel):
    """Output de la sub-tarea de análisis."""
    tendencias: list[str] = Field(description="Tendencias identificadas")
    evidencia: dict[str, list[str]] = Field(description="Evidencia por tendencia")
    contradicciones: list[str] = Field(description="Contradicciones encontradas")
    conclusion_principal: str = Field(description="La conclusión más importante")

class SeccionInforme(BaseModel):
    """Una sección del informe final."""
    titulo: str
    contenido: str
    citas: list[str]

class InformeOutput(BaseModel):
    """Output final del pipeline."""
    titulo: str
    resumen_ejecutivo: str
    secciones: list[SeccionInforme]
    conclusiones: list[str]
    recomendaciones: list[str]


# ============================================================
# PROMPTS ESPECIALIZADOS
# ============================================================

PROMPT_RESEARCH = """Eres un investigador experto. Recopila información sobre el siguiente tema.

TEMA: {query}

Proporciona 8-12 hallazgos clave basados en tu conocimiento. Para cada hallazgo:
- Sé específico y basado en hechos conocidos
- Indica el tipo de fuente (estadística, estudio académico, tendencia de mercado, ejemplo práctico)

Responde en JSON:
{{
    "hallazgos": [
        "hallazgo 1",
        "hallazgo 2"
    ],
    "fuentes_tipo": [
        "estadística",
        "estudio"
    ],
    "confianza": 0.85
}}"""

PROMPT_ANALYZE = """Eres un analista senior. Sintetiza los siguientes hallazgos en insights accionables.

HALLAZGOS:
{hallazgos_str}

TEMA ORIGINAL: {query}

Identifica:
1. Tendencias principales (3-5 tendencias claras)
2. Evidencia que soporta cada tendencia
3. Contradicciones o tensiones entre hallazgos
4. La conclusión más importante

Responde en JSON:
{{
    "tendencias": ["tendencia 1", "tendencia 2"],
    "evidencia": {{
        "tendencia 1": ["evidencia A", "evidencia B"],
        "tendencia 2": ["evidencia C"]
    }},
    "contradicciones": ["contradicción si existe"],
    "conclusion_principal": "La conclusión más importante en 2 oraciones"
}}"""

PROMPT_WRITE = """Eres un redactor técnico. Genera un informe profesional basado en este análisis.

TEMA: {query}
CONCLUSIÓN PRINCIPAL: {conclusion}
TENDENCIAS IDENTIFICADAS:
{tendencias_str}

El informe debe:
1. Tener título descriptivo
2. Resumen ejecutivo (2-3 oraciones)
3. 2-3 secciones principales (una por tendencia clave)
4. Conclusiones (3 puntos)
5. Recomendaciones concretas (3-5 acciones)

Responde en JSON según el schema provisto."""


# ============================================================
# IMPLEMENTACIÓN DEL PIPELINE
# ============================================================

def sub_tarea_research(query: str) -> HallazgoResearch:
    """
    Sub-tarea 1: Research - Recopilar información sobre el tema.
    """
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": PROMPT_RESEARCH.format(query=query)}],
        temperature=0,
        max_tokens=800,
        response_format={"type": "json_object"}
    )
    
    datos = json.loads(response.choices[0].message.content)
    return HallazgoResearch(**datos)

def sub_tarea_analyze(query: str, research: HallazgoResearch) -> AnalisisOutput:
    """
    Sub-tarea 2: Analysis - Sintetizar hallazgos en insights.
    Depende de: sub_tarea_research
    """
    hallazgos_str = "\n".join([f"- {h}" for h in research.hallazgos])
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": PROMPT_ANALYZE.format(
                hallazgos_str=hallazgos_str,
                query=query
            )
        }],
        temperature=0,
        max_tokens=800,
        response_format={"type": "json_object"}
    )
    
    datos = json.loads(response.choices[0].message.content)
    return AnalisisOutput(**datos)

def sub_tarea_write(query: str, analisis: AnalisisOutput) -> InformeOutput:
    """
    Sub-tarea 3: Write - Generar informe estructurado.
    Depende de: sub_tarea_analyze
    """
    tendencias_str = "\n".join([
        f"- {t}: {', '.join(analisis.evidencia.get(t, [])[:2])}"
        for t in analisis.tendencias
    ])
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": PROMPT_WRITE.format(
                query=query,
                conclusion=analisis.conclusion_principal,
                tendencias_str=tendencias_str
            )
        }],
        temperature=0.2,
        max_tokens=1000,
        response_format={"type": "json_object"}
    )
    
    datos = json.loads(response.choices[0].message.content)
    # Construir InformeOutput (manejo flexible del schema)
    secciones = [SeccionInforme(**s) for s in datos.get("secciones", [])]
    return InformeOutput(
        titulo=datos.get("titulo", "Informe"),
        resumen_ejecutivo=datos.get("resumen_ejecutivo", ""),
        secciones=secciones,
        conclusiones=datos.get("conclusiones", []),
        recomendaciones=datos.get("recomendaciones", [])
    )

def pipeline_research_analyze_write(query: str, verbose: bool = True) -> dict:
    """
    Pipeline completo: Research → Analyze → Write
    
    Args:
        query: El tema a investigar y sobre el que escribir
        verbose: Si True, imprime el progreso
    
    Returns:
        dict con los outputs de las 3 etapas
    """
    import time
    t_total = time.time()
    
    if verbose:
        print(f"📚 Investigando: {query}\n")
    
    # Etapa 1: Research
    t0 = time.time()
    research = sub_tarea_research(query)
    if verbose:
        print(f"✓ Research completado ({time.time()-t0:.1f}s)")
        print(f"  {len(research.hallazgos)} hallazgos, confianza: {research.confianza:.0%}")
    
    # Etapa 2: Analysis
    t0 = time.time()
    analisis = sub_tarea_analyze(query, research)
    if verbose:
        print(f"✓ Análisis completado ({time.time()-t0:.1f}s)")
        print(f"  {len(analisis.tendencias)} tendencias identificadas")
        print(f"  Conclusión: {analisis.conclusion_principal[:80]}...")
    
    # Etapa 3: Write
    t0 = time.time()
    informe = sub_tarea_write(query, analisis)
    if verbose:
        print(f"✓ Informe generado ({time.time()-t0:.1f}s)")
        print(f"  Título: {informe.titulo}")
        print(f"  {len(informe.secciones)} secciones, {len(informe.recomendaciones)} recomendaciones")
        print(f"\nTiempo total: {time.time()-t_total:.1f}s")
    
    return {
        "research": research.model_dump(),
        "analisis": analisis.model_dump(),
        "informe": informe.model_dump()
    }


# Ejemplo:
if __name__ == "__main__":
    resultado = pipeline_research_analyze_write(
        "El impacto de los LLMs en el mercado laboral de programación en 2025-2026"
    )
    print("\n=== INFORME FINAL ===")
    informe = resultado["informe"]
    print(f"# {informe['titulo']}\n")
    print(f"## Resumen Ejecutivo\n{informe['resumen_ejecutivo']}\n")
    print("## Recomendaciones")
    for r in informe["recomendaciones"]:
        print(f"- {r}")

Framework de Decomposition: Patrones Comunes

Patrón 1: Extract → Transform → Load (ETL para LLMs)

class ETLPipeline:
    """
    Patrón ETL adaptado para pipelines de LLM:
    - Extract: Extraer información del input (entidades, datos, estructura)
    - Transform: Transformar/analizar los datos extraídos
    - Load: Formatear el output en el schema deseado
    """
    
    def __init__(self, dominio: str):
        self.dominio = dominio
    
    def extract(self, documento: str) -> dict:
        """Extraer información estructurada del documento."""
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"""Extrae la información clave de este {self.dominio}:
            
{documento}

Responde en JSON con los campos relevantes para este tipo de documento."""}],
            temperature=0,
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)
    
    def transform(self, datos_extraidos: dict, objetivo: str) -> dict:
        """Transformar/analizar los datos según el objetivo."""
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"""Analiza estos datos para el objetivo: {objetivo}

Datos: {json.dumps(datos_extraidos, ensure_ascii=False)}

Genera insights, calcula métricas relevantes, identifica patrones.
Responde en JSON."""}],
            temperature=0,
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)
    
    def load(self, datos_transformados: dict, formato: str) -> str:
        """Formatear el output en el formato deseado."""
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"""Formatea estos datos en {formato}:

{json.dumps(datos_transformados, ensure_ascii=False)}"""}],
            temperature=0.2
        )
        return response.choices[0].message.content
    
    def run(self, documento: str, objetivo: str, formato_salida: str) -> dict:
        extraido = self.extract(documento)
        transformado = self.transform(extraido, objetivo)
        output = self.load(transformado, formato_salida)
        return {"extraido": extraido, "transformado": transformado, "output": output}


# Ejemplo: Análisis de contrato laboral
etl = ETLPipeline("contrato laboral")
resultado = etl.run(
    documento="Contrato de trabajo. Empleado: Juan García. Salario: €45,000/año. Jornada: 40h/semana...",
    objetivo="identificar condiciones no estándar o cláusulas de riesgo",
    formato_salida="resumen ejecutivo en bullet points"
)
print(resultado["output"])

Patrón 2: Map-Reduce para Documentos Largos

from typing import Any

def map_reduce_documentos(
    documentos: list[str],
    prompt_map: str,
    prompt_reduce: str,
    batch_size: int = 5
) -> str:
    """
    Procesa múltiples documentos con Map-Reduce:
    - Map: Aplica prompt_map a cada documento individualmente
    - Reduce: Sintetiza todos los resultados en uno
    
    Args:
        documentos: Lista de textos a procesar
        prompt_map: Template para procesar cada documento. Usa {documento}
        prompt_reduce: Template para sintetizar. Usa {resultados}
        batch_size: Procesar en lotes si hay muchos documentos
    
    Returns:
        Síntesis final de todos los documentos
    """
    # MAP: Procesar cada documento
    resultados_map = []
    for i, doc in enumerate(documentos):
        print(f"Map: procesando documento {i+1}/{len(documentos)}...")
        result = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt_map.format(documento=doc)}],
            temperature=0,
            max_tokens=400
        ).choices[0].message.content
        resultados_map.append(result)
    
    # Si hay muchos resultados, hacer reduce jerárquico
    while len(resultados_map) > batch_size:
        nuevos_resultados = []
        for i in range(0, len(resultados_map), batch_size):
            lote = resultados_map[i:i+batch_size]
            lote_str = "\n---\n".join([f"Resultado {i+j+1}:\n{r}" for j, r in enumerate(lote)])
            
            reduce_parcial = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt_reduce.format(resultados=lote_str)}],
                temperature=0,
                max_tokens=400
            ).choices[0].message.content
            nuevos_resultados.append(reduce_parcial)
        
        resultados_map = nuevos_resultados
    
    # REDUCE: Síntesis final
    todos_str = "\n---\n".join([f"Resultado {i+1}:\n{r}" for i, r in enumerate(resultados_map)])
    
    sintesis_final = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt_reduce.format(resultados=todos_str)}],
        temperature=0.1,
        max_tokens=600
    ).choices[0].message.content
    
    return sintesis_final


# Ejemplo: Analizar múltiples reseñas de clientes
resenas = [
    "El producto es excelente pero el soporte técnico es pésimo.",
    "Llegó tarde pero la calidad superó mis expectativas.",
    "No recomendaría. El manual está en inglés y no hay versión en español.",
    "Perfecto en todos los aspectos. Vale cada euro.",
    "Funciona bien pero se calienta demasiado después de 2 horas de uso.",
]

prompt_map_resenas = "Extrae en 1-2 oraciones el problema o elogio principal de esta reseña: {documento}"
prompt_reduce_resenas = "Sintetiza estos análisis de reseñas en un resumen ejecutivo con: tendencias positivas, tendencias negativas, y 3 acciones prioritarias.\n\n{resultados}"

sintesis = map_reduce_documentos(resenas, prompt_map_resenas, prompt_reduce_resenas)
print("Síntesis de reseñas:", sintesis)

Patrón 3: Decomposition con Paralelización Inteligente

import asyncio
from openai import AsyncOpenAI
from dataclasses import dataclass
from typing import set

@dataclass
class SubTareaConfig:
    nombre: str
    prompt_template: str
    dependencias: list[str]  # Nombres de sub-tareas de las que depende
    puede_paralelizar: bool = True

def construir_dag(sub_tareas: list[SubTareaConfig]) -> dict[str, set]:
    """Construye un DAG (Directed Acyclic Graph) de dependencias."""
    dag = {t.nombre: set(t.dependencias) for t in sub_tareas}
    return dag

def orden_topologico(dag: dict[str, set]) -> list[list[str]]:
    """
    Calcula el orden topológico del DAG.
    Retorna grupos de tareas que pueden ejecutarse en paralelo.
    """
    orden_grupos = []
    dag_copia = {k: set(v) for k, v in dag.items()}
    
    while dag_copia:
        # Tareas sin dependencias pendientes (pueden ejecutarse en este paso)
        sin_deps = [nombre for nombre, deps in dag_copia.items() if not deps]
        
        if not sin_deps:
            raise ValueError("Ciclo detectado en el DAG de dependencias")
        
        orden_grupos.append(sin_deps)
        
        # Remover estas tareas del grafo
        for nombre in sin_deps:
            del dag_copia[nombre]
        
        # Remover estas tareas de las dependencias de las restantes
        for nombre in list(dag_copia.keys()):
            dag_copia[nombre] -= set(sin_deps)
    
    return orden_grupos

async def ejecutar_con_paralelizacion(
    sub_tareas: list[SubTareaConfig],
    input_inicial: str,
    verbose: bool = True
) -> dict[str, str]:
    """
    Ejecuta sub-tareas respetando dependencias y paralelizando cuando es posible.
    """
    client_async = AsyncOpenAI()
    dag = construir_dag(sub_tareas)
    grupos = orden_topologico(dag)
    
    configs_map = {t.nombre: t for t in sub_tareas}
    resultados = {"_input_inicial": input_inicial}
    
    if verbose:
        print(f"Plan de ejecución: {len(grupos)} grupos")
        for i, grupo in enumerate(grupos, 1):
            print(f"  Grupo {i}: {grupo} ({'paralelo' if len(grupo) > 1 else 'secuencial'})")
    
    for i, grupo in enumerate(grupos, 1):
        if verbose:
            print(f"\n[Grupo {i}] Ejecutando: {grupo}")
        
        async def ejecutar_subtarea(nombre: str) -> tuple[str, str]:
            config = configs_map[nombre]
            # Construir context con resultados disponibles
            context = {**resultados}
            context["input"] = resultados.get(config.dependencias[-1], input_inicial) if config.dependencias else input_inicial
            
            prompt = config.prompt_template.format(**context)
            response = await client_async.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                temperature=0,
                max_tokens=400
            )
            return nombre, response.choices[0].message.content
        
        # Ejecutar grupo en paralelo
        tareas_async = [ejecutar_subtarea(nombre) for nombre in grupo]
        nuevos_resultados = await asyncio.gather(*tareas_async)
        resultados.update(dict(nuevos_resultados))
        
        if verbose:
            for nombre, resultado in nuevos_resultados:
                print(f"  ✓ {nombre}: {resultado[:80]}...")
    
    return {k: v for k, v in resultados.items() if not k.startswith("_")}


# Ejemplo de uso con DAG:
sub_tareas_analisis = [
    SubTareaConfig(
        nombre="extraccion_entidades",
        prompt_template="Extrae personas, organizaciones y fechas de: {input}",
        dependencias=[]  # No depende de nada
    ),
    SubTareaConfig(
        nombre="resumen",
        prompt_template="Resume en 3 oraciones: {input}",
        dependencias=[]  # No depende de nada (paralela a extraccion_entidades)
    ),
    SubTareaConfig(
        nombre="analisis_sesgo",
        prompt_template="Analiza el sesgo de este texto considerando las entidades: {extraccion_entidades}\n\nTexto: {input}",
        dependencias=["extraccion_entidades"]  # Necesita las entidades
    ),
    SubTareaConfig(
        nombre="informe_final",
        prompt_template="Genera un informe basado en: Resumen: {resumen} | Sesgo: {analisis_sesgo} | Entidades: {extraccion_entidades}",
        dependencias=["resumen", "analisis_sesgo", "extraccion_entidades"]  # Necesita todo
    )
]

# asyncio.run(ejecutar_con_paralelizacion(sub_tareas_analisis, "El texto a analizar..."))

Diseño de Interfaces entre Prompts

El Principio de Interface Contracts

from pydantic import BaseModel, field_validator
from typing import Literal

class InterfaceExtraccion(BaseModel):
    """
    Contrato de interface entre el prompt de extracción y el de análisis.
    Cada campo tiene un tipo explícito y validaciones.
    """
    entidades: list[str]
    fechas: list[str]
    numeros_clave: list[float]
    sentimiento: Literal["positivo", "negativo", "neutral", "mixto"]
    
    @field_validator("numeros_clave", mode="before")
    @classmethod
    def limpiar_numeros(cls, v):
        """Normalizar números: remover comas, porcentajes, etc."""
        resultado = []
        for item in v:
            if isinstance(item, (int, float)):
                resultado.append(float(item))
            elif isinstance(item, str):
                cleaned = item.replace(",", "").replace("%", "").strip()
                try:
                    resultado.append(float(cleaned))
                except ValueError:
                    pass
        return resultado

class InterfaceAnalisis(BaseModel):
    """
    Contrato de interface entre análisis y generación de informe.
    """
    conclusiones: list[str]
    nivel_riesgo: Literal["bajo", "medio", "alto", "crítico"]
    acciones_requeridas: list[str]
    datos_clave: dict[str, float]

def prompt_con_interface_contract(
    texto: str,
    schema: type[BaseModel]
) -> BaseModel:
    """
    Ejecuta un prompt y valida el output contra un schema de interface.
    Reintenta si el formato es incorrecto.
    """
    schema_json = schema.model_json_schema()
    
    prompt = f"""Analiza el siguiente texto y responde en JSON según este schema:
{json.dumps(schema_json, indent=2)}

Texto: {texto}

IMPORTANTE: El JSON debe ser válido y contener todos los campos requeridos."""
    
    for intento in range(3):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            response_format={"type": "json_object"}
        ).choices[0].message.content
        
        try:
            datos = json.loads(response)
            return schema(**datos)
        except Exception as e:
            if intento < 2:
                prompt += f"\n\nError en intento anterior: {e}. Por favor corrige."
            else:
                raise ValueError(f"No se pudo parsear el output después de 3 intentos: {e}")

# Ejemplo:
texto_noticia = "Apple reportó ingresos de $124.3 millones el 15 de enero de 2026. Tim Cook destacó el crecimiento del 23% en servicios."
interface_result = prompt_con_interface_contract(texto_noticia, InterfaceExtraccion)
print(f"Entidades: {interface_result.entidades}")
print(f"Números: {interface_result.numeros_clave}")
print(f"Sentimiento: {interface_result.sentimiento}")

Troubleshooting de Decomposition

Problema 1: Sub-tareas demasiado grandes

Síntoma: Una sub-tarea tiene múltiples outputs distintos y el prompt es demasiado complejo.

Señal de alerta: Un prompt que usa más de 5 "extrae/genera/analiza" en la misma instrucción.

Solución:

# ❌ SUB-TAREA DEMASIADO GRANDE
prompt_gigante = """Analiza este contrato.
Extrae todas las partes involucradas.
Identifica las cláusulas de riesgo.
Calcula las penalizaciones.
Compara con estándares del sector.
Genera recomendaciones.
Traduce el resumen al inglés."""

# ✓ DIVIDIR EN 5 SUB-TAREAS ESPECIALIZADAS
prompts_especializados = {
    "partes": "Identifica todas las partes (personas y organizaciones) en este contrato: {input}",
    "clausulas_riesgo": "Lista las cláusulas que representan riesgo legal o financiero: {input}",
    "penalizaciones": "Extrae y calcula las penalizaciones mencionadas en el contrato: {input}",
    "comparacion": "Compara estas cláusulas {clausulas_riesgo} con estándares estándar del sector",
    "recomendaciones": "Genera 5 recomendaciones basadas en el análisis de riesgo: {clausulas_riesgo}"
}

Problema 2: Interfaces ambiguas

Síntoma: La etapa 2 no sabe exactamente qué formato esperar de la etapa 1.

Solución:

# ❌ INTERFACE AMBIGUA
prompt_vago = "Extrae los datos importantes del texto."

# ✓ INTERFACE CON SCHEMA EXPLÍCITO
prompt_especifico = """Extrae del texto:
1. Fecha (formato: DD/MM/YYYY)
2. Importe total (solo número, sin símbolos)
3. Partes (lista de nombres propios)
4. Tipo de documento (contrato/factura/carta/otro)

Responde en JSON:
{{"fecha": "DD/MM/YYYY o null", "importe": number o null, "partes": [str], "tipo": str}}"""

Problema 3: Pérdida de contexto entre sub-tareas

Síntoma: La etapa 3 no tiene suficiente contexto del problema original porque solo recibe el output de la etapa 2.

Solución: Pasar el input original más los outputs intermedios relevantes:

def prompt_con_contexto_completo(
    input_original: str,
    outputs_previos: dict,
    template_actual: str
) -> str:
    """
    Construye un prompt que incluye contexto del problema original
    y los outputs relevantes de etapas anteriores.
    """
    context = f"CONTEXTO ORIGINAL: {input_original[:500]}\n\n"
    
    for nombre, output in outputs_previos.items():
        context += f"RESULTADO DE {nombre.upper()}:\n{output[:200]}\n\n"
    
    return context + template_actual.format(input=outputs_previos.get("ultimo", input_original))

Ejercicios

Ejercicio 1: Decomposition de análisis de datos financieros

Tienes que analizar un reporte anual de una empresa. Descompón esta tarea en sub-tareas, define las interfaces, y especifica las dependencias.

Ver solución 
sub_tareas_financieras = [
    SubTareaConfig(
        nombre="extraccion_metricas",
        prompt_template="""Extrae los siguientes números del reporte financiero:
- Revenue total
- EBITDA
- Net income
- Deuda neta
- Empleados
Responde en JSON: {{"revenue": float, "ebitda": float, "net_income": float, "deuda_neta": float, "empleados": int}}
Reporte: {input}""",
        dependencias=[]
    ),
    SubTareaConfig(
        nombre="extraccion_narrativa",
        prompt_template="Extrae los 5 puntos clave de la narrativa del CEO y el mensaje estratégico: {input}",
        dependencias=[]  # Paralelo a extraccion_metricas
    ),
    SubTareaConfig(
        nombre="calculo_ratios",
        prompt_template="""Calcula estos ratios financieros con los datos:
{extraccion_metricas}
- Margen EBITDA (EBITDA/Revenue)
- ROE si hay equity disponible
- Apalancamiento (Deuda/EBITDA)
Responde en JSON con valores y interpretación.""",
        dependencias=["extraccion_metricas"]
    ),
    SubTareaConfig(
        nombre="sintesis_estrategica",
        prompt_template="""Genera un análisis estratégico usando:
Métricas: {extraccion_metricas}
Ratios: {calculo_ratios}
Narrativa: {extraccion_narrativa}
Incluye: salud financiera, riesgos, oportunidades.""",
        dependencias=["extraccion_metricas", "calculo_ratios", "extraccion_narrativa"]
    )
]

Ejercicio 2: Implementar Map-Reduce para FAQ

Tienes 20 preguntas frecuentes de clientes. Usa Map-Reduce para:

  1. (Map) Clasificar cada pregunta por categoría
  2. (Reduce) Generar una respuesta consolidada por categoría
Ver solución 
faqs = [
    "¿Cómo cambio mi contraseña?",
    "¿Por qué no puedo iniciar sesión?",
    "¿Dónde está mi pedido?",
    "¿Cuánto tarda el envío?",
    "¿Cómo devuelvo un producto?",
    "¿Tienen aplicación móvil?",
    "¿Cómo cancelo mi suscripción?",
]

prompt_map = """Clasifica esta pregunta y extrae la intención:
Pregunta: {documento}
JSON: {{"categoria": "cuenta|pedidos|devoluciones|app|suscripcion|otro", "intencion": str}}"""

prompt_reduce = """Tienes estas clasificaciones de FAQs:
{resultados}

Genera:
1. Resumen por categoría (N preguntas por categoría)
2. Las 3 categorías más frecuentes
3. Gaps de información que podrían convertirse en FAQs adicionales"""

resultado = map_reduce_documentos(faqs, prompt_map, prompt_reduce)
print(resultado)

Ejercicio 3: Diseñar interfaces con Pydantic

Para un pipeline de análisis de código Python, diseña los schemas Pydantic para las interfaces entre:

  1. Análisis de complejidad
  2. Detección de code smells
  3. Sugerencias de refactoring
Ver solución 
from pydantic import BaseModel, Field
from typing import Literal

class AnalisisComplejidad(BaseModel):
    complejidad_ciclomatica: int = Field(ge=1, description="Complejidad ciclomática de la función")
    lineas_codigo: int
    funciones: list[str]
    nivel: Literal["simple", "moderado", "complejo", "muy_complejo"]

class CodeSmell(BaseModel):
    tipo: str
    descripcion: str
    linea: Optional[int] = None
    severidad: Literal["baja", "media", "alta"]

class AnalisisCodeSmells(BaseModel):
    smells: list[CodeSmell]
    deuda_tecnica_estimada_horas: float
    codigo_limpio_score: float = Field(ge=0, le=10)

class SugerenciaRefactoring(BaseModel):
    titulo: str
    descripcion: str
    impacto: Literal["bajo", "medio", "alto"]
    esfuerzo: Literal["bajo", "medio", "alto"]
    codigo_sugerido: Optional[str] = None

class PlanRefactoring(BaseModel):
    sugerencias: list[SugerenciaRefactoring]
    prioridad_recomendada: list[str]  # Nombres de sugerencias en orden
    beneficio_esperado: str

Resumen

  • Decomposition framework: Identificar sub-tareas → definir interfaces → diseñar prompts → ordenar con dependencias
  • Patterns: ETL para procesamiento, Map-Reduce para colecciones, DAG para dependencias complejas
  • Interfaces: Usar Pydantic para schemas de interface. Hace el pipeline robusto y validable.
  • Paralelización: Calcular el orden topológico del DAG y ejecutar grupos en paralelo con asyncio
  • Sub-tareas bien diseñadas: Una responsabilidad clara, schema de input/output explícito, < 300 tokens de prompt

Recursos adicionales

  1. Least-to-Most Prompting (Zhou et al., 2022)
  2. Pydantic v2 - Field validation
  3. Python asyncio - gather for parallel tasks
  4. LangChain Chains documentation
  5. OpenAI Cookbook - Structured outputs
  6. Directed Acyclic Graphs (topological sort)
← Anterior2. Prompt Chaining: Output → InputSiguiente →4. Pipelines Multi-Etapa