4 mar 2026

Patrones Multi-Agente: Orquestadores, Workers y Pipelines

Los agentes individuales son buenos en tareas bien delimitadas. En el momento en que una tarea requiere conocimiento especializado en distintos dominios, trabajo paralelo sobre entradas grandes, o decisiones que deben validarse antes de ejecutarse, necesitas múltiples agentes.

Los sistemas multi-agente no son intrínsecamente más complejos — simplemente están estructurados de forma diferente. La clave es elegir el patrón correcto para el problema. Tres patrones cubren la mayoría de los casos de uso: orquestador-worker, pipeline y fan-out paralelo.

Por Qué Múltiples Agentes

El caso para los sistemas multi-agente se reduce a tres razones prácticas.

Especialización. Un agente único con 50 herramientas en contexto se confunde. Un agente especializado con 5 herramientas enfocadas rinde mejor. Divide por dominio — investigación, escritura, código, verificación — y cada agente hace una sola cosa bien.

Paralelismo. Algunas tareas se descomponen en subtareas independientes. Analizar 20 documentos secuencialmente es lento; analizarlos de forma concurrente con agentes paralelos es rápido.

Verificación. Tener un agente que produce la salida y un segundo agente que la critica de forma independiente detecta errores que la auto-revisión pasa por alto. El revisor no tiene interés en defender la respuesta original.

Patrón 1: Orquestador-Worker

Un agente orquestador planifica y delega. Los agentes worker ejecutan tareas específicas y devuelven resultados. El orquestador ensambla la salida final.

Este es el patrón más flexible. El orquestador puede adaptar su plan en función de los resultados intermedios, reintentar tareas fallidas o escalar a un worker diferente.

import anthropic
import json

client = anthropic.Anthropic()

def run_worker(system_prompt: str, task: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=system_prompt,
        messages=[{"role": "user", "content": task}]
    )
    return response.content[0].text

def orchestrator(user_request: str) -> str:
    # Paso 1: planificar el trabajo
    plan_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="""Eres un agente de planificación. Dado un pedido del usuario, divídelo en
2-4 subtareas específicas. Devuelve solo un array JSON de descripciones de tareas.
Ejemplo: ["Investigar X", "Analizar Y", "Sintetizar hallazgos"]""",
        messages=[{"role": "user", "content": user_request}]
    )

    tasks = json.loads(plan_response.content[0].text)

    # Paso 2: ejecutar cada tarea con un worker especializado
    results = []
    for task in tasks:
        result = run_worker(
            system_prompt="Eres un agente de ejecución enfocado. Completa la tarea asignada a fondo.",
            task=task
        )
        results.append({"task": task, "result": result})

    # Paso 3: sintetizar
    synthesis_prompt = f"""Solicitud original: {user_request}

Resultados de los workers:
{json.dumps(results, indent=2)}

Sintetiza estos resultados en una respuesta final coherente."""

    final = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": synthesis_prompt}]
    )
    return final.content[0].text

El patrón orquestador funciona mejor cuando la estructura de la tarea no se conoce de antemano. Si no sabes cuántas subtareas necesitarás hasta que hayas analizado el problema, usa un orquestador.

Un error común: los orquestadores pueden alucinar subtareas que no tienen sentido. Restringe el formato de salida (array JSON, lista numerada) y valídalo antes de ejecutar los workers. Un try/except alrededor del parseo JSON con un paso de replanificación como fallback lo maneja graciosamente.

Patrón 2: Pipeline

Los agentes forman una cadena secuencial. Cada agente transforma la entrada y pasa su salida al siguiente. Ningún agente conoce a los demás — reciben entrada y producen salida.

Este es el patrón más simple de implementar y razonar. Funciona bien para tareas de transformación con etapas bien definidas.

def run_pipeline(input_text: str) -> str:
    stages = [
        {
            "name": "Investigador",
            "system": "Extrae y organiza todos los hechos clave de la entrada. "
                      "Formatea como una lista estructurada con fuentes donde estén disponibles.",
        },
        {
            "name": "Escritor",
            "system": "Transforma las notas de investigación en prosa clara y legible. "
                      "Mantén todo el contenido factual. Dirígete a una audiencia técnica.",
        },
        {
            "name": "Editor",
            "system": "Mejora la claridad y concisión. Elimina redundancias. "
                      "No cambies los hechos. Devuelve solo el texto mejorado.",
        },
        {
            "name": "Verificador de Hechos",
            "system": "Revisa la consistencia interna. Señala cualquier afirmación que "
                      "se contradiga o parezca sin respaldo. "
                      "Si no hay problemas, devuelve 'VERIFICADO: ' seguido del texto original.",
        },
    ]

    current = input_text
    for stage in stages:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=stage["system"],
            messages=[{"role": "user", "content": current}]
        )
        current = response.content[0].text
        print(f"[{stage['name']}] completo ({len(current)} chars)")

    return current

Los pipelines acumulan errores. Si el investigador omite algo, el escritor no puede añadirlo de vuelta. Diseña tus etapas para que sean aditivas en lugar de con pérdida — evita etapas que eliminen información que la siguiente etapa pueda necesitar.

Un ajuste práctico: pasa la entrada original junto con la salida de cada etapa cuando los agentes posteriores necesiten contexto que las etapas anteriores puedan haber comprimido.

Patrón 3: Fan-Out Paralelo

Divide una entrada grande en fragmentos independientes, procesa cada uno de forma concurrente con agentes separados y luego agrega los resultados.

Este es el patrón correcto cuando procesas más datos de los que caben cómodamente en una ventana de contexto, o cuando el tiempo de procesamiento importa.

import asyncio

async def analyze_document(doc: str, index: int) -> dict:
    """Analiza un único documento de forma asíncrona."""
    system = """Analiza este documento y devuelve un objeto JSON con:
- "sentiment": positive/negative/neutral
- "key_topics": lista de 3-5 temas principales
- "summary": resumen de 2-3 oraciones
- "flags": lista de preocupaciones (lista vacía si no hay ninguna)"""

    result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            system=system,
            messages=[{"role": "user", "content": doc}]
        ).content[0].text
    )
    return {"index": index, **json.loads(result)}

async def parallel_analysis(documents: list[str]) -> dict:
    # Fan out: analizar todos los documentos de forma concurrente
    tasks = [analyze_document(doc, i) for i, doc in enumerate(documents)]
    analyses = await asyncio.gather(*tasks)

    # Agregar con un agente de síntesis dedicado
    synthesis_input = json.dumps({
        "document_count": len(documents),
        "analyses": analyses
    })

    aggregate_result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system="Sintetiza los análisis de documentos en: distribución general de sentimiento, "
                   "temas principales en todos los documentos y patrones notables en las señales. "
                   "Devuelve como JSON.",
            messages=[{"role": "user", "content": synthesis_input}]
        ).content[0].text
    )

    return {
        "individual": analyses,
        "aggregate": json.loads(aggregate_result)
    }

La etapa de agregación es donde la mayoría de las implementaciones toman atajos. No concatenes resultados — pásalos a un agente que entienda la tarea de agregación. Una unión de cadenas de 20 análisis no es útil; un resumen sintetizado sí lo es.

Elegir el Patrón Correcto

Situación	Patrón
Estructura de tarea desconocida hasta analizar	Orquestador-worker
Etapas de transformación bien definidas	Pipeline
Entrada grande, fragmentos independientes	Fan-out paralelo
Se necesita verificación independiente	Orquestador o pipeline con etapa de revisión
Minimizar latencia en entradas grandes	Fan-out paralelo

Estos patrones se componen. Un sistema real puede usar un orquestador que distribuye algunas tareas en paralelo mientras ejecuta otras a través de un pipeline. Empieza con el patrón más simple que se ajuste y añade complejidad solo cuando el enfoque más simple falle.

Consideraciones Prácticas

Costo. Los sistemas multi-agente multiplican las llamadas a la API. Un pipeline de 4 etapas puede costar 4× una sola llamada más el overhead de síntesis. Mezcla modelos estratégicamente: usa Opus para orquestación y planificación donde importa el juicio, Haiku para tareas de ejecución de alto volumen.

Propagación de errores. Decide de antemano cómo maneja cada agente los fallos. Opciones: propagar el error (detener), devolver un objeto de error (dejar decidir al orquestador) o reintentar con un prompt modificado (se recupera graciosamente, añade latencia). Para la mayoría de los sistemas en producción, devolver objetos de error estructurados y dejar que el orquestador decida es el valor predeterminado correcto.

Trazabilidad. Un sistema multi-agente donde no puedes ver qué hizo cada agente es una pesadilla para depurar. Registra cada llamada a agente con: entrada, salida, modelo, latencia y conteo de tokens. Etiqueta cada llamada con un ID de traza para poder reconstruir la ruta de ejecución completa.

Paso de contexto. Sé deliberado sobre qué contexto recibe cada agente. Pasar el historial completo de conversación a cada agente es costoso y a menudo confuso — los agentes se distraen con contexto previo irrelevante. Pasa solo lo que cada agente necesita para hacer su trabajo específico.

Qué Construir a Continuación

Los patrones aquí son la base. Lo que construyes sobre ellos depende de tu problema:

Añade uso de herramientas a los workers — deja que los agentes especializados llamen APIs, consulten bases de datos o ejecuten código
Añade puntos de control humano-en-el-bucle donde el orquestador se pausa antes de acciones de alto riesgo
Añade memoria persistiendo las salidas de los agentes en un almacén de vectores que los futuros agentes puedan consultar
Añade evaluación enrutando las salidas a través de un agente juez antes de devolverlas al usuario

Los sistemas multi-agente son donde ocurre la ingeniería de IA más interesante en este momento. Los patrones son simples; el juicio está en aplicarlos correctamente a tu problema específico.