6. Multi-Step Reasoning Pipelines

Descripción

Para problemas complejos, un solo prompt CoT no es suficiente. Los pipelines multi-step descomponen el razonamiento en etapas explícitas: Understand → Plan → Execute → Verify. Cada etapa es una llamada separada, y el output de cada etapa alimenta la siguiente.

Esta cápsula cubre: el patrón de 4 etapas, cuándo usar pipeline vs. single CoT, manejo de errores entre etapas, optimización de costos, y casos de uso reales.

Tiempo estimado: 90-120 minutos

¿Por Qué Pipelines Multi-Step?

Single CoT pide al modelo hacer todo en una llamada: entender el problema, planificar la solución, ejecutarla y verificarla. Para problemas simples, eso funciona. Para problemas complejos:

Problemas con single CoT en problemas complejos:

1. El modelo puede "olvidar" parte del problema al llegar al final (limitación de atención)
2. Los errores en la comprensión inicial se propagan a todos los pasos
3. No puedes inspeccionar ni intervenir en pasos intermedios
4. El razonamiento mezclado en un blob es difícil de auditar

Ventajas del pipeline:

1. Cada etapa tiene un propósito claro y verificable
2. Puedes inspeccionar, loggear y corregir cada etapa
3. El contexto se puede resumir entre etapas para no perder información
4. Las etapas pueden ejecutarse con diferentes parámetros (temperatura, modelo, etc.)
5. Puedes cachear etapas costosas

El Pipeline de 4 Etapas: UPEV

U - Understand (Comprender): ¿Qué se pregunta? ¿Qué datos hay?
P - Plan (Planificar): ¿Qué pasos hay que seguir?
E - Execute (Ejecutar): Ejecutar cada paso del plan
V - Verify (Verificar): ¿La respuesta es correcta?

Implementación Completa

from openai import OpenAI
from dataclasses import dataclass, field

client = OpenAI()


@dataclass
class EtapaPipeline:
    nombre: str
    output: str = ""
    tokens: int = 0
    error: str | None = None


@dataclass
class ResultadoPipeline:
    problema: str
    etapas: list[EtapaPipeline] = field(default_factory=list)
    respuesta_final: str = ""
    tokens_total: int = 0
    exito: bool = False


def pipeline_razonamiento(
    problema: str,
    verbose: bool = True
) -> ResultadoPipeline:
    """
    Pipeline de 4 etapas: Understand → Plan → Execute → Verify
    
    Args:
        problema: El problema a resolver
        verbose: Si True, imprime el output de cada etapa
    
    Returns:
        ResultadoPipeline con todos los outputs y metadatos
    """
    resultado = ResultadoPipeline(problema=problema)
    
    # ===== ETAPA 1: UNDERSTAND =====
    prompt_understand = f"""Analiza el siguiente problema y extrae su estructura.

Problema: {problema}

Responde en este formato exacto:
PREGUNTA CENTRAL: [qué se pide exactamente en una oración]
DATOS DADOS: [lista de datos/números/hechos proporcionados]
RESTRICCIONES: [limitaciones o condiciones que debe cumplir la solución]
TIPO DE PROBLEMA: [matemático/lógico/optimización/otro]
DIFICULTAD ESTIMADA: [simple/moderado/complejo]"""

    try:
        r_understand = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt_understand}],
            temperature=0,
            max_tokens=400
        )
        etapa_understand = EtapaPipeline(
            nombre="understand",
            output=r_understand.choices[0].message.content,
            tokens=r_understand.usage.total_tokens
        )
        resultado.etapas.append(etapa_understand)
        
        if verbose:
            print(f"\n{'='*50}")
            print("ETAPA 1: UNDERSTAND")
            print(etapa_understand.output)
    
    except Exception as e:
        resultado.etapas.append(EtapaPipeline(nombre="understand", error=str(e)))
        resultado.exito = False
        return resultado
    
    # ===== ETAPA 2: PLAN =====
    prompt_plan = f"""Dado el análisis del problema, propón un plan detallado de resolución.

Problema original: {problema}

Análisis:
{etapa_understand.output}

Crea un plan numerado con pasos concretos y accionables.
Cada paso debe ser específico (no "calcular algo", sino "calcular X usando Y").
Incluye qué herramienta o técnica usar en cada paso.

PLAN DE RESOLUCIÓN:"""

    try:
        r_plan = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt_plan}],
            temperature=0,
            max_tokens=500
        )
        etapa_plan = EtapaPipeline(
            nombre="plan",
            output=r_plan.choices[0].message.content,
            tokens=r_plan.usage.total_tokens
        )
        resultado.etapas.append(etapa_plan)
        
        if verbose:
            print(f"\n{'='*50}")
            print("ETAPA 2: PLAN")
            print(etapa_plan.output)
    
    except Exception as e:
        resultado.etapas.append(EtapaPipeline(nombre="plan", error=str(e)))
        resultado.exito = False
        return resultado
    
    # ===== ETAPA 3: EXECUTE =====
    prompt_execute = f"""Ejecuta el plan paso a paso para resolver el problema.

Problema: {problema}

Plan a ejecutar:
{etapa_plan.output}

Instrucciones:
- Ejecuta CADA paso del plan
- Muestra el trabajo para cada paso (operaciones, razonamiento)
- Si un paso produce un resultado intermedio, anótalo claramente
- No saltes pasos

EJECUCIÓN:"""

    try:
        r_execute = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt_execute}],
            temperature=0,
            max_tokens=900
        )
        etapa_execute = EtapaPipeline(
            nombre="execute",
            output=r_execute.choices[0].message.content,
            tokens=r_execute.usage.total_tokens
        )
        resultado.etapas.append(etapa_execute)
        
        if verbose:
            print(f"\n{'='*50}")
            print("ETAPA 3: EXECUTE")
            print(etapa_execute.output)
    
    except Exception as e:
        resultado.etapas.append(EtapaPipeline(nombre="execute", error=str(e)))
        resultado.exito = False
        return resultado
    
    # ===== ETAPA 4: VERIFY =====
    prompt_verify = f"""Verifica la solución obtenida.

Problema original: {problema}

Solución obtenida:
{etapa_execute.output}

Verifica:
1. ¿La solución responde exactamente a lo que se preguntó?
2. ¿Los cálculos son correctos? (verifica los pasos clave)
3. ¿La respuesta tiene sentido en el contexto del problema?
4. Si hay errores, corrígelos.

VERIFICACIÓN:
Estado: CORRECTO / INCORRECTO / PARCIALMENTE CORRECTO
Notas de verificación: [detalles]
RESPUESTA FINAL: [la respuesta corregida o confirmada, solo el resultado]"""

    try:
        r_verify = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt_verify}],
            temperature=0,
            max_tokens=400
        )
        etapa_verify = EtapaPipeline(
            nombre="verify",
            output=r_verify.choices[0].message.content,
            tokens=r_verify.usage.total_tokens
        )
        resultado.etapas.append(etapa_verify)
        
        if verbose:
            print(f"\n{'='*50}")
            print("ETAPA 4: VERIFY")
            print(etapa_verify.output)
        
        # Extraer respuesta final
        import re
        match = re.search(
            r'RESPUESTA FINAL:\s*(.+?)(?:\n|$)', 
            etapa_verify.output, 
            re.IGNORECASE
        )
        resultado.respuesta_final = match.group(1).strip() if match else etapa_verify.output[-200:]
        resultado.exito = True
    
    except Exception as e:
        resultado.etapas.append(EtapaPipeline(nombre="verify", error=str(e)))
        resultado.exito = False
    
    # Calcular tokens totales
    resultado.tokens_total = sum(e.tokens for e in resultado.etapas)
    
    if verbose:
        print(f"\n{'='*50}")
        print(f"RESPUESTA FINAL: {resultado.respuesta_final}")
        print(f"Tokens totales: {resultado.tokens_total}")
    
    return resultado


# Ejemplo de uso
if __name__ == "__main__":
    problema_complejo = """
    Una empresa de logística tiene 3 rutas disponibles:
    - Ruta A: 200 km, $0.50/km, tiempo estimado 3h
    - Ruta B: 150 km, $0.70/km, tiempo estimado 2h
    - Ruta C: 250 km, $0.40/km, tiempo estimado 4h
    
    El cliente quiere minimizar costos, pero el tiempo máximo es 3.5 horas.
    Las rutas B y C están disponibles, pero A tiene un cargo adicional de $20 por peaje.
    ¿Cuál es la ruta óptima y cuánto costará?
    """
    
    resultado = pipeline_razonamiento(problema_complejo)
    print(f"\n\nRESUMEN: {resultado.respuesta_final}")

Cuándo Usar Pipeline vs. Single CoT

Criterios de Decisión

Árbol de Decisión

def decidir_estrategia(problema: str) -> str:
    """
    Heurística simple para decidir qué estrategia usar.
    En producción, podrías usar el LLM para esta clasificación.
    """
    indicadores_pipeline = [
        len(problema) > 500,  # Problema largo
        problema.count('\n') > 5,  # Múltiples líneas de datos
        any(w in problema.lower() for w in ['primero', 'luego', 'después', 'finalmente', 'pasos']),
        any(w in problema.lower() for w in ['optimiz', 'comparar', 'analizar y', 'diseñar']),
    ]
    
    if sum(indicadores_pipeline) >= 2:
        return "pipeline"
    else:
        return "single_cot"


# Casos de prueba
ejemplos = [
    "¿Cuánto es 17 × 23?",
    "Analiza este contrato legal de 50 páginas, identifica cláusulas problemáticas, propón mejoras y genera un resumen ejecutivo",
    "¿Es válido este argumento lógico: Si P entonces Q, P, por lo tanto Q?",
    "Diseña un sistema de recomendaciones para un e-commerce: primero analiza los requisitos, luego propón la arquitectura técnica, después detalla el modelo de datos, y finalmente describe cómo medir el éxito",
]

for ej in ejemplos:
    estrategia = decidir_estrategia(ej)
    print(f"{'Pipeline' if estrategia == 'pipeline' else 'Single CoT':12} | {ej[:70]}...")

Pipeline con Manejo de Errores y Recuperación

from typing import Callable


def pipeline_robusto(
    problema: str,
    max_reintentos: int = 2
) -> ResultadoPipeline:
    """
    Pipeline con manejo de errores y reintentos por etapa.
    Si una etapa falla, intenta recuperarse antes de fallar globalmente.
    """
    
    etapas_config = [
        {
            "nombre": "understand",
            "prompt_fn": lambda p, prev: f"Analiza el problema:\n{p}\nExtrae: pregunta central, datos, restricciones.",
            "max_tokens": 400,
            "temperatura": 0
        },
        {
            "nombre": "plan",
            "prompt_fn": lambda p, prev: f"Problema: {p}\nAnálisis: {prev}\nCrea un plan de resolución paso a paso.",
            "max_tokens": 500,
            "temperatura": 0
        },
        {
            "nombre": "execute",
            "prompt_fn": lambda p, prev: f"Ejecuta este plan para resolver: {p}\nPlan:\n{prev}",
            "max_tokens": 900,
            "temperatura": 0
        },
        {
            "nombre": "verify",
            "prompt_fn": lambda p, prev: f"Verifica esta solución. Problema: {p}\nSolución: {prev}\nRESPUESTA FINAL:",
            "max_tokens": 400,
            "temperatura": 0
        }
    ]
    
    resultado = ResultadoPipeline(problema=problema)
    output_anterior = ""
    
    for config in etapas_config:
        nombre = config["nombre"]
        exito_etapa = False
        
        for intento in range(max_reintentos + 1):
            try:
                prompt = config["prompt_fn"](problema, output_anterior)
                
                # En reintentos, añadir contexto del error
                if intento > 0:
                    prompt += f"\n\n[Intento {intento + 1}: el intento anterior no dio un resultado completo. Por favor sé más detallado.]"
                
                response = client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{"role": "user", "content": prompt}],
                    temperature=config["temperatura"],
                    max_tokens=config["max_tokens"]
                )
                
                etapa = EtapaPipeline(
                    nombre=nombre,
                    output=response.choices[0].message.content,
                    tokens=response.usage.total_tokens
                )
                resultado.etapas.append(etapa)
                output_anterior = etapa.output
                exito_etapa = True
                break
                
            except Exception as e:
                if intento == max_reintentos:
                    etapa = EtapaPipeline(nombre=nombre, error=str(e))
                    resultado.etapas.append(etapa)
                    # Continuar con lo que tenemos (graceful degradation)
                    output_anterior = f"[Error en etapa {nombre}: {str(e)[:100]}]"
        
        if not exito_etapa and nombre in ["execute"]:
            # Las etapas críticas requieren éxito para continuar
            resultado.exito = False
            resultado.tokens_total = sum(e.tokens for e in resultado.etapas)
            return resultado
    
    # Extraer respuesta final
    import re
    if resultado.etapas:
        ultima_etapa = resultado.etapas[-1]
        match = re.search(r'RESPUESTA FINAL:\s*(.+?)(?:\n|$)', ultima_etapa.output, re.IGNORECASE)
        resultado.respuesta_final = match.group(1).strip() if match else ultima_etapa.output[-200:]
    
    resultado.exito = True
    resultado.tokens_total = sum(e.tokens for e in resultado.etapas)
    return resultado

Pipeline con Resumen de Contexto

Para problemas muy largos, el contexto entre etapas puede crecer excesivamente. Esta implementación resume el output antes de pasarlo a la siguiente etapa:

def resumir_etapa(output_largo: str, max_chars: int = 500) -> str:
    """Condensa el output de una etapa para la siguiente."""
    if len(output_largo) <= max_chars:
        return output_largo
    
    prompt_resumen = f"""Resume el siguiente texto manteniendo SOLO los puntos clave necesarios para la siguiente etapa de resolución del problema. Máximo {max_chars} caracteres.

Texto a resumir:
{output_largo}

Resumen:"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt_resumen}],
        temperature=0,
        max_tokens=200
    )
    return response.choices[0].message.content[:max_chars]


def pipeline_con_contexto_resumido(problema: str) -> ResultadoPipeline:
    """
    Pipeline que resume el contexto entre etapas para evitar token overflow.
    Útil para problemas con outputs muy largos en las etapas intermedias.
    """
    resultado = pipeline_razonamiento(problema, verbose=False)
    return resultado

Pipelines Especializados por Dominio

Pipeline: Análisis de Código Complejo

def pipeline_code_review(codigo: str, lenguaje: str = "Python") -> dict:
    """
    Pipeline especializado para análisis exhaustivo de código.
    4 etapas: Entender → Identificar problemas → Sugerir mejoras → Generar código mejorado
    """
    
    # Etapa 1: Entender el código
    r1 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Analiza el siguiente código {lenguaje}:
```{lenguaje.lower()}
{codigo}

¿Qué hace este código? Describe su propósito, inputs, outputs y estructura en 5-7 oraciones. """}], temperature=0, max_tokens=300 ) descripcion = r1.choices[0].message.content

# Etapa 2: Identificar problemas
r2 = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": f"""

Código:

{codigo}

Descripción: {descripcion}

Identifica TODOS los problemas (bugs, edge cases, rendimiento, seguridad, legibilidad). Para cada problema: tipo, línea, descripción, severidad (Alta/Media/Baja). """}], temperature=0, max_tokens=600 ) problemas = r2.choices[0].message.content

# Etapa 3: Sugerir mejoras
r3 = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": f"""

Código:

{codigo}

Problemas identificados: {problemas}

Para cada problema de severidad Alta o Media, propón la corrección específica. Format: "Problema X → Corrección: [código corregido]" """}], temperature=0, max_tokens=700 ) mejoras = r3.choices[0].message.content

# Etapa 4: Generar código mejorado
r4 = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": f"""

Código original:

{codigo}

Mejoras a aplicar: {mejoras}

Genera el código mejorado aplicando TODAS las correcciones de Alta y Media severidad. Solo el código, sin explicaciones adicionales.

        """}],
        temperature=0, max_tokens=800
    )
    codigo_mejorado = r4.choices[0].message.content
    
    return {
        "descripcion": descripcion,
        "problemas": problemas,
        "mejoras_sugeridas": mejoras,
        "codigo_mejorado": codigo_mejorado,
        "tokens_total": sum([
            r1.usage.total_tokens, r2.usage.total_tokens,
            r3.usage.total_tokens, r4.usage.total_tokens
        ])
    }

Pipeline: Investigación y Síntesis

def pipeline_investigacion(pregunta: str, contexto_disponible: str) -> dict:
    """
    Pipeline para responder preguntas complejas que requieren análisis profundo.
    Útil en sistemas RAG donde hay múltiples documentos de contexto.
    """
    
    # Etapa 1: Descomponer la pregunta
    r1 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Descompón esta pregunta compleja en sub-preguntas más simples:
Pregunta: {pregunta}

Lista 3-5 sub-preguntas que, al responderse, respondan la pregunta principal.
        """}],
        temperature=0, max_tokens=300
    )
    subpreguntas = r1.choices[0].message.content
    
    # Etapa 2: Buscar evidencia en el contexto
    r2 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Para cada sub-pregunta, encuentra la evidencia relevante en el contexto dado.

Sub-preguntas:
{subpreguntas}

Contexto disponible:
{contexto_disponible}

Para cada sub-pregunta, cita la parte relevante del contexto o indica "Sin evidencia".
        """}],
        temperature=0, max_tokens=700
    )
    evidencia = r2.choices[0].message.content
    
    # Etapa 3: Sintetizar respuesta
    r3 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Sintetiza la información para responder la pregunta principal.

Pregunta original: {pregunta}

Evidencia encontrada: {evidencia}

Genera una respuesta coherente y bien estructurada que:
1. Responda directamente la pregunta
2. Esté basada en la evidencia
3. Indique claramente si hay información insuficiente
        """}],
        temperature=0, max_tokens=600
    )
    sintesis = r3.choices[0].message.content
    
    # Etapa 4: Validar y añadir confianza
    r4 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Evalúa la calidad de esta respuesta:

Pregunta: {pregunta}
Respuesta: {sintesis}
Evidencia usada: {evidencia}

Evalúa:
1. ¿La respuesta está completamente respaldada por la evidencia?
2. ¿Hay afirmaciones sin evidencia?
3. Confianza general: Alta/Media/Baja + razón

RESPUESTA VALIDADA: [la respuesta con modificaciones si aplica]
CONFIANZA: [nivel]
        """}],
        temperature=0, max_tokens=400
    )
    validacion = r4.choices[0].message.content
    
    return {
        "pregunta": pregunta,
        "subpreguntas": subpreguntas,
        "evidencia": evidencia,
        "sintesis": sintesis,
        "validacion": validacion
    }

Optimización de Costos en Pipelines

from functools import lru_cache
import hashlib


def hash_texto(texto: str) -> str:
    """Genera hash del texto para caching."""
    return hashlib.md5(texto.encode()).hexdigest()


class PipelineConCache:
    """
    Pipeline que cachea resultados intermedios para evitar re-procesar
    el mismo problema o partes idénticas.
    """
    
    def __init__(self):
        self._cache: dict[str, str] = {}
    
    def _llamada_con_cache(self, prompt: str, **kwargs) -> tuple[str, bool]:
        """
        Hace una llamada al LLM, usando caché si el prompt ya fue procesado.
        
        Returns:
            tuple de (respuesta, desde_cache)
        """
        cache_key = hash_texto(prompt)
        
        if cache_key in self._cache:
            return self._cache[cache_key], True
        
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            **kwargs
        )
        respuesta = response.choices[0].message.content
        self._cache[cache_key] = respuesta
        return respuesta, False
    
    def pipeline_con_cache(self, problema: str) -> dict:
        """Pipeline con caching de etapas."""
        
        prompt_understand = f"Analiza: {problema}\nExtrae pregunta, datos, restricciones."
        entendimiento, desde_cache = self._llamada_con_cache(
            prompt_understand, temperature=0, max_tokens=400
        )
        
        prompt_plan = f"Problema: {problema}\nAnálisis: {entendimiento}\nCrea plan de resolución."
        plan, _ = self._llamada_con_cache(prompt_plan, temperature=0, max_tokens=500)
        
        prompt_execute = f"Ejecuta este plan:\nProblema: {problema}\nPlan: {plan}"
        ejecucion, _ = self._llamada_con_cache(prompt_execute, temperature=0, max_tokens=900)
        
        prompt_verify = f"Verifica. Problema: {problema}\nSolución: {ejecucion}\nRESPUESTA FINAL:"
        verificacion, _ = self._llamada_con_cache(prompt_verify, temperature=0, max_tokens=400)
        
        return {
            "entendimiento": entendimiento,
            "plan": plan,
            "ejecucion": ejecucion,
            "verificacion": verificacion,
            "understand_desde_cache": desde_cache
        }

Medición de Rendimiento: Pipeline vs. Single CoT

import time
from statistics import mean


def benchmarkar_enfoques(
    problemas: list[str],
    n_runs: int = 3
) -> dict:
    """
    Compara rendimiento y accuracy de single CoT vs. pipeline.
    """
    
    tiempos_single = []
    tiempos_pipeline = []
    tokens_single = []
    tokens_pipeline = []
    
    for problema in problemas[:n_runs]:
        # Single CoT
        t_inicio = time.time()
        r_single = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"{problema}\n\nPiensa paso a paso."}],
            temperature=0, max_tokens=700
        )
        tiempos_single.append(time.time() - t_inicio)
        tokens_single.append(r_single.usage.total_tokens)
        
        # Pipeline
        t_inicio = time.time()
        r_pipeline = pipeline_razonamiento(problema, verbose=False)
        tiempos_pipeline.append(time.time() - t_inicio)
        tokens_pipeline.append(r_pipeline.tokens_total)
    
    return {
        "single_cot": {
            "tiempo_promedio_s": mean(tiempos_single),
            "tokens_promedio": mean(tokens_single),
            "costo_relativo": 1.0
        },
        "pipeline": {
            "tiempo_promedio_s": mean(tiempos_pipeline),
            "tokens_promedio": mean(tokens_pipeline),
            "costo_relativo": mean(tokens_pipeline) / mean(tokens_single)
        }
    }

Tabla: Cuándo Usar Cada Enfoque

Troubleshooting

Problema 1: Error en una etapa intermedia contamina las siguientes

# ❌ Sin manejo de error, un mal output de "plan" genera una ejecución incorrecta
# ✅ Implementar validación por etapa

def validar_etapa_plan(output_plan: str) -> bool:
    """Verifica que el plan tiene la estructura esperada."""
    # Un plan válido debe tener al menos 3 pasos numerados
    import re
    pasos = re.findall(r'^\d+[\.\)]\s', output_plan, re.MULTILINE)
    return len(pasos) >= 2


def pipeline_con_validacion(problema: str) -> ResultadoPipeline:
    """Pipeline que valida cada etapa antes de continuar."""
    resultado = pipeline_razonamiento(problema, verbose=False)
    
    # Verificar si el plan fue generado correctamente
    etapa_plan = next((e for e in resultado.etapas if e.nombre == "plan"), None)
    if etapa_plan and not validar_etapa_plan(etapa_plan.output):
        # Regenerar el plan con instrucciones más explícitas
        print("Plan inválido, regenerando...")
        # ... lógica de reintento
    
    return resultado

Problema 2: Acumulación de tokens entre etapas

# ❌ Pasar todo el output anterior crea prompts cada vez más largos
# ✅ Resumir outputs anteriores

LIMITE_CONTEXTO_CHARS = 800  # Límite razonable por etapa anterior

def pasar_contexto_resumido(output_etapa: str, max_chars: int = LIMITE_CONTEXTO_CHARS) -> str:
    """Prepara el output de una etapa para pasarlo a la siguiente."""
    if len(output_etapa) <= max_chars:
        return output_etapa
    
    # Tomar la primera y última parte (generalmente la más importante)
    mitad = max_chars // 2
    return output_etapa[:mitad] + "\n...[resumen]...\n" + output_etapa[-mitad:]

Problema 3: Latencia alta en pipelines secuenciales

# ❌ Pipeline secuencial: 4 llamadas en serie = alta latencia
# ✅ Paralelizar etapas independientes cuando sea posible

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def pipeline_parcialmente_paralelo(problema: str) -> dict:
    """
    Algunas etapas pueden ejecutarse en paralelo.
    Por ejemplo: "understand" para distintas sub-partes del problema.
    """
    # Etapa 1: Understand (secuencial, prerequisito)
    r_understand = await async_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Analiza: {problema}"}],
        temperature=0, max_tokens=400
    )
    entendimiento = r_understand.choices[0].message.content
    
    # Etapas 2a y 2b: Ejecutar en paralelo si el problema tiene sub-partes
    r_plan, r_check_restricciones = await asyncio.gather(
        async_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"Plan para: {problema}\n{entendimiento}"}],
            temperature=0, max_tokens=400
        ),
        async_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"Lista restricciones de: {problema}"}],
            temperature=0, max_tokens=200
        )
    )
    
    return {
        "entendimiento": entendimiento,
        "plan": r_plan.choices[0].message.content,
        "restricciones": r_check_restricciones.choices[0].message.content
    }


# Uso
async def main():
    resultado = await pipeline_parcialmente_paralelo("Tu problema aquí")
    print(resultado)

# asyncio.run(main())

Ejercicios

Ejercicio 1: Implementar un pipeline para análisis de reseñas

Diseña un pipeline de 3 etapas para analizar reseñas de productos: (1) Extraer aspectos mencionados, (2) Clasificar sentimiento por aspecto, (3) Generar resumen ejecutivo.

from openai import OpenAI

client = OpenAI()

def pipeline_analisis_reseña(reseña: str, producto: str = "") -> dict:
    """Pipeline para análisis de reseñas de productos."""
    
    # Etapa 1: Extraer aspectos
    r1 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Analiza esta reseña y extrae todos los aspectos del producto mencionados.
Producto: {producto or "desconocido"}
Reseña: {reseña}

Lista los aspectos mencionados (ej: calidad, precio, envío, atención al cliente, etc.)
        """}],
        temperature=0, max_tokens=300
    )
    aspectos = r1.choices[0].message.content
    
    # Etapa 2: Sentimiento por aspecto
    r2 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Para cada aspecto identificado, clasifica el sentimiento expresado.

Reseña: {reseña}
Aspectos identificados: {aspectos}

Format: Aspecto: [nombre] | Sentimiento: POSITIVO/NEGATIVO/NEUTRO | Evidencia: "[cita]"
        """}],
        temperature=0, max_tokens=500
    )
    sentimientos = r2.choices[0].message.content
    
    # Etapa 3: Resumen ejecutivo
    r3 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Genera un resumen ejecutivo de 3-4 líneas para un gerente de producto.

Reseña analizada: {reseña}
Análisis de sentimientos: {sentimientos}

El resumen debe incluir: sentimiento general, puntos más fuertes, puntos más débiles, 
y una recomendación accionable.
        """}],
        temperature=0, max_tokens=300
    )
    
    return {
        "aspectos": aspectos,
        "sentimientos_por_aspecto": sentimientos,
        "resumen_ejecutivo": r3.choices[0].message.content
    }


# Prueba
reseña_ejemplo = """
Compré este aspirador hace 3 meses. La succión es increíble, mejor que el que tenía antes.
Sin embargo, el cable es demasiado corto para habitaciones grandes y tuve que comprar una 
extensión. El envío fue rapidísimo (llegó al día siguiente) pero la caja llegaba golpeada.
El servicio al cliente cuando llamé fue muy amable y resolvió mis dudas. El precio me parece 
justo para la calidad que ofrece.
"""

resultado = pipeline_analisis_reseña(reseña_ejemplo, "Aspirador X200")
for etapa, contenido in resultado.items():
    print(f"\n=== {etapa.upper()} ===")
    print(contenido)

Ejercicio 2: Pipeline para research con RAG simplificado

Diseña un pipeline que (1) descomponga una pregunta en sub-preguntas, (2) "busque" en un corpus local, (3) sintetice la respuesta.

CORPUS_EJEMPLO = {
    "python": "Python es un lenguaje interpretado, de alto nivel, con tipado dinámico. Creado por Guido van Rossum en 1991. Muy usado en ciencia de datos, ML, y backend.",
    "fastapi": "FastAPI es un framework web moderno para Python 3.8+. Basado en Starlette y Pydantic. Genera documentación OpenAPI automáticamente. Muy alto rendimiento.",
    "openai": "OpenAI fue fundada en 2015. Desarrolla GPT-4, DALL-E, y Whisper. La API de OpenAI permite acceso a modelos de lenguaje mediante HTTP.",
}

def buscar_en_corpus(query: str, corpus: dict) -> str:
    """Búsqueda simple por keywords en el corpus."""
    resultados = []
    query_lower = query.lower()
    for clave, contenido in corpus.items():
        if clave in query_lower or any(w in contenido.lower() for w in query_lower.split()):
            resultados.append(f"[{clave}]: {contenido}")
    return "\n".join(resultados) if resultados else "Sin resultados relevantes"

def pipeline_research(pregunta: str, corpus: dict) -> dict:
    # Etapa 1: Descomponer pregunta
    r1 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Descompón en 2-3 sub-preguntas: {pregunta}"}],
        temperature=0, max_tokens=200
    )
    subpreguntas = r1.choices[0].message.content
    
    # Etapa 2: Buscar contexto
    contexto = buscar_en_corpus(pregunta, corpus)
    
    # Etapa 3: Sintetizar
    r3 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Pregunta: {pregunta}
Sub-preguntas: {subpreguntas}
Contexto disponible: {contexto}

Responde usando SOLO la información del contexto.
        """}],
        temperature=0, max_tokens=400
    )
    
    return {
        "pregunta": pregunta,
        "subpreguntas": subpreguntas,
        "contexto_usado": contexto,
        "respuesta": r3.choices[0].message.content
    }

resultado = pipeline_research("¿Qué ventajas tiene FastAPI sobre otros frameworks Python?", CORPUS_EJEMPLO)
print(resultado["respuesta"])

Ejercicio 3: Medir el overhead del pipeline vs. single CoT

Diseña un experimento para medir cuántos tokens y tiempo adicionales requiere el pipeline comparado con single CoT para 5 problemas idénticos.

import time
from openai import OpenAI

client = OpenAI()

PROBLEMAS_BENCHMARK = [
    "Si tengo 3 pizzas de 8 porciones y invito a 10 personas, ¿cuántas porciones recibe cada uno?",
    "Un banco ofrece 5% de interés anual. Si invierto $1,000 por 3 años con interés compuesto, ¿cuánto tengo al final?",
    "En un torneo de 8 equipos con eliminación directa, ¿cuántos partidos hay en total?",
    "Si un tren sale a las 9:00 y va a 80 km/h, y otro sale a las 9:30 en la misma ruta a 120 km/h, ¿cuándo alcanza al primero?",
    "¿Cuántos números enteros del 1 al 100 son divisibles por 3 o por 5?",
]

resultados = []

for i, problema in enumerate(PROBLEMAS_BENCHMARK, 1):
    print(f"\nProblema {i}: {problema[:50]}...")
    
    # Single CoT
    t0 = time.time()
    r_single = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"{problema}\n\nPiensa paso a paso."}],
        temperature=0, max_tokens=600
    )
    tiempo_single = time.time() - t0
    tokens_single = r_single.usage.total_tokens
    
    # Pipeline
    t0 = time.time()
    r_pipeline = pipeline_razonamiento(problema, verbose=False)
    tiempo_pipeline = time.time() - t0
    tokens_pipeline = r_pipeline.tokens_total
    
    resultados.append({
        "problema": i,
        "tokens_single": tokens_single,
        "tokens_pipeline": tokens_pipeline,
        "multiplicador_tokens": tokens_pipeline / tokens_single,
        "tiempo_single_s": tiempo_single,
        "tiempo_pipeline_s": tiempo_pipeline,
        "multiplicador_tiempo": tiempo_pipeline / tiempo_single
    })

print("\n=== RESUMEN BENCHMARK ===")
avg_tokens = sum(r["multiplicador_tokens"] for r in resultados) / len(resultados)
avg_tiempo = sum(r["multiplicador_tiempo"] for r in resultados) / len(resultados)
print(f"Tokens promedio (Pipeline vs Single): {avg_tokens:.1f}x más")
print(f"Tiempo promedio (Pipeline vs Single): {avg_tiempo:.1f}x más")

Resumen

• Pipeline UPEV: Understand → Plan → Execute → Verify; cada etapa es una llamada separada
• Cuándo usar: Problemas complejos con múltiples sub-tareas, necesidad de inspección/auditoría
• Manejo de errores: Validar cada etapa, tener estrategias de reintento y graceful degradation
• Contexto: Resumir outputs largos antes de pasarlos a la siguiente etapa
• Costo: 3-5x más tokens que single CoT; justificado para problemas críticos
• Paralelismo: Etapas independientes pueden ejecutarse en paralelo con asyncio
• Caching: Cachear etapas costosas que se repiten entre llamadas

Recursos adicionales

1. ReAct: Synergizing Reasoning and Acting (Yao et al., 2022)
2. Least-to-Most Prompting (Zhou et al., 2022)
3. Decomposed Prompting (Khot et al., 2022)
4. LangChain Chains Documentation
5. OpenAI Assistants API - Multi-step tasks
6. Async OpenAI Python SDK