¡Hola a todos! Alex aquí de agntai.net. Estamos en marzo de 2026 y he estado pasando demasiado tiempo últimamente pensando en cómo construimos agentes de IA. Específicamente, he estado lidiando con el “código pegamento”: lo que conecta todas las salidas elegantes de LLM, llamadas a herramientas y gestión de estado. Todos hemos visto las impresionantes demostraciones, ¿verdad? Agentes haciendo cosas asombrosas. Pero luego intentas construir uno para un problema del mundo real y te encuentras con un muro de callbacks, lógica condicional y actualizaciones de estado. Se siente menos como construir un sistema inteligente y más como gestionar una fábrica de espaguetis muy compleja.
Así que hoy quiero hablar de algo que ha ido ganando terreno silenciosamente y, francamente, está salvando mi cordura: Arquitecturas Basadas en Eventos para Agentes de IA. No es un concepto nuevo en ingeniería de software, pero aplicarlo de manera reflexiva a los agentes de IA, especialmente aquellos que orquestan múltiples interacciones con LLM y herramientas externas, se siente como un soplo de aire fresco. Olvidemos el pensamiento lineal, paso a paso, por un momento. Pensemos en sistemas reactivos.
Mi Lucha Personal con los Monolitos de Agentes
Hace unos meses, estaba trabajando en un agente diseñado para ayudarme a gestionar mi pipeline de escritura freelance. La idea era simple: supervisaría mi bandeja de entrada en busca de nuevas consultas, redactaría respuestas iniciales, sugeriría artículos relevantes de mi base de conocimiento y hasta ayudaría a programar llamadas de seguimiento. Parecía lo suficientemente sencillo.
Mi enfoque inicial era bastante típico: un bucle principal. Obtener correo electrónico. Analizar el correo electrónico. Decidir acción (redactar, programar, buscar). Llamar a LLM. Procesar la salida de LLM. Llamar a la herramienta (API del calendario, API de correo electrónico, API de base de conocimiento). Actualizar el estado interno. Repetir.
Comenzó bien, pero a medida que añadía más “inteligencia” y más herramientas, se convirtió en una pesadilla. ¿Qué pasaría si la llamada a la API del calendario fallaba? ¿Qué pasaría si el LLM alucinaba un contacto que no existía? ¿Qué pasaría si necesitaba pausar y pedir entrada humana para una decisión crítica? Mi único script de agente monolítico rápidamente se convirtió en un laberinto anidado de `if/else` con bloques `try/except` por todas partes. Depurar era un desastre. Modificar una parte a menudo rompía otra. Me sentía como si estuviera constantemente parchando goteras en un barco hundiéndose.
Recuerdo una noche tarde, tratando de averiguar por qué mi agente seguía redactando respuestas para correos electrónicos que ya había procesado. Resultó que la actualización del estado de “correo procesado” ocurría *después* de una posible reejecución de LLM en una ruta de falla. Era una condición de carrera clásica en un sistema que no estaba diseñado para manejar operaciones asincrónicas y no determinísticas con elegancia. Fue entonces cuando comencé a buscar una mejor manera.
Por qué los Agentes Basados en Eventos Tienen Sentido
Pensemos en cómo trabajamos los humanos. No seguimos normalmente un guion estricto y predefinido para cada interacción. Reaccionamos ante las cosas. Alguien hace una pregunta: eso es un evento. Lo procesamos y respondemos: eso es otro evento. Recibimos un nuevo dato: evento. Decidimos usar una herramienta (como abrir un navegador): evento. Nuestro “estado” interno cambia constantemente basado en estos eventos.
Una arquitectura basada en eventos (EDA) para agentes de IA refleja este patrón de interacción natural. En lugar de un flujo de control rígido, los componentes emiten eventos cuando sucede algo significativo. Otros componentes (escuchadores, manejadores) reaccionan a estos eventos. Esto trae varios beneficios clave:
- Modularidad: Los componentes se vuelven poco acoplados. Un ejecutor de herramientas no necesita saber quién lo llamó o qué sucederá después; simplemente emite un evento como “tool_call_succeeded” o “tool_call_failed”.
- Flexibilidad: Es mucho más fácil añadir nuevas características o modificar las existentes. ¿Quieres una nueva herramienta? Simplemente añade un manejador que escuche un evento de intención específico. ¿Necesitas registrar cada llamada a LLM? Añade un registrador que escuche “llm_response_received”.
- Resiliencia: Si un componente falla, es menos probable que derribe todo el sistema. Un evento puede reintentarse, o un manejador alternativo puede hacerse cargo. Puedes construir colas de mensajes muertas para eventos que no pueden ser procesados.
- Concurrente: Muchos eventos pueden ser procesados en paralelo, ya sea por diferentes manejadores o por el mismo manejador en diferentes instancias de evento. Esto es crucial para agentes que necesitan gestionar múltiples tareas en curso.
- Observabilidad: El flujo de eventos proporciona un registro claro y auditable de todo lo que está haciendo el agente. Puedes rastrear fácilmente el flujo de información y decisiones.
La Idea Central: Eventos, Despachadores y Manejadores
En su esencia, una EDA necesita tres cosas:
- Eventos: Estructuras de datos simples que describen algo que ocurrió (por ejemplo, `ToolCalled`, `LLMResponseReceived`, `UserQueryReceived`).
- Un Despachador de Eventos: Un mecanismo central que toma un evento y lo dirige a todas las partes interesadas.
- Manejadores de Eventos: Funciones o clases que “escuchan” tipos específicos de eventos y ejecutan alguna lógica cuando reciben uno.
Veamos un ejemplo simplificado. Imagina nuestro agente de pipeline de escritura. En lugar de una función gigante, tenemos:
- Un evento `UserQueryReceived` (cuando llega un nuevo correo electrónico).
- Un evento `LLMInputGenerated` (cuando hemos creado un mensaje para el LLM).
- Un evento `LLMResponseReceived` (cuando el LLM devuelve su salida).
- Un evento `ToolCallRequested` (cuando el LLM sugiere usar una herramienta).
- Un evento `ToolCallSucceeded` / `ToolCallFailed` (después de una interacción con la herramienta).
- Un evento `DraftResponseReady` (cuando un borrador está listo para revisión).
Cada uno de estos eventos lleva datos relevantes: el contenido del correo electrónico, el mensaje/respuesta del LLM, el nombre y los argumentos de la herramienta, etc.
Bloques de Construcción: Un Enfoque Pythonic
No necesitas una cola de mensajes de alto rendimiento como Kafka para sistemas de agentes simples (aunque para producción, agentes distribuidos, ¡definitivamente podrías!). Para un agente de proceso único, un simple despachador de eventos en memoria funciona de maravilla.
Paso 1: Define tus Eventos
Me gusta usar `dataclasses` para eventos porque son limpias y explícitas.
from dataclasses import dataclass
from typing import Any, Dict, Optional
@dataclass
class AgentEvent:
"""Clase base para todos los eventos de agente."""
timestamp: float # Añadir un timestamp para ordenar y depurar
metadata: Dict[str, Any] = None
def __post_init__(self):
if self.metadata is None:
self.metadata = {}
@dataclass
class UserQueryReceived(AgentEvent):
query_id: str
content: str
source: str = "email"
@dataclass
class LLMRequestSent(AgentEvent):
query_id: str
model_name: str
prompt: str
@dataclass
class LLMResponseReceived(AgentEvent):
query_id: str
model_name: str
response_text: str
tool_calls: Optional[list[Dict[str, Any]]] = None
@dataclass
class ToolCallRequested(AgentEvent):
query_id: str
tool_name: str
tool_args: Dict[str, Any]
@dataclass
class ToolCallSucceeded(AgentEvent):
query_id: str
tool_name: str
tool_args: Dict[str, Any]
result: Any
@dataclass
class ToolCallFailed(AgentEvent):
query_id: str
tool_name: str
tool_args: Dict[str, Any]
error_message: str
@dataclass
class AgentThoughtEvent(AgentEvent):
query_id: str
thought: str
@dataclass
class FinalResponseReady(AgentEvent):
query_id: str
response_content: str
action_taken: str
Observa el `query_id`. ¡Esto es crítico! Nos permite correlacionar eventos que pertenecen a la misma interacción o tarea general del usuario. Sin él, tu flujo de eventos se convierte en un caos.
Paso 2: Crea un Despachador de Eventos
Este es el lugar donde se dirigen los eventos. Un diccionario simple que mapea tipos de eventos a listas de manejadores funciona bien.
import time
from collections import defaultdict
from typing import Callable, Type, List, Union
class EventDispatcher:
def __init__(self):
self._handlers: defaultdict[Type[AgentEvent], List[Callable[[AgentEvent], None]]] = defaultdict(list)
def register_handler(self, event_type: Type[AgentEvent], handler: Callable[[AgentEvent], None]):
"""Registrar una función para manejar un tipo de evento específico."""
self._handlers[event_type].append(handler)
def dispatch(self, event: AgentEvent):
"""Enviar un evento a todos los manejadores registrados."""
# Asegúrate de que el timestamp esté establecido si aún no lo está
if not hasattr(event, 'timestamp') or event.timestamp is None:
event.timestamp = time.time()
# Despachar a los manejadores específicos del tipo de evento
for handler in self._handlers[type(event)]:
try:
handler(event)
except Exception as e:
print(f"Error en el manejador {handler.__name__} para el evento {type(event).__name__}: {e}")
# Potencialmente despachar un evento de error aquí para solidez
# También despachar a manejadores registrados para el tipo base AgentEvent
# Esto permite un registro o monitoreo genérico
for handler in self._handlers[AgentEvent]:
try:
handler(event)
except Exception as e:
print(f"Error en el manejador genérico {handler.__name__} para el evento {type(event).__name__}: {e}")
Paso 3: Define tus Manejadores
Cada manejador es una función simple que toma un objeto de evento. Realiza su tarea específica y, crucialmente, puede despachar nuevos eventos.
Esbozamos algunos manejadores para nuestro agente de escritura:
# Asumiendo que 'dispatcher' es una instancia de EventDispatcher
# --- Manejador para la consulta inicial del usuario ---
def handle_user_query(event: UserQueryReceived):
print(f"[{event.query_id}] Consulta del usuario recibida: {event.content[:50]}...")
# Aquí, típicamente usaríamos un LLM para decidir la intención inicial
# Para simplificar, asumamos que siempre va al LLM para redactar
prompt = f"Eres un asistente útil para un escritor freelance. Redacta una respuesta inicial y cortés a la siguiente consulta del cliente, y sugiere una acción de seguimiento (por ejemplo, 'schedule_call', 'search_knowledge_base'):\n\n{event.content}\n\nSalida en JSON con 'draft_response' y 'suggested_action'."
# Despachar un evento para enviar al LLM
dispatcher.dispatch(LLMRequestSent(
query_id=event.query_id,
model_name="gpt-4",
prompt=prompt,
metadata={"previous_event": type(event).__name__}
))
# --- Manejador para las respuestas del LLM ---
def handle_llm_response(event: LLMResponseReceived):
print(f"[{event.query_id}] Respuesta del LLM recibida: {event.response_text[:50]}...")
# Analizar la respuesta del LLM (esto sería más sólido con Pydantic)
try:
llm_output = json.loads(event.response_text)
draft = llm_output.get("draft_response")
action = llm_output.get("suggested_action")
dispatcher.dispatch(AgentThoughtEvent(
query_id=event.query_id,
thought=f"Acción sugerida por el LLM: {action}"
))
if draft:
dispatcher.dispatch(DraftResponseReady(
query_id=event.query_id,
response_content=draft,
action_taken="drafted_initial_response"
))
if action == "schedule_call":
# Asumir que el LLM también proporcionó detalles de la llamada si es necesario
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="calendar_scheduler",
tool_args={"client_email": "[email protected]", "duration": "30min"} # Placeholder
))
elif action == "search_knowledge_base":
# Asumir que el LLM proporcionó la consulta de búsqueda
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="knowledge_base_search",
tool_args={"query": "artículos relacionados sobre agentes de IA"} # Placeholder
))
except json.JSONDecodeError:
print(f"[{event.query_id}] Respuesta del LLM no es un JSON válido. Enviando para revisión humana.")
dispatcher.dispatch(FinalResponseReady(
query_id=event.query_id,
response_content="Error al analizar la salida del LLM, necesita humano. Respuesta original del LLM: " + event.response_text,
action_taken="human_review_needed"
))
# --- Manejador para solicitudes de llamadas a herramientas ---
def handle_tool_call_request(event: ToolCallRequested):
print(f"[{event.query_id}] Solicitud de llamada a herramienta: {event.tool_name} con args {event.tool_args}")
# Simular ejecución de herramienta
if event.tool_name == "calendar_scheduler":
# En un sistema real, esto llamaría a una API real
print(f"Programando llamada para {event.tool_args.get('client_email')}...")
time.sleep(1) # Simular retraso de red
if random.random() > 0.1: # 90% tasa de éxito
dispatcher.dispatch(ToolCallSucceeded(
query_id=event.query_id,
tool_name=event.tool_name,
tool_args=event.tool_args,
result={"status": "scheduled", "meeting_link": "https://meet.google.com/abc-xyz"}
))
else:
dispatcher.dispatch(ToolCallFailed(
query_id=event.query_id,
tool_name=event.tool_name,
tool_args=event.tool_args,
error_message="Error en la API del calendario o ocupado"
))
# ... otras herramientas ...
# --- Manejador genérico de registros ---
def log_all_events(event: AgentEvent):
print(f"LOG: {type(event).__name__} - {event.query_id} - {event.timestamp}")
# --- Registrar manejadores ---
dispatcher = EventDispatcher()
dispatcher.register_handler(UserQueryReceived, handle_user_query)
dispatcher.register_handler(LLMResponseReceived, handle_llm_response)
dispatcher.register_handler(ToolCallRequested, handle_tool_call_request)
# ... otros manejadores para ToolCallSucceeded, ToolCallFailed, etc.
dispatcher.register_handler(AgentEvent, log_all_events) # Manejador genérico para todos los eventos
Este es un ejemplo muy simplificado, pero puedes ver cómo cada pieza es independiente. El `handle_user_query` no sabe *cómo* se enviará la solicitud al LLM, solo que necesita emitir un evento `LLMRequestSent`. De manera similar, `handle_llm_response` no se preocupa por quién envió el mensaje original; solo procesa la respuesta y decide qué hacer a continuación.
Simulando llamadas a LLM y a herramientas
Para un sistema real, `LLMRequestSent` dispararía un componente que realmente llama a la API del LLM, y luego despacha `LLMResponseReceived` cuando regresa el resultado. Aquí es donde `asyncio` o un simple grupo de hilos pueden ser útiles para llamadas concurrentes al LLM o ejecuciones de herramientas sin bloquear el bucle de eventos.
import asyncio
import json
import random
import time
# ... (Definiciones de eventos y EventDispatcher de arriba) ...
# Mock de API del LLM
async def mock_llm_call(prompt: str) -> str:
print(f" [Mock LLM] Procesando prompt: {prompt[:80]}...")
await asyncio.sleep(random.uniform(1.0, 3.0)) # Simular latencia del LLM
# Lógica muy básica para nuestro caso de uso
if "schedule_call" in prompt:
return json.dumps({
"draft_response": "¡Gracias por tu consulta! Me encantaría hablar más. ¿Qué tal si programamos una llamada rápida la próxima semana?",
"suggested_action": "schedule_call"
})
elif "search_knowledge_base" in prompt:
return json.dumps({
"draft_response": "¡Gran pregunta! He redactado una respuesta y también he buscado algunos artículos relevantes.",
"suggested_action": "search_knowledge_base"
})
else:
return json.dumps({
"draft_response": "¡Gracias por contactarme! He revisado tu solicitud y redactado una respuesta inicial.",
"suggested_action": "none"
})
# Componente del Agente LLM (escucha LLMRequestSent, despacha LLMResponseReceived)
async def llm_agent_component(event: LLMRequestSent, dispatcher: EventDispatcher):
response_text = await mock_llm_call(event.prompt)
# En un sistema real, analizarías para llamadas a herramientas de la respuesta del LLM
tool_calls = [] # Placeholder
dispatcher.dispatch(LLMResponseReceived(
query_id=event.query_id,
model_name=event.model_name,
response_text=response_text,
tool_calls=tool_calls,
metadata={"original_prompt_event": event.timestamp}
))
# Registrar el manejador asíncrono
dispatcher.register_handler(LLMRequestSent, lambda e: asyncio.create_task(llm_agent_component(e, dispatcher)))
# ... (otros manejadores de arriba) ...
# Para ejecutar un ejemplo:
async def main():
query_id = "user_email_123"
dispatcher.dispatch(UserQueryReceived(
query_id=query_id,
content="Necesito un artículo sobre agentes de IA impulsados por eventos y una llamada de seguimiento.",
timestamp=time.time()
))
# Dar algo de tiempo para procesar eventos
await asyncio.sleep(10)
print("\n--- Procesamiento completado para user_email_123 ---\n")
query_id_2 = "user_email_456"
dispatcher.dispatch(UserQueryReceived(
query_id=query_id_2,
content="¿Puedes resumir mis artículos pasados sobre arquitecturas de aprendizaje profundo?",
timestamp=time.time()
))
await asyncio.sleep(10)
print("\n--- Procesamiento completado para user_email_456 ---\n")
if __name__ == "__main__":
asyncio.run(main())
Introduje `asyncio.create_task` para permitir que el `llm_agent_component` se ejecute en paralelo con otros manejadores o despachos subsiguientes. Aquí es donde las arquitecturas impulsadas por eventos realmente destacan por su rendimiento y capacidad de respuesta en los agentes de IA.
Conclusiones prácticas para tu próximo proyecto de agente
- Comienza simple, piensa en eventos: Incluso para un agente pequeño, esboza los eventos clave que ocurren. ¿Qué desencadena qué? ¿Qué información necesita ser transmitida?
- Define esquemas de eventos claros: Usa `dataclasses` o modelos de Pydantic para tus eventos. Esto asegura consistencia y facilita la depuración. Siempre incluye un `query_id` o `correlation_id`.
- Separa las preocupaciones: Cada manejador debe hacer una cosa bien. No intentes meter demasiada lógica en un solo manejador. Si un manejador necesita hacer una llamada externa, debe despachar un evento de solicitud y esperar por un evento de respuesta correspondiente.
- Adopta la asincronía: Las interacciones de agentes de IA (llamadas LLM, ejecución de herramientas) son inherentemente asíncronas. Utiliza `asyncio` o un marco similar para manejar estas concurrentemente sin bloquear tu bucle de eventos.
- Incorpora observabilidad: Un registrador de eventos genérico (como mi `log_all_events`) es increíblemente valioso. Puedes fácilmente enviar estos eventos a un sistema de monitoreo o simplemente imprimirlos para desarrollo. Este flujo de eventos se convierte en el log interno del “proceso de pensamiento” de tu agente.
- Manejo de errores con eventos: En lugar de una anidación profunda de `try/except`, despacha eventos `ErrorEvent` o `ToolCallFailed`. Otros manejadores pueden escuchar específicamente estos para implementar lógica de reintentos, copias de seguridad o solicitudes de intervención humana.
Pasar a un modelo impulsado por eventos cambió completamente cómo pienso acerca de construir agentes. Me alejó de intentar anticipar cada posible camino en un flujo lineal y hacia construir un sistema que reacciona inteligentemente a su entorno y sus propias operaciones internas. Es una manera más resiliente, escalable y, francamente, más agradable de construir agentes de IA complejos.
Pruébalo para tu próximo proyecto de agente. ¡Podrías encontrarte desenredando ese código espagueti más rápido de lo que piensas!
🕒 Published: