Tool Orchestration en AI Agents 2026: El 90% Ejecuta en Serie y se Deja un 47% de Latencia
Aprende a paralelizar tool calls en AI Agents usando el Patrón de Orquestación por Capas de Dependencia. Código Python, DAGs y asyncio.gather() para reducir latencia.
El 90% de los AI Agents que Ves en GitHub Ejecutan Tool Calls como si Fuera 1999
En serie. Uno detrás de otro. Como una lista de la compra.
Y eso está matando tu latencia.
Abre cualquier repo de ejemplo de LangChain, AutoGPT o Mastra. El patrón es idéntico: el LLM decide la siguiente tool, la ejecuta con `await`, espera el resultado, y repite. Un bucle FIFO. Secuencial. Predecible.
*El problema no es que funcione mal. Es que podrías estar sirviendo la misma respuesta en la mitad de tiempo. *
La estimación basada en escenarios reales con 8 tool calls — donde 5 son independientes y 3 dependientes en cadena — muestra una reducción de latencia del ~47% cuando paralelizas las calls que no necesitan esperarse entre sí. No es un benchmark de laboratorio. Es física básica de grafos aplicada a agent loops.
El Supuesto Falso: "El Orden Importa Siempre"
La mayoría de desarrolladores asumen que las tool calls de su agente son una secuencia lineal obligada. Y es comprensible: cuando depuras paso a paso, ves que Tool B usa el output de Tool A y asumes que todas las relaciones son así.
Pero la realidad es muy distinta.
En un agente típico de atención al cliente que necesita:
1. Buscar el usuario en la base de datos
2. Consultar el clima en su ciudad
3. Revisar el calendario de citas disponibles
4. Obtener el historial de pedidos
5. Generar una respuesta
Las calls 2, 3 y 4 son *totalmente independientes* entre sí. No necesitan el output de las demás. Pero ejecutadas secuencialmente, suman su latencia individual: 300ms + 400ms + 350ms = 1.050ms. En paralelo: 400ms (la más lenta).
Esa diferencia de 650ms es el 62% de latencia que te estás comiendo sin necesidad.
❌ Enfoque secuencial por defecto:
```python
Anti-patrón: latencia total = suma de todas las latencias
user_data = await db.lookup_user(user_id) # 200ms
weather = await weather_api.get_forecast(city) # 300ms
calendar = await calendar_api.get_slots(user_id) # 400ms
orders = await orders_api.get_history(user_id) # 350ms
response = await llm.generate(context) # 800ms
Latencia total: 200 + 300 + 400 + 350 + 800 = 2.050ms
```
✅ Enfoque DAG con paralelización inteligente:
```python
Latencia total = batch paralelo más lento + cadena secuencial
user_data = await db.lookup_user(user_id) # 200ms (requerido para calendar y orders)
Paralelizar independientes + dependientes del usuario
weather, calendar, orders = await asyncio.gather(
weather_api.get_forecast(city), # 300ms
calendar_api.get_slots(user_id), # 400ms (necesita user_id)
orders_api.get_history(user_id) # 350ms (necesita user_id)
)
response = await llm.generate(context) # 800ms
Latencia total: 200 + 400 + 800 = 1.400ms → 32% menos
```
El Patrón de 3 Fases para Tool Orchestration
El framework que he estado usando en producción se llama Patrón de Orquestación por Capas de Dependencia. Son tres fases:
1. Análisis de Dependencias
Antes de ejecutar nada, parsea el plan de tool calls y construye un grafo acíclico dirigido (DAG). Identifica qué calls necesitan el output de otras calls y cuáles son hojas independientes.
```python
from collections import defaultdict
class DependencyAnalyzer:
"""Fase 1: Construir el grafo de dependencias entre tool calls."""
def __init__(self, tool_calls: list[dict]):
self.tool_calls = tool_calls
self.graph = defaultdict(set) # tool_id -> set de tool_ids que lo necesitan
self.depends_on = defaultdict(set) # tool_id -> set de tool_ids de los que depende
def analyze(self):
"""Analizar dependencias basadas en referencias a outputs previos."""
for call in self.tool_calls:
call_id = call["id"]
for param_key, param_value in call.get("params", {}).items():
Si un parámetro referencia el output de otra tool
if isinstance(param_value, str) and param_value.startswith("$tool."):
referenced_tool = param_value.split(".")[1]
self.graph[referenced_tool].add(call_id)
self.depends_on[call_id].add(referenced_tool)
return self.graph, self.depends_on
def get_independent_calls(self):
"""Devolver calls que no dependen de ninguna otra."""
return [c for c in self.tool_calls if c["id"] not in self.depends_on]
```
2. Agrupación Paralela
Agrupa todas las tool calls independientes en batches ejecutables concurrentemente. Cada batch se lanza con `asyncio.gather()`.
```python
import asyncio
from collections import deque
class ParallelBatchExecutor:
"""Fase 2: Ejecutar batches paralelos manteniendo el orden topológico."""
def __init__(self, graph, depends_on, tool_map):
self.graph = graph
self.depends_on = depends_on
self.tool_map = tool_map # id -> función ejecutable
self.results = {}
async def execute_all(self):
"""Orden topológico = batches paralelos por nivel."""
Calcular in-degree de cada nodo
in_degree = {tid: len(self.depends_on.get(tid, set()))
for tid in self.tool_map}
queue = deque([tid for tid, deg in in_degree.items() if deg == 0])
processed = set()
while queue:
Batch paralelo: todos los nodos con in-degree 0
batch = list(queue)
queue.clear()
Ejecutar batch en paralelo
tasks = {tid: self._execute_tool(tid) for tid in batch}
batch_results = await asyncio.gather(*tasks.values(),
return_exceptions=True)
for tid, result in zip(batch, batch_results):
if isinstance(result, Exception):
self.results[tid] = {"error": str(result)}
else:
self.results[tid] = result
processed.add(tid)
Reducir in-degree de dependientes
for dependent in self.graph.get(tid, set()):
in_degree[dependent] -= 1
if in_degree[dependent] == 0 and dependent not in processed:
queue.append(dependent)
return self.results
async def _execute_tool(self, tool_id):
fn = self.tool_map[tool_id]
Inyectar resultados de dependencias como contexto
deps = self.depends_on.get(tool_id, set())
context = {dep: self.results[dep] for dep in deps if dep in self.results}
return await fn(context)
```
3. Ejecución con Merge
Ejecuta batches en orden topológico, recolectando outputs y pasándolos como contexto a tools dependientes en batches posteriores. Cada batch es un nivel del DAG.
```python
Implementación completa del planificador de 3 fases
class ToolOrchestrator:
"""Fase 1 + 2 + 3: Pipeline completo de orquestación."""
def __init__(self, tool_definitions: list[dict], executor_map: dict):
self.tool_defs = tool_definitions
self.executor_map = executor_map
self.analyzer = DependencyAnalyzer(tool_definitions)
async def run(self):
Fase 1: Analizar dependencias
graph, depends_on = self.analyzer.analyze()
Fase 2+3: Crear ejecutor y lanzar batches paralelos
tool_map = {t["id"]: self.executor_map[t["tool"]]
for t in self.tool_defs}
executor = ParallelBatchExecutor(graph, depends_on, tool_map)
results = await executor.execute_all()
Merge: pasar todo el contexto al LLM para generar respuesta
merged_context = self._build_final_context(results)
return merged_context
def _build_final_context(self, results):
context_parts = []
for tool_id, result in results.items():
if "error" not in result:
context_parts.append(f"Tool {tool_id}: {result}")
return "\n".join(context_parts)
```
Por Qué el ~47% No es un Número Mágico y Cuándo Aplica
La métrica del ~47% no sale de un benchmark estándar con 40 GPUs. Es una estimación situacional basada en el escenario descrito: 8 tool calls, 5 independientes, 3 en cadena.
*Latencia secuencial: * 8 unidades de tiempo.
*Latencia paralelizada: * ~4.3 unidades (el batch paralelo más lento + la cadena secuencial).
Pero esto varía enormemente según el dominio:
Agente de investigación web (búsquedas mayoritariamente independientes): ganancia del 60-70%.
Agente de procesamiento de pagos (cada paso depende del anterior): ganancia del 0%.
Chatbot con búsqueda RAG (vector DB + API de contexto de usuario): ganancia del 30-40%.
El punto no es paralelizar siempre. *El punto es paralelizar inteligentemente. *
Haz este ejercicio: audita el 100% de las tool calls en 3 sesiones típicas de tu agente. Cuenta cuántas podrían ejecutarse simultáneamente. Te vas a llevar una sorpresa.
Manejando el Error en un Batch Paralelo
La objeción más común es: "Paralelizar introduce complejidad de errores — si una tool falla, ¿cómo manejas el batch entero?"
Respuesta: el patrón de resultados parciales.
```python
async def execute_batch_with_failures(tasks: dict) -> dict:
"""Ejecuta un batch y recolecta resultados + errores por separado."""
results = {}
errors = {}
return_exceptions=True evita que una tool falle todo el batch
raw_results = await asyncio.gather(
*tasks.values(),
return_exceptions=True
)
for task_id, result in zip(tasks.keys(), raw_results):
if isinstance(result, Exception):
errors[task_id] = str(result)
results[task_id] = None
else:
results[task_id] = result
Devolver ambos para que el LLM decida si puede continuar
return {"results": results, "errors": errors}
```
Esto añade unas 15 líneas de código pero cambia drásticamente la robustez. El LLM recibe tanto los resultados exitosos como los errores, y decide si puede continuar con datos parciales o si el error es fatal.
El Efecto Secundario que Nadie Menciona: Menos Llamadas al LLM
Paralelizar tool calls no solo reduce latencia. También reduce el número total de llamadas al LLM.
En un loop secuencial, tras cada tool call el LLM necesita re-evaluar el contexto completo para decidir el siguiente paso. Con un DAG bien estructurado, el LLM planifica una vez (o con menos re-planificaciones) y ejecuta tools en batches.
*Menos re-planificaciones = menos tokens consumidos = menos coste de API. *
Este efecto secundario es a menudo más valioso que la reducción de latencia. Sobre todo cuando estás usando modelos frontier que cobran por token de output.
Qué Hacer Mañana Mismo
1. Audita tus tool calls. Saca un log de 3 sesiones típicas de tu agente. Identifica qué calls son independientes y cuáles no. Verás que el 60-70% son paralelizables.
2. Implementa el DependencyAnalyzer. Son 30 líneas de Python. No necesitas un framework nuevo. Solo una función que construya el grafo.
3. Cambia tu loop secuencial por batches paralelos. Usa `asyncio.gather()` con `return_exceptions=True`. Son 10 líneas de diff en tu código actual.
4. Mide antes y después. Latencia total. Tokens consumidos. Errores. El ~47% no es promesa — es hipótesis. Valídala con tus datos.
El 90% de los AI Agents en producción ejecutan tool calls secuencialmente porque es más fácil de programar y depurar. Pero en producción real, con 8-12 tool calls por tarea, la latencia se acumula linealmente.
*No necesitas un framework nuevo. Necesitas aplicar teoría de grafos a tu agent loop. *
Y la teoría de grafos tiene 60 años. No es IA de frontera. Es algoritmia básica que la mayoría sigue ignorando por costumbre.
Lee el artículo completo en brianmenagomez.com
Más sobre mis servicios en brianmenagomez.com
Herramientas: Conversor IAE CNAE · Gestorias cerca de ti · Calculadora IRPF