¿Qué es un Orquestador de IA? Coordinando Agentes y Flujos

¿Qué es un orquestador en IA?

Un orquestador coordina y gestiona la ejecución de múltiples agentes, herramientas y LLMs para completar tareas complejas. Es el director que decide qué se ejecuta, cuándo, en qué orden y cómo manejar errores.

Funciones:

• Routing: dirigir consultas al agente especializado correcto
• Gestión de estado: mantener contexto entre múltiples pasos
• Manejo de errores: reintentar, escalar o recuperar
• Paralelización: ejecutar tareas independientes simultáneamente
• Human-in-the-loop: pausar para validación cuando es necesario
• Observabilidad: logs y trazas de cada paso

Patrones de orquestación

1. Secuencial: A → B → C. Úsalo cuando cada paso depende del anterior.

2. Paralelo (fan-out/fan-in): Una tarea se divide, ejecuta en paralelo, y se combina. Ideal para analizar el mismo documento con 3 agentes simultáneamente.

3. Condicional: Según el resultado, el flujo toma caminos distintos. Ejemplo: cliente premium → flujo VIP; cliente nuevo → flujo onboarding.

4. Iterativo: Un agente refina hasta cumplir un criterio. Ejemplo: generar código → ejecutar tests → si falla regenerar → hasta pasar.

5. Jerárquico: Un supervisor delega a especialistas y agrega resultados. El patrón más poderoso para sistemas complejos.

LangGraph: el orquestador Python más usado

LangGraph es el estándar en el ecosistema Python/LangChain. Modela el flujo como grafo donde nodos son operaciones y aristas son transiciones. Con checkpointing en Redis o PostgreSQL, los flujos sobreviven reinicios y permiten pausas para validación humana.

Otros orquestadores: Prefect, Airflow, Temporal

• Prefect: Workflows Python con UI de monitoreo. Ideal para pipelines de datos.
• Apache Airflow: Estándar empresarial para pipelines complejos.
• Temporal: Motor durable para flujos de horas o días (due diligence, auditorías).

Ejemplo de código

# Orquestador multi-agente: analisis paralelo con LangGraph
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class State(TypedDict):
    tarea: str
    analisis: Annotated[List[str], operator.add]
    reporte: str

def agente_financiero(state: State) -> State:
    resultado = llm.invoke(f"Analiza aspectos financieros de: {state['tarea']}")
    return {"analisis": [f"Finanzas: {resultado.content}"]}

def agente_legal(state: State) -> State:
    resultado = llm.invoke(f"Analiza aspectos legales de: {state['tarea']}")
    return {"analisis": [f"Legal: {resultado.content}"]}

def agente_riesgo(state: State) -> State:
    resultado = llm.invoke(f"Evalua riesgos de: {state['tarea']}")
    return {"analisis": [f"Riesgos: {resultado.content}"]}

def sintesis(state: State) -> State:
    todos = "\n\n".join(state["analisis"])
    reporte = llm.invoke(f"Genera reporte ejecutivo:\n{todos}")
    return {"reporte": reporte.content}

graph = StateGraph(State)
for nombre, fn in [("financiero", agente_financiero), ("legal", agente_legal),
                   ("riesgo", agente_riesgo), ("sintesis", sintesis)]:
    graph.add_node(nombre, fn)

graph.set_entry_point("financiero")
# Los 3 agentes se ejecutan, luego sintesis agrega
graph.add_edge("financiero", "sintesis")
graph.add_edge("legal", "sintesis")
graph.add_edge("riesgo", "sintesis")
graph.add_edge("sintesis", END)