Módulo 6: Prompt Composition y Chaining

2. Prompt Chaining: Output → Input

Siguiente →

Descripción

Prompt chaining es el patrón fundamental de la composición de prompts: el output de un prompt se convierte en el input del siguiente. Esta técnica permite construir pipelines donde cada etapa es un prompt especializado que transforma la información de forma incremental.

En esta cápsula aprenderás los tres patrones principales de chaining: sequential (A→B→C), conditional (bifurcación según el contenido), y parallel (múltiples prompts simultáneos), junto con estrategias robustas de error handling.

Por Qué Prompt Chaining

Un único prompt tiene limitaciones:

❌ Un solo prompt para análisis complejo:
   "Lee este contrato, identifica las partes, extrae cláusulas de riesgo, 
    calcula si los plazos son razonables, sugiere modificaciones, y genera
    un resumen ejecutivo en 3 idiomas."
   
   Resultado: Respuesta superficial que intenta hacer todo a medias.

✓ Chaining especializado:
   Prompt 1: Extraer partes y estructura → datos limpios
   Prompt 2: Identificar cláusulas de riesgo → análisis profundo
   Prompt 3: Evaluar plazos → evaluación focalizada
   Prompt 4: Sugerir modificaciones → basado en análisis real
   Prompt 5: Resumen en 3 idiomas → traducción de síntesis final
   
   Resultado: Análisis detallado en cada dimensión.

Patrón 1: Sequential Chaining

Implementación Base

from openai import OpenAI
from typing import Optional, Callable, Any
import json

client = OpenAI()

def llamar_llm(
    prompt: str,
    temperatura: float = 0,
    max_tokens: int = 500,
    json_mode: bool = False
) -> str:
    """Helper básico para llamar al LLM."""
    kwargs = {
        "model": "gpt-4o-mini",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": temperatura,
        "max_tokens": max_tokens
    }
    if json_mode:
        kwargs["response_format"] = {"type": "json_object"}
    
    return client.chat.completions.create(**kwargs).choices[0].message.content

def chain_sequential(
    etapas: list[dict],
    input_inicial: str,
    verbose: bool = True
) -> dict:
    """
    Ejecuta una cadena de prompts secuencial.
    
    Args:
        etapas: Lista de {nombre: str, prompt_template: str, temperatura: float}
                El prompt_template usa {input} para recibir el output anterior
        input_inicial: El input de la primera etapa
        verbose: Si True, imprime el proceso
    
    Returns:
        dict con todos los outputs por nombre de etapa
    """
    outputs = {"_input_inicial": input_inicial}
    input_actual = input_inicial
    
    for etapa in etapas:
        nombre = etapa["nombre"]
        template = etapa["prompt_template"]
        temperatura = etapa.get("temperatura", 0)
        json_mode = etapa.get("json_mode", False)
        max_tokens = etapa.get("max_tokens", 500)
        
        if verbose:
            print(f"[Chain] Ejecutando etapa: {nombre}")
        
        # Formatear el template con el input actual
        prompt = template.format(
            input=input_actual,
            **{k: v for k, v in outputs.items() if not k.startswith("_")}
        )
        
        output = llamar_llm(prompt, temperatura, max_tokens, json_mode)
        outputs[nombre] = output
        input_actual = output  # El output se convierte en el input de la siguiente
        
        if verbose:
            print(f"  Output ({len(output)} chars): {output[:100]}...")
    
    return outputs

# Ejemplo: Pipeline de análisis de reseña de producto
etapas_resena = [
    {
        "nombre": "extraccion",
        "prompt_template": """Extrae de esta reseña de producto: 
1. Aspectos mencionados (calidad, precio, envío, etc.)
2. Opinión sobre cada aspecto (positiva/negativa/neutral)
3. Calificación implícita (1-5)

Reseña: {input}

Responde en JSON: {{"aspectos": [{{"nombre": str, "opinion": str, "calificacion": 1-5}}], "calificacion_general": 1-5}}""",
        "json_mode": True,
        "temperatura": 0
    },
    {
        "nombre": "analisis",
        "prompt_template": """Basándote en este análisis de una reseña de producto:
{input}

Identifica:
1. ¿Es una reseña genuina o podría ser fake? (señales de autenticidad)
2. ¿Qué aspectos del producto necesitan mejora urgente?
3. ¿Qué fortalezas destacar en el marketing?

Responde de forma concisa.""",
        "temperatura": 0
    },
    {
        "nombre": "accion",
        "prompt_template": """Basándote en este análisis:
{input}

Genera 3 acciones concretas que el equipo de producto debería tomar.
Cada acción: quién la hace, qué hace, en cuánto tiempo.
Formato: bullet points.""",
        "temperatura": 0
    }
]

resena = "El producto llegó en perfectas condiciones y funciona exactamente como se describe. El precio es un poco alto pero la calidad lo justifica. El envío tardó más de lo esperado (5 días en vez de 2), pero el vendedor fue muy atento y me mantuvo informado."

resultado = chain_sequential(etapas_resena, resena, verbose=True)
print("\n=== OUTPUT FINAL (ACCIONES) ===")
print(resultado["accion"])

Chain con Transformación de Formato

from pydantic import BaseModel
from typing import Union

class ExtraccionResena(BaseModel):
    aspectos: list[dict]
    calificacion_general: int

def chain_con_validacion(
    texto_resena: str
) -> dict:
    """
    Chain con validación de schema entre etapas usando Pydantic.
    """
    # Etapa 1: Extracción estructurada
    prompt_extraccion = f"""Extrae de esta reseña:
- aspectos mencionados con su opinión
- calificación general implícita (1-5)

Reseña: {texto_resena}

Responde en JSON: {{"aspectos": [{{"nombre": str, "opinion": "positiva|negativa|neutral", "calificacion": 1-5}}], "calificacion_general": int}}"""
    
    output_json = llamar_llm(prompt_extraccion, json_mode=True)
    
    # Validar con Pydantic
    try:
        extraccion = ExtraccionResena(**json.loads(output_json))
    except Exception as e:
        print(f"Error de validación en etapa 1: {e}. Reintentando...")
        output_json = llamar_llm(
            prompt_extraccion + "\n\nIMPORTANTE: La respuesta DEBE ser JSON válido con 'aspectos' y 'calificacion_general'.",
            json_mode=True
        )
        extraccion = ExtraccionResena(**json.loads(output_json))
    
    # Etapa 2: Análisis basado en extracción validada
    aspectos_str = json.dumps([a for a in extraccion.aspectos], ensure_ascii=False)
    prompt_analisis = f"""Analiza estos aspectos de una reseña de producto:
{aspectos_str}
Calificación general: {extraccion.calificacion_general}/5

¿Cuáles son los 2 puntos más críticos que necesitan mejora?"""
    
    analisis = llamar_llm(prompt_analisis)
    
    return {
        "extraccion": extraccion.model_dump(),
        "analisis": analisis
    }

Patrón 2: Conditional Chaining

from enum import Enum

class TipoDocumento(Enum):
    CONTRATO = "contrato"
    FACTURA = "factura"
    CORREO = "correo"
    INFORME = "informe"
    OTRO = "otro"

def clasificar_documento(texto: str) -> TipoDocumento:
    """Primer paso del conditional chain: clasificar el tipo de documento."""
    prompt = f"""Clasifica este texto en una de estas categorías:
CONTRATO, FACTURA, CORREO, INFORME, OTRO

Texto (primeros 500 chars):
{texto[:500]}

Responde SOLO con una palabra: CONTRATO, FACTURA, CORREO, INFORME, o OTRO"""
    
    clasificacion = llamar_llm(prompt).strip().upper()
    
    mapping = {
        "CONTRATO": TipoDocumento.CONTRATO,
        "FACTURA": TipoDocumento.FACTURA,
        "CORREO": TipoDocumento.CORREO,
        "INFORME": TipoDocumento.INFORME
    }
    return mapping.get(clasificacion, TipoDocumento.OTRO)

# Prompts especializados por tipo de documento
PROMPTS_ESPECIALIZADOS = {
    TipoDocumento.CONTRATO: """Analiza este contrato legal. Identifica:
1. Partes involucradas
2. Objeto del contrato
3. Obligaciones principales de cada parte
4. Cláusulas de rescisión
5. Penalizaciones si las hay

Contrato: {input}""",
    
    TipoDocumento.FACTURA: """Extrae la siguiente información de esta factura:
1. Emisor y receptor
2. Número de factura y fecha
3. Líneas de factura (concepto, cantidad, precio unitario, total)
4. Base imponible, IVA, total

Factura: {input}

Responde en JSON estructurado.""",
    
    TipoDocumento.CORREO: """Analiza este email de negocios:
1. Remitente y destinatario
2. Propósito principal
3. Acción requerida (si la hay)
4. Tono y urgencia
5. Respuesta sugerida

Email: {input}""",
    
    TipoDocumento.INFORME: """Analiza este informe:
1. Tema principal
2. Métricas o datos clave
3. Conclusiones principales
4. Recomendaciones si las hay

Informe: {input}""",
    
    TipoDocumento.OTRO: """Resume el siguiente documento:
- Tipo de contenido
- Puntos principales
- Información relevante

Documento: {input}"""
}

def chain_conditional_documento(texto: str, verbose: bool = True) -> dict:
    """
    Chain condicional: clasifica el documento y luego aplica
    el análisis especializado para ese tipo.
    """
    # Etapa 1: Clasificación (siempre se ejecuta)
    tipo = clasificar_documento(texto)
    
    if verbose:
        print(f"[Conditional Chain] Tipo detectado: {tipo.value}")
    
    # Etapa 2: Análisis especializado (condicional según tipo)
    prompt_especializado = PROMPTS_ESPECIALIZADOS[tipo].format(input=texto)
    analisis = llamar_llm(
        prompt_especializado,
        json_mode=(tipo == TipoDocumento.FACTURA)
    )
    
    # Etapa 3: Acciones recomendadas (siempre se ejecuta pero con contexto del tipo)
    prompt_acciones = f"""Dado este análisis de un documento tipo {tipo.value}:
{analisis}

¿Qué 2-3 acciones concretas debería tomar el receptor de este documento?"""
    
    acciones = llamar_llm(prompt_acciones)
    
    return {
        "tipo_documento": tipo.value,
        "analisis_especializado": analisis,
        "acciones_recomendadas": acciones
    }

# Ejemplo:
documentos = [
    "Estimado cliente, adjunto encontrará la factura #2026-0342 por servicios prestados en febrero...",
    "Por el presente contrato, PARTE A se compromete a entregar los servicios descritos en el Anexo I...",
]

for doc in documentos:
    resultado = chain_conditional_documento(doc, verbose=True)
    print(f"  Análisis: {resultado['analisis_especializado'][:100]}...\n")

Patrón 3: Parallel Chaining

import asyncio
from openai import AsyncOpenAI

async def chain_parallel_analisis(
    articulo: str,
    analisis_solicitados: list[str] = None
) -> dict:
    """
    Ejecuta múltiples análisis en paralelo sobre el mismo artículo.
    Reduce la latencia total significativamente.
    
    Args:
        articulo: Texto del artículo a analizar
        analisis_solicitados: Lista de análisis deseados. Default: todos.
    
    Returns:
        dict con todos los análisis completados
    """
    client_async = AsyncOpenAI()
    
    if analisis_solicitados is None:
        analisis_solicitados = ["resumen", "keywords", "sentimiento", "entidades"]
    
    PROMPTS_PARALELOS = {
        "resumen": f"Resume en 3 oraciones concisas: {articulo}",
        
        "keywords": f"""Extrae 5-8 keywords del siguiente artículo.
Formato: Lista separada por comas.
Artículo: {articulo}""",
        
        "sentimiento": f"""Analiza el sentimiento y tono de este artículo.
Indica: sentimiento (POSITIVO/NEGATIVO/NEUTRAL), tono (formal/informal/técnico), y emotividad (alta/media/baja).
Artículo: {articulo}""",
        
        "entidades": f"""Extrae entidades nombradas de este artículo.
Categorías: personas, organizaciones, lugares, fechas/periodos, productos.
Artículo: {articulo}
Formato JSON.""",
        
        "preguntas": f"""¿Cuáles son las 3 preguntas más importantes que este artículo responde?
Artículo: {articulo}""",
        
        "audiencia": f"""¿A quién está dirigido principalmente este artículo?
Describe la audiencia objetivo y por qué.
Artículo: {articulo}"""
    }
    
    async def ejecutar_analisis(nombre: str) -> tuple[str, str]:
        """Ejecuta un análisis individual."""
        if nombre not in PROMPTS_PARALELOS:
            return nombre, f"Análisis '{nombre}' no disponible"
        
        response = await client_async.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": PROMPTS_PARALELOS[nombre]}],
            temperature=0,
            max_tokens=300
        )
        return nombre, response.choices[0].message.content
    
    # Ejecutar todos en paralelo
    import time
    t0 = time.time()
    
    tareas = [ejecutar_analisis(nombre) for nombre in analisis_solicitados]
    resultados_raw = await asyncio.gather(*tareas)
    
    tiempo_total = time.time() - t0
    resultados = dict(resultados_raw)
    resultados["_metadata"] = {
        "tiempo_total": tiempo_total,
        "n_analisis": len(analisis_solicitados),
        "tiempo_promedio_si_secuencial": tiempo_total * len(analisis_solicitados)
    }
    
    return resultados

# Para correr en script normal (no notebook):
async def ejemplo_parallel():
    articulo = """La inteligencia artificial está transformando la industria financiera. 
    Los algoritmos de machine learning permiten detectar fraudes en tiempo real 
    con una precisión del 99.5%, reduciendo las pérdidas en un 60%..."""
    
    import time
    t0 = time.time()
    resultados = await chain_parallel_analisis(
        articulo,
        analisis_solicitados=["resumen", "keywords", "sentimiento", "entidades"]
    )
    t_total = time.time() - t0
    
    print(f"Tiempo total: {t_total:.2f}s (vs ~{t_total * 4:.1f}s si fuera secuencial)")
    for nombre, resultado in resultados.items():
        if not nombre.startswith("_"):
            print(f"\n{nombre.upper()}:\n{resultado[:150]}...")

# asyncio.run(ejemplo_parallel())  # Descomentar para ejecutar

Patrón 4: Fan-Out / Fan-In

async def chain_fan_out_fan_in(
    documento: str,
    n_perspectivas: int = 3
) -> dict:
    """
    Fan-Out: Analiza el documento desde múltiples perspectivas en paralelo.
    Fan-In: Sintetiza todos los análisis en un output coherente.
    
    Útil para análisis multi-dimensional donde cada perspectiva
    puede revelar aspectos diferentes del mismo contenido.
    """
    client_async = AsyncOpenAI()
    
    perspectivas = [
        ("técnica", "Analiza los aspectos técnicos, precisión y metodología"),
        ("negocios", "Analiza las implicaciones para negocios y ROI"),
        ("usuario_final", "Analiza el impacto para el usuario final"),
        ("ética", "Analiza consideraciones éticas y de privacidad"),
        ("competitivo", "Analiza en contexto del panorama competitivo")
    ][:n_perspectivas]
    
    async def analizar_perspectiva(nombre: str, instruccion: str) -> tuple[str, str]:
        response = await client_async.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"""Analiza este documento desde la perspectiva {nombre}:
{instruccion}

Documento: {documento[:3000]}

Proporciona 3 puntos clave desde esta perspectiva."""
            }],
            temperature=0,
            max_tokens=300
        )
        return nombre, response.choices[0].message.content
    
    # Fan-Out: múltiples análisis en paralelo
    tareas = [analizar_perspectiva(n, i) for n, i in perspectivas]
    resultados_perspectivas = dict(await asyncio.gather(*tareas))
    
    # Fan-In: sintetizar todos los análisis
    perspectivas_str = "\n\n".join([
        f"PERSPECTIVA {nombre.upper()}:\n{analisis}"
        for nombre, analisis in resultados_perspectivas.items()
    ])
    
    sintesis = client_async  # Reusar client async
    sintesis_resp = await sintesis.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"""Tienes {n_perspectivas} análisis diferentes del mismo documento:

{perspectivas_str}

Sintetiza una evaluación holística de 3-5 párrafos que integre todas las perspectivas.
Destaca: consensos entre perspectivas, tensiones o contradicciones, y conclusión global."""
        }],
        temperature=0,
        max_tokens=600
    )
    
    return {
        "perspectivas": resultados_perspectivas,
        "sintesis": sintesis_resp.choices[0].message.content
    }

Error Handling Robusto en Chains

from typing import Optional
import logging

logger = logging.getLogger(__name__)

class ErrorEtapa(Exception):
    """Error específico de una etapa del chain."""
    def __init__(self, etapa: str, mensaje: str, input_data: str = ""):
        self.etapa = etapa
        self.mensaje = mensaje
        self.input_data = input_data
        super().__init__(f"Error en etapa '{etapa}': {mensaje}")

def chain_con_manejo_errores(
    etapas: list[dict],
    input_inicial: str,
    max_retries: int = 2,
    fallback_output: Optional[dict] = None
) -> dict:
    """
    Sequential chain con manejo robusto de errores.
    
    Features:
    - Retry automático por etapa
    - Prompt mejorado en retry
    - Fallback si se agotan los retries
    - Logging de errores
    """
    outputs = {"_input_inicial": input_inicial}
    input_actual = input_inicial
    
    for etapa in etapas:
        nombre = etapa["nombre"]
        template = etapa["prompt_template"]
        temperatura = etapa.get("temperatura", 0)
        json_mode = etapa.get("json_mode", False)
        validador = etapa.get("validador", None)  # Función opcional de validación
        
        ultimo_error = None
        
        for intento in range(max_retries + 1):
            try:
                prompt = template.format(
                    input=input_actual,
                    **{k: v for k, v in outputs.items() if not k.startswith("_")}
                )
                
                # En retries, hacer el prompt más explícito
                if intento > 0:
                    prompt += f"\n\nIMPORTANTE: Intento {intento+1}. Error anterior: {ultimo_error}. Por favor responde en el formato exacto solicitado."
                
                output = llamar_llm(prompt, temperatura, 600, json_mode)
                
                # Validación opcional
                if validador:
                    valido, error_validacion = validador(output)
                    if not valido:
                        raise ValueError(f"Validación falló: {error_validacion}")
                
                outputs[nombre] = output
                input_actual = output
                break  # Éxito, salir del loop de retries
                
            except Exception as e:
                ultimo_error = str(e)
                logger.warning(f"Intento {intento+1} fallido en etapa '{nombre}': {e}")
                
                if intento == max_retries:
                    # Agotar retries
                    if fallback_output and nombre in fallback_output:
                        outputs[nombre] = fallback_output[nombre]
                        input_actual = fallback_output[nombre]
                        outputs[f"_{nombre}_falló"] = True
                        logger.error(f"Usando fallback para etapa '{nombre}'")
                    else:
                        raise ErrorEtapa(nombre, ultimo_error, input_actual)
    
    return outputs

# Validador de ejemplo para JSON
def validar_json_con_campos(campos_requeridos: list[str]):
    """Genera un validador que verifica que el JSON tiene los campos requeridos."""
    def validar(output: str) -> tuple[bool, str]:
        try:
            datos = json.loads(output)
            for campo in campos_requeridos:
                if campo not in datos:
                    return False, f"Falta el campo '{campo}' en el JSON"
            return True, ""
        except json.JSONDecodeError as e:
            return False, f"JSON inválido: {e}"
    return validar

# Uso:
etapas_con_validacion = [
    {
        "nombre": "extraccion",
        "prompt_template": "Extrae las entidades de: {input}\nJSON: {{\"personas\": [], \"organizaciones\": [], \"fechas\": []}}",
        "json_mode": True,
        "temperatura": 0,
        "validador": validar_json_con_campos(["personas", "organizaciones"])
    }
]

try:
    resultado = chain_con_manejo_errores(
        etapas_con_validacion,
        "Apple lanzó el iPhone 17 el 15 de marzo de 2026. Tim Cook presentó el dispositivo.",
        max_retries=2
    )
    print(resultado)
except ErrorEtapa as e:
    print(f"Error no recuperable en etapa '{e.etapa}': {e.mensaje}")

Casos de Uso Reales

Caso 1: Pipeline de Procesamiento de CVs

def pipeline_cv(texto_cv: str) -> dict:
    """Pipeline completo para analizar un CV."""
    etapas = [
        {
            "nombre": "extraccion",
            "prompt_template": """Extrae del siguiente CV:
1. Nombre y datos de contacto
2. Experiencia laboral (empresa, rol, duración)
3. Educación
4. Skills técnicas
5. Idiomas

CV: {input}

Responde en JSON estructurado.""",
            "json_mode": True
        },
        {
            "nombre": "evaluacion",
            "prompt_template": """Basándote en esta extracción de CV:
{input}

Evalúa:
1. Años de experiencia total
2. Nivel técnico estimado (junior/mid/senior)
3. Área de especialización principal
4. Fortalezas (3 puntos)
5. Áreas de mejora (2 puntos)

Sé objetivo y basado en evidencia del CV."""
        },
        {
            "nombre": "preguntas_entrevista",
            "prompt_template": """Dado este análisis de CV:
{input}

Genera 5 preguntas de entrevista específicas para este candidato.
Incluye: 2 técnicas, 2 de experiencia, 1 de situación difícil.
Cada pregunta en formato: "Pregunta: [pregunta] (Objetivo: [qué evalúa])"
"""
        }
    ]
    
    return chain_con_manejo_errores(etapas, texto_cv)

# Test:
cv_ejemplo = """
Juan García
Backend Developer | juan@email.com | LinkedIn: /in/jgarcia
5+ años de experiencia en Python, FastAPI, PostgreSQL.
Trabajo en TechCorp (2021-presente) como Lead Backend Developer...
"""
resultado = pipeline_cv(cv_ejemplo)
print("EVALUACIÓN:", resultado["evaluacion"][:300])

Caso 2: Análisis Paralelo de Competidores

async def analisis_competidores(empresa: str, competidores: list[str]) -> dict:
    """
    Analiza múltiples competidores en paralelo y compara con la empresa principal.
    """
    client_async = AsyncOpenAI()
    
    async def analizar_competidor(nombre_competidor: str) -> tuple[str, str]:
        response = await client_async.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"""Analiza brevemente a {nombre_competidor} como competidor de {empresa}:
1. Fortalezas principales (2-3 puntos)
2. Debilidades (2 puntos)
3. Diferenciadores vs {empresa}
(Usa conocimiento general; indica si algo es incierto)"""
            }],
            temperature=0,
            max_tokens=250
        )
        return nombre_competidor, response.choices[0].message.content
    
    # Fan-out: analizar todos en paralelo
    tareas = [analizar_competidor(c) for c in competidores]
    analisis_individuales = dict(await asyncio.gather(*tareas))
    
    # Fan-in: síntesis comparativa
    comparaciones = "\n\n".join([f"### {comp}:\n{analisis}" 
                                  for comp, analisis in analisis_individuales.items()])
    
    sintesis_resp = await client_async.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"""Análisis de competidores de {empresa}:

{comparaciones}

Genera:
1. Tabla de comparación con fortalezas/debilidades
2. Principales amenazas competitivas
3. Oportunidades de diferenciación para {empresa}"""
        }],
        temperature=0,
        max_tokens=600
    )
    
    return {
        "analisis_individuales": analisis_individuales,
        "sintesis_comparativa": sintesis_resp.choices[0].message.content
    }

# asyncio.run(analisis_competidores("FastAPI", ["Django", "Flask", "Express.js"]))

Troubleshooting

Problema 1: Output truncado entre etapas

Síntoma: El output de la etapa 1 está cortado y la etapa 2 trabaja con información incompleta.

Causas:

  • max_tokens demasiado bajo
  • El modelo genera más tokens de los esperados

Solución:

def chain_con_verificacion_truncamiento(
    template: str,
    input_data: str,
    max_tokens: int = 500
) -> str:
    """Detecta y maneja outputs truncados."""
    output = llamar_llm(template.format(input=input_data), max_tokens=max_tokens)
    
    # Señales de truncamiento
    señales_truncamiento = [
        output.endswith("..."),
        output.endswith(","),
        len(output.split()) > max_tokens * 0.9,
        output.count("{") != output.count("}"),  # JSON desbalanceado
    ]
    
    if any(señales_truncamiento):
        # Intentar con más tokens o solicitar completar
        output_completo = llamar_llm(
            f"{output}\n\n[CONTINÚA desde donde te cortaste]",
            max_tokens=max_tokens * 2
        )
        return output + output_completo
    
    return output

Problema 2: Formato inconsistente entre etapas

Síntoma: La etapa 2 espera JSON pero la etapa 1 devuelve texto libre.

Solución: Usar json_mode=True cuando necesites JSON, y validar con Pydantic:

def etapa_con_json_garantizado(prompt: str) -> dict:
    """Garantiza que el output es JSON válido."""
    max_intentos = 3
    for intento in range(max_intentos):
        output = llamar_llm(
            prompt if intento == 0 else f"{prompt}\n\nCRÍTICO: Responde SOLO JSON válido.",
            json_mode=True
        )
        try:
            return json.loads(output)
        except json.JSONDecodeError:
            if intento == max_intentos - 1:
                return {"error": "No se pudo parsear JSON", "raw": output}
    return {}

Problema 3: Cascada de errores

Síntoma: Un error en la etapa 2 hace que la etapa 3 falle también, y así sucesivamente.

Solución: Cada etapa debe tener un fallback independiente:

def etapa_con_fallback(
    template: str,
    input_data: str,
    fallback_message: str = "Análisis no disponible para esta etapa"
) -> str:
    """Etapa individual con fallback."""
    try:
        return llamar_llm(template.format(input=input_data))
    except Exception as e:
        print(f"⚠ Etapa con error: {e}. Usando fallback.")
        return fallback_message

Ejercicios

Ejercicio 1: Chain de análisis de feedback

Implementa un chain de 3 etapas para analizar feedback de usuarios:

  1. Categorizar el feedback (bug, feature request, queja, cumplido)
  2. Evaluar prioridad (alta/media/baja) según categoría y urgencia
  3. Generar respuesta automática apropiada al tipo
Ver solución 
def chain_feedback_usuario(feedback: str) -> dict:
    etapas = [
        {
            "nombre": "categorizacion",
            "prompt_template": """Categoriza este feedback de usuario:
Categorías: BUG, FEATURE_REQUEST, QUEJA, CUMPLIDO, PREGUNTA

Feedback: {input}

Responde en JSON: {{"categoria": str, "subcategoria": str, "urgencia": "alta|media|baja", "sentimiento": "positivo|negativo|neutral"}}""",
            "json_mode": True
        },
        {
            "nombre": "prioridad",
            "prompt_template": """Dado este análisis de feedback:
{input}

Determina:
1. Prioridad de atención (1-5, donde 5 es más urgente)
2. Equipo responsable (engineering, product, support, marketing)
3. Tiempo de respuesta sugerido (inmediato/24h/72h/semana)

JSON: {{"prioridad": int, "equipo": str, "tiempo_respuesta": str, "razon": str}}""",
            "json_mode": True
        },
        {
            "nombre": "respuesta_automatica",
            "prompt_template": """Feedback original: {_input_inicial}
Análisis: {prioridad}

Genera una respuesta automática empática y profesional para este feedback.
- Si es BUG: Agradece el reporte, indica que se investigará
- Si es FEATURE_REQUEST: Agradece la sugerencia, indica que se considerará
- Si es QUEJA: Empatiza, pide disculpas si aplica, indica acción
- Si es CUMPLIDO: Agradece genuinamente
Máximo 3 oraciones."""
        }
    ]
    
    return chain_con_manejo_errores(etapas, feedback, max_retries=1)

feedback_test = "Llevo 3 días intentando exportar mis datos a CSV y el botón no funciona. Es urgente porque tengo una presentación mañana y necesito esos datos."
resultado = chain_feedback_usuario(feedback_test)
print("Respuesta:", resultado["respuesta_automatica"])

Ejercicio 2: Chain con ramificación por complejidad

Implementa un chain que:

  1. Analiza la complejidad de una pregunta técnica (simple/compleja)
  2. Si es simple: responde directamente
  3. Si es compleja: descompone en sub-preguntas, responde cada una, sintetiza
Ver solución 
def chain_pregunta_adaptativo(pregunta: str) -> dict:
    # Paso 1: Evaluar complejidad
    clasif = llamar_llm(f"¿Esta pregunta es SIMPLE (respuesta directa) o COMPLEJA (requiere múltiples pasos)?\nPregunta: {pregunta}\nResponde solo: SIMPLE o COMPLEJA")
    
    if "SIMPLE" in clasif.upper():
        respuesta = llamar_llm(f"Responde de forma concisa: {pregunta}")
        return {"complejidad": "simple", "respuesta": respuesta}
    
    # Paso 2 (solo para complejas): Descomponer
    sub_preguntas_raw = llamar_llm(f"Descompón esta pregunta compleja en 2-4 sub-preguntas más simples:\n{pregunta}\nFormato: '1. ...' '2. ...'")
    
    sub_preguntas = [l.lstrip('0123456789.-) ').strip() 
                     for l in sub_preguntas_raw.split('\n') 
                     if l.strip() and l.strip()[0].isdigit()]
    
    # Paso 3: Responder cada sub-pregunta
    respuestas_parciales = []
    for sp in sub_preguntas[:4]:
        resp = llamar_llm(f"Responde brevemente: {sp}")
        respuestas_parciales.append(f"Q: {sp}\nA: {resp}")
    
    # Paso 4: Sintetizar
    contexto = "\n\n".join(respuestas_parciales)
    sintesis = llamar_llm(f"Pregunta original: {pregunta}\n\nRespuestas parciales:\n{contexto}\n\nSintetiza una respuesta completa.")
    
    return {
        "complejidad": "compleja",
        "sub_preguntas": sub_preguntas,
        "respuestas_parciales": respuestas_parciales,
        "sintesis": sintesis
    }

Ejercicio 3: Pipeline paralelo de análisis de contenido

Implementa un pipeline async que analice un artículo en paralelo para: resumen, keywords, audiencia objetivo, y clasificación temática. El output debe estar en un JSON estructurado.

Ver solución 
async def pipeline_analisis_articulo(articulo: str) -> dict:
    client_async = AsyncOpenAI()
    
    ANALISIS = {
        "resumen": f"Genera un resumen ejecutivo de 2 oraciones: {articulo[:2000]}",
        "keywords": f"Lista 5 keywords principales separadas por comas: {articulo[:2000]}",
        "audiencia": f"¿A quién está dirigido este artículo? (2-3 oraciones): {articulo[:2000]}",
        "tema": f"Clasifica en UN tema: tecnología, negocios, ciencia, política, cultura, deportes, otro. Solo una palabra: {articulo[:500]}"
    }
    
    async def ejecutar(nombre: str, prompt: str) -> tuple[str, str]:
        r = await client_async.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0, max_tokens=200
        )
        return nombre, r.choices[0].message.content.strip()
    
    resultados = dict(await asyncio.gather(*[ejecutar(n, p) for n, p in ANALISIS.items()]))
    return resultados

# Para ejecutar:
# resultado = asyncio.run(pipeline_analisis_articulo("El texto del artículo..."))

Resumen

  • Sequential chaining: A→B→C, el output de cada etapa alimenta la siguiente. Simple pero eficaz para procesos lineales.
  • Conditional chaining: Bifurcación según el contenido. Permite especialización por tipo de input.
  • Parallel chaining: Múltiples análisis simultáneos con asyncio. Reduce latencia de N*T a ~T.
  • Fan-out/fan-in: Múltiples análisis paralelos + síntesis. Ideal para análisis multi-perspectiva.
  • Error handling: Retry por etapa, validación con Pydantic, fallback, logging. No propagar errores sin contexto.

Recursos adicionales

  1. LangChain Expression Language (LCEL) - Framework para chaining
  2. Python asyncio - gather
  3. OpenAI AsyncOpenAI client
  4. Pydantic v2 Documentation
  5. Prompt Engineering Guide - Prompt Chaining
  6. OpenAI Cookbook - How to build chained prompts
← Anterior1. Introducción: Un Prompt No es SuficienteSiguiente →3. Decomposition de Tareas Complejas