Orquestación de Agentes: Buenas Prácticas para Sistemas Multi-Agente

La orquestación de agentes de IA es el arte de coordinar múltiples agentes para resolver tareas complejas en conjunto, de manera confiable, eficiente y trazable. Los agentes individuales raramente son el problema. El verdadero reto está en la coordinación: ¿quién hace qué? ¿en qué orden? ¿qué sucede si algo falla? ¿cómo controlas los costos?

TL;DR — Orquestación de Agentes en 90 Segundos

¿Qué es la Orquestación de Agentes?

La orquestación de agentes es la coordinación de múltiples agentes de IA con:

• División de tareas: ¿Quién hace qué? ¿Qué experiencia necesita cada parte?
• Comunicación: ¿Cómo intercambian información los agentes? ¿A través de estado, mensajes o eventos?
• Control: ¿Quién decide el siguiente paso? ¿Un orquestador central o rutas condicionales?
• Manejo de errores: ¿Qué sucede cuando algo falla? ¿Reintentos, alternativas, intervención humana?
• Gestión de recursos: ¿Cómo optimizar las llamadas a LLM? ¿Caché, selección de modelo, límites de tokens?
• Observabilidad: ¿Cómo rastrear lo que sucede? ¿Logging, tracing, métricas?

Agente Simple vs. Multi-Agente: ¿Cuándo necesitas Orquestación?

Un agente es suficiente cuando: las tareas son lineales, hay una sola área de expertise, no hay iteración necesaria, la tarea se resuelve en una llamada a LLM.

Multi-agente es necesario cuando: hay múltiples áreas de expertise (investigación + redacción + revisión), iteración (redactor ↔ revisor), paralelismo, tareas que superan el límite de contexto, herramientas diversas, intervención humana en el proceso.

Patrones de Orquestación: En Detalle

1. Hub-and-Spoke (Orquestrador Central)

Un agente orquestrador central decide cuándo ejecutar cada agente.

         [Orchestrator]
        /     |      \
[Agent A] [Agent B] [Agent C]

Ventajas: flujo de control claro, manejo de errores simple, totalmente trazable. Desventajas: punto único de fallo, el orquestrador es un cuello de botella, costos adicionales de LLM.

def orchestrator(state):
    task = state["task"]
    # LLM decide qué agente se necesita
    response = llm.invoke(f"¿Qué agente para: {task}? (research/code/write)")
    task_type = response.content.strip().lower()
    return {"task_type": task_type}

def route_to_agent(state):
    return state["task_type"]

# En LangGraph:
workflow.add_conditional_edges("orchestrator", route_to_agent, {
    "research": "research_agent",
    "code": "code_agent",
    "write": "writer_agent"
})

2. Pipeline (Secuencial)

Los agentes trabajan uno tras otro, cada uno pasa su resultado al siguiente. El Process.sequential de CrewAI usa este patrón por defecto.

[Agent A] → [Agent B] → [Agent C] → [Output]

Ventajas: sencillo, predecible, cada agente tiene todos los resultados anteriores. Desventajas: sin paralelismo, un agente fallido bloquea todo.

¿Cuándo usarlo? En flujos lineales sin ramificaciones (investigación → redacción → revisión).

3. DAG (Directed Acyclic Graph)

Los agentes trabajan según dependencias, con paralelismo donde sea posible. Un DAG permite ejecutar tareas independientes en paralelo y coordinar tareas dependientes.

[Agent A] ──→ [Agent C] ──→ [Agent E] (Merge)
     └──→ [Agent B] ──→ [Agent D] ──┘

Ventajas: paralelismo — las tareas independientes se ejecutan simultáneamente. Eficiencia — el tiempo total es el camino más largo. Desventajas: difícil de depurar, las dependencias deben estar bien definidas, posibles race conditions con estado compartido.

Ejemplo con LangGraph (investigadores paralelos):

class ResearchState(TypedDict):
    topic: str
    general_research: str
    technical_research: str
    combined_report: str

def general_researcher(state: ResearchState) -> dict:
    response = llm.invoke(f"Información general sobre: {state['topic']}")
    return {"general_research": response.content}

def technical_researcher(state: ResearchState) -> dict:
    response = llm.invoke(f"Detalles técnicos sobre: {state['topic']}")
    return {"technical_research": response.content}

def merge_results(state: ResearchState) -> dict:
    combined = f"General:\n{state['general_research']}\n\nTécnico:\n{state['technical_research']}"
    return {"combined_report": combined}

# Grafo con nodos paralelos
workflow = StateGraph(ResearchState)
workflow.add_node("general", general_researcher)
workflow.add_node("technical", technical_researcher)
workflow.add_node("merge", merge_results)

workflow.set_entry_point("general")
workflow.add_edge("general", "technical")  # Ambos en paralelo
workflow.add_edge("general", "merge")
workflow.add_edge("technical", "merge")
workflow.add_edge("merge", END)

¿Cuándo usarlo? Cuando tareas independientes pueden ejecutarse en paralelo (por ejemplo, 3 investigaciones que luego se combinan).

4. Event-Driven (Reactivo)

Los agentes reaccionan a eventos en lugar de ser llamados directamente. Un bus de eventos distribuye eventos a todos los agentes interesados.

[Event Bus]
  ├── [Agent A] (escucha "research.done")
  ├── [Agent B] (escucha "code.reviewed")
  └── [Agent C] (escucha "test.failed")

Ventajas: acoplamiento débil, escalable, agentes agregables dinámicamente. Desventajas: difícil de rastrear, posibles race conditions, depuración compleja.

from collections import defaultdict

class EventBus:
    def __init__(self):
        self.subscribers = defaultdict(list)
    
    def subscribe(self, event_type: str, handler):
        self.subscribers[event_type].append(handler)
    
    def publish(self, event_type: str, data: dict):
        results = []
        for handler in self.subscribers[event_type]:
            results.append(handler(data))
        return results

bus = EventBus()
bus.subscribe("research.done", writer_agent)
bus.subscribe("research.done", fact_checker_agent)
bus.subscribe("code.reviewed", refactor_agent)
bus.publish("research.done", {"topic": "AI 2026", "result": "..."})

¿Cuándo usarlo? En sistemas débilmente acoplados, donde los agentes se agregan dinámicamente y el orden no está predeterminado.

Comparación de Patrones

Patrón	Complejidad	Paralelismo	Flexibilidad	Depuración	Mejor para
Pipeline	Baja	❌	Baja	Sencilla	Flujos lineales
Hub-and-Spoke	Media	❌	Media	Media	Enrutamiento dinámico
DAG	Alta	✅	Alta	Difícil	Tareas paralelas
Event-Driven	Alta	✅	Muy alta	Muy difícil	Sistemas débilmente acoplados

Mejores Prácticas — En Detalle

1. Define roles claros (Principio de Responsabilidad Única)

Cada agente debe tener exactamente una responsabilidad. Un agente que investiga, escribe y prueba simultáneamente es difícil de depurar, costoso y produce resultados peores que tres agentes especializados.

Incorrecto:

agent = Agent(
    role="Researcher, Writer and Tester",
    goal="Research, write and test everything"
)

Correcto:

researcher = Agent(role="Researcher", goal="Find information")
writer = Agent(role="Writer", goal="Write content based on research")
tester = Agent(role="Tester", goal="Test functionality and report bugs")

2. Haz los estados explícitos

Usa TypedDict en lugar de comunicación implícita. Esto hace el flujo de trabajo rastreable y resistente a errores.

from typing import TypedDict, Optional, List, Annotated
from operator import add

class WorkflowState(TypedDict):
    input: str
    task_type: str
    research_result: Optional[str]
    draft: Optional[str]
    review_feedback: Optional[str]
    final_output: Optional[str]
    error: Optional[str]
    retry_count: int
    messages: Annotated[List[str], add]  # Se añade, no se sobrescribe

3. Manejo de errores y lógica de reintento

Los LLMs son poco confiables: límites de velocidad, timeouts, alucinaciones. Cada agente necesita manejo de errores:

import logging

logger = logging.getLogger("agent_orchestrator")

def agent_with_retry(agent_func, state, max_retries=3, fallback=None):
    for attempt in range(max_retries):
        try:
            logger.info(f"Agent {agent_func.__name__} - Intento {attempt + 1}")
            result = agent_func(state)
            if validate_result(result):
                return result
            else:
                logger.warning(f"Agent {agent_func.__name__} - resultado inválido")
                state["error"] = "Invalid result"
        except Exception as e:
            logger.error(f"Agent {agent_func.__name__} - Error: {str(e)}")
            state["error"] = str(e)
            state["retry_count"] = attempt + 1
    
    if fallback:
        return fallback(state)
    return {**state, "error": f"Agent failed after {max_retries} retries"}

def validate_result(result):
    if not result:
        return False
    if "error" in result and result["error"]:
        return False
    return True

4. Control de costos

Las llamadas a LLM son caras. Con 5 agentes y 3 revisiones cada uno, estamos hablando de 15+ llamadas a LLM. Con GPT-4o pueden ser 5-20$ por ejecución.

class CostTracker:
    def __init__(self, max_budget=10.0):
        self.costs = {}
        self.max_budget = max_budget
    
    def track(self, agent_name, tokens, model="gpt-4o"):
        cost_per_1k = {
            "gpt-4o": 0.005,
            "gpt-4o-mini": 0.0003,
            "claude-3.5-sonnet": 0.003,
        }
        rate = cost_per_1k.get(model, 0.005)
        cost = (tokens / 1000) * rate
        self.costs[agent_name] = self.costs.get(agent_name, 0) + cost
        
        if self.total() > self.max_budget:
            raise Exception(f"Budget excedido: ${self.total():.2f}")
    
    def total(self):
        return sum(self.costs.values())
    
    def report(self):
        lines = [f"Cost Report (Total: ${self.total():.4f})"]
        for agent, cost in sorted(self.costs.items()):
            lines.append(f"  {agent}: ${cost:.4f}")
        return "\n".join(lines)

Estrategias de optimización de costos:

• Caching: No ejecutes el mismo prompt dos veces
• Selección de modelo: Tareas simples con GPT-4o-mini ($0.30/M), complejas con GPT-4o ($5.00/M)
• Parada anticipada: Detente cuando el resultado sea “suficientemente bueno”
• Límites de tokens: Establece max_tokens por agente
• Modelos locales: Usa Ollama para desarrollo ($0)

5. Observabilidad y Logging

Sin observabilidad estás a ciegas. No sabes cuánto tiempo tarda cada agente o dónde ocurren los errores.

import logging
import time
from functools import wraps

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agent_orchestrator")

def observed_agent(agent_func):
    @wraps(agent_func)
    def wrapper(state):
        name = agent_func.__name__
        start = time.time()
        logger.info(f"[{name}] START - Input keys: {list(state.keys())}")
        try:
            result = agent_func(state)
            duration = time.time() - start
            logger.info(f"[{name}] DONE - {duration:.2f}s")
            return result
        except Exception as e:
            duration = time.time() - start
            logger.error(f"[{name}] ERROR - {duration:.2f}s - {str(e)}")
            raise
    return wrapper

Además: LangSmith para tracing detallado, que muestra cada llamada a LLM, transición de estado y cálculo de tokens visualmente:

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."

6. Human-in-the-Loop

Para decisiones críticas, un humano debe confirmar. LangGraph lo soporta nativamente con interrupt_before:

app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["publisher"]  # Pausa antes de Publisher
)

# Primera ejecución: corre hasta antes de Publisher
result = app.invoke(initial_state, config={"configurable": {"thread_id": "1"}})

# El humano revisa y aprueba
approval = input("¿Aprobar? (s/n): ")
result = app.invoke(
    {"approved": approval == "s"},
    config={"configurable": {"thread_id": "1"}}
)

¿Cuándo usar Human-in-the-Loop? Publicación de contenido, cambios de código en producción, consecuencias financieras, confianza baja del agente.

Antipatrones — Qué debes evitar

1. God Agent

Un agente que lo hace todo. El prompt crece enormemente, el LLM pierde el enfoque y la calidad baja. Difícil de depurar, costoso en producción. Solución: Divídelo en 3-5 agentes especializados.

2. Bucles infinitos

Agentes que se llaman entre sí sin condición de parada. Solución: Establece iteraciones máximas e implementa un freno de emergencia:

MAX_ITERATIONS = 5

def route_after_review(state):
    if state.get("approved"):
        return "publish"
    if state.get("iteration_count", 0) >= MAX_ITERATIONS:
        return "human_review"  # Freno de emergencia
    return "writer"

3. Sin validación

Confiar ciegamente en las salidas de los agentes. Los LLMs alucina e inventan fuentes. Solución: Valida cada salida:

def validate_research(result: str) -> bool:
    if not result or len(result) < 100:
        return False
    if "http" not in result:  # Al menos una fuente
        return False
    return True

4. Demasiados agentes

Más agentes equivalen a más coordinación, costos más altos y más puntos de fallo. Regla general: 3-5 agentes. Si necesitas más, divide en múltiples crews independientes.

5. Sin control de costos

Sin seguimiento de costos, los sistemas multi-agente pueden llegar a costar cientos de dólares por ejecución. Solución: CostTracker con límite presupuestario (ver arriba).

Comparación de frameworks para orquestación

Característica	LangGraph	CrewAI	Custom
Pipeline	✅	✅ (secuencial)	✅
Hub-and-Spoke	✅ (aristas condicionales)	✅ (jerárquico)	✅
DAG/Paralelo	✅ (nativo)	❌	✅ (asyncio)
Event-Driven	❌ (no nativo)	❌	✅ (bus personalizado)
Ciclos	✅ (aristas condicionales)	Limitado	✅
Human-in-the-Loop	✅ (interrupt)	Manual	✅
Checkpointing	✅ (nativo)	❌	✅ (personalizado)

Puntos relevantes para el examen

• Orquestación: Coordinación de múltiples agentes con distribución de tareas, comunicación, control, manejo de errores y gestión de recursos
• 4 Patrones: Pipeline (secuencial), Hub-and-Spoke (centralizado), DAG (paralelo), Event-Driven (reactivo)
• Mejores prácticas: Roles claros (SRP), estados tipificados explícitamente, lógica de reintentos con max_retries, control de costos con CostTracker, observabilidad con Logging + LangSmith, Human-in-the-Loop para decisiones críticas
• Anti-patrones: God Agent, bucles ilimitados, falta de validación, demasiados agentes, ausencia de control de costos
• Selección de framework: LangGraph para grafos complejos, CrewAI para prototipos rápidos, soluciones personalizadas para Event-Driven

Preguntas frecuentes

¿Cuándo necesito Multi-Agent en lugar de Single-Agent? Cuando la tarea requiere diferentes especialidades, es necesaria iteración (Writer ↔ Reviewer), es posible el paralelismo, o la tarea es demasiado compleja para un único prompt.

¿Cuál es el número óptimo de agentes? De 3 a 5 para la mayoría de casos de uso. Más de 8 se vuelven difíciles de coordinar y costosos. Para requisitos complejos es mejor dividir en múltiples Crews o grafos independientes.

¿Cuál es el mejor patrón de orquestación? Pipeline para tareas lineales simples, Hub-and-Spoke para enrutamiento dinámico, DAG para tareas paralelas, Event-Driven para sistemas débilmente acoplados. La mayoría de sistemas en producción utilizan una combinación de estos.

¿Cómo evito bucles infinitos? Siempre establece un número máximo de iteraciones. Implementa un mecanismo de parada que, tras demasiadas iteraciones, dirija hacia Human-in-the-Loop o END.

¿Cómo controlo los costos en sistemas Multi-Agent?

1. CostTracker con límite de presupuesto. 2) Modelos económicos para tareas simples. 3) Caché para prompts idénticos. 4) Early Stopping. 5) Modelos locales para desarrollo.

¿Necesito LangSmith en producción? Fuertemente recomendado. Sin tracing estás a ciegas: no sabes cuánto tiempo tarda cada agente ni dónde surgen los errores. LangSmith ofrece un nivel gratuito para proyectos pequeños.

Lecturas recomendadas