Olá a todos, Alex aqui do agntai.net! Estamos em março de 2026 e eu passei muito tempo recentemente pensando sobre como construímos agentes de IA. Em particular, estou tendo dificuldades com o “código de ligação” – os elementos que conectam todas as saídas LLM sofisticadas, as chamadas de ferramentas e a gestão de estado. Todos nós já vimos as demonstrações impressionantes, não é? Agentes fazendo coisas incríveis. Mas então, você tenta construir um para um problema do mundo real e se depara com uma parede de lembretes, lógica condicional e atualizações de estado. Isso se parece menos com a construção de um sistema inteligente e mais com a gestão de uma fábrica de espaguete muito complexa.
Então hoje, quero falar sobre algo que ganhou popularidade discretamente e, para ser honesto, preserva minha saúde mental: Arquiteturas Orientadas a Eventos para Agentes de IA. Este não é um novo conceito em engenharia de software, longe disso, mas aplicá-lo de forma reflexiva aos agentes de IA, especialmente aqueles que orquestram múltiplas interações LLM e ferramentas externas, parece ser um sopro de ar fresco. Vamos esquecer por um momento o pensamento linear e passo a passo. Vamos pensar em sistemas reativos.
Minha Luta Pessoal com os Monólitos de Agentes
Há alguns meses, eu estava trabalhando em um agente projetado para me ajudar a gerenciar meu pipeline de redação freelance. A ideia era simples: ele monitoraria minha caixa de entrada para novas solicitações, redigiria respostas iniciais, sugeriria artigos antigos relevantes de minha base de conhecimento e até me ajudaria a agendar chamadas de acompanhamento. Isso parecia bastante simples.
Minha abordagem inicial era bastante típica: um loop principal. Receber um email. Analisar o email. Determinar a ação (redigir, agendar, pesquisar). Chamar LLM. Processar a saída LLM. Chamar uma ferramenta (API de calendário, API de email, API de base de conhecimento). Atualizar o estado interno. Repetir.
Começou bem, mas à medida que eu adicionava mais “inteligência” e mais ferramentas, isso se tornava um pesadelo. O que aconteceria se a chamada à API de calendário falhasse? O que aconteceria se o LLM alucinasse um contato que não existia? O que aconteceria se eu precisasse fazer uma pausa e pedir uma opinião humana para uma decisão crítica? Meu script de agente único e monolítico rapidamente se transformou em um labirinto emaranhado de `if/else` com blocos `try/except` por toda parte. A depuração era um pesadelo. Alterar uma parte muitas vezes quebrava outra. Era como se eu precisasse constantemente tampão de vazamentos em um barco que estava afundando.
Eu me lembro de uma noite tardia, tentando entender por que meu agente continuava a redigir respostas para emails que já havia processado. Aconteceu que a atualização de estado para “email processado” ocorria *após* uma nova execução possível de LLM em um caminho de falha. Era uma condição de concorrência clássica em um sistema que não foi projetado para gerenciar operações assíncronas e não determinísticas de maneira adequada. Foi nesse momento que comecei a buscar uma maneira melhor de fazer.
Por que os Agentes Orientados a Eventos Fazem Sentido
Pense na forma como os humanos funcionam. Geralmente, não seguimos um roteiro rígido e pré-definido para cada interação. Nós reagimos às coisas. Alguém faz uma pergunta – isso é um evento. Nós o processamos e respondemos – isso é outro evento. Recebemos uma nova informação – evento. Decidimos usar uma ferramenta (como abrir um navegador) – evento. Nosso “estado” interno muda constantemente com base nesses eventos.
Uma arquitetura orientada a eventos (AOE) para agentes de IA reflete esse modelo de interação natural. Em vez de um fluxo de controle rígido, os componentes emitem eventos quando algo significativo acontece. Outros componentes (ouvintes, gerenciadores) reagem a esses eventos. Isso traz vários benefícios-chave:
“`html
- Modularidade: Os componentes se tornam fracamente acoplados. Um executor de ferramenta não precisa saber quem o chamou ou o que acontecerá em seguida; ele simplesmente emite um evento como “tool_call_succeeded” ou “tool_call_failed.”
- Flexibilidade: É muito mais fácil adicionar novas funcionalidades ou modificar as existentes. Quer uma nova ferramenta? Basta adicionar um gerenciador que escuta um evento de intenção específica. Precisa registrar cada chamada LLM? Adicione um gravador que escuta “llm_response_received.”
- Resiliência: Se um componente falhar, é menos provável que derrube todo o sistema. Um evento pode ser refeito, ou um gerenciador alternativo pode lidá-lo. Você pode integrar filas para eventos que não podem ser processados.
- Concorrência: Vários eventos podem ser processados em paralelo, seja por diferentes gerenciadores, seja pelo mesmo gerenciador em diferentes instâncias de eventos. Isso é crucial para agentes que precisam gerenciar várias tarefas ao mesmo tempo.
- Observabilidade: O fluxo de eventos fornece um registro claro e auditável de tudo o que o agente faz. Você pode facilmente rastrear o fluxo de informações e decisões.
A Ideia Principal: Eventos, Despachantes e Gerenciadores
No coração de uma AOE, você precisa de três coisas:
- Eventos: Estruturas de dados simples que descrevem algo que ocorreu (por exemplo, `ToolCalled`, `LLMResponseReceived`, `UserQueryReceived`).
- Um Despachante de Eventos: Um mecanismo central que leva um evento e o direciona para todas as partes interessadas.
- Gerenciadores de Eventos: Funções ou classes que “ouvem” tipos específicos de eventos e executam uma lógica quando recebem um.
Vamos olhar um exemplo simplificado. Imagine nosso agente de pipeline de escrita. Em vez de uma enorme função, temos:
- Um evento `UserQueryReceived` (quando um novo e-mail chega).
- Um evento `LLMInputGenerated` (quando elaboramos um prompt para o LLM).
- Um evento `LLMResponseReceived` (quando o LLM retorna sua saída).
- Um evento `ToolCallRequested` (quando o LLM sugere usar uma ferramenta).
- Um evento `ToolCallSucceeded` / `ToolCallFailed` (após uma interação com uma ferramenta).
- Um evento `DraftResponseReady` (quando um rascunho está pronto para ser revisado).
Cada um desses eventos transmite dados relevantes – o conteúdo do e-mail, o prompt/resposta LLM, o nome da ferramenta e os argumentos, etc.
Elementos Básicos: Uma Abordagem Pythonica
Você não precisa de uma fila de mensagens pesada como Kafka para sistemas simples de agentes (embora para agentes distribuídos em produção, isso possa ser necessário!). Para um agente de processo único, um simples despachante de eventos em memória funciona maravilhas.
Passo 1: Defina Seus Eventos
Eu gosto de usar `dataclasses` para os eventos porque são claros e explícitos.
“““html
from dataclasses import dataclass
from typing import Any, Dict, Optional
@dataclass
class AgentEvent:
"""Classe base para todos os eventos de agente."""
timestamp: float # Adicione um carimbo de data/hora para ordenação e depuração
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
Note o `query_id`. Isso é crítico! Isso nos permite correlacionar eventos que pertencem à mesma interação ou tarefa do usuário. Sem isso, seu fluxo de eventos se torna uma bagunça caótica.
Etapa 2: Crie um Despachador de Eventos
É aqui que os eventos são direcionados. Um dicionário simples associando tipos de eventos a listas de manipuladores funciona bem.
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 uma função para gerenciar um tipo específico de evento."""
self._handlers[event_type].append(handler)
def dispatch(self, event: AgentEvent):
"""Enviar um evento para todos os manipuladores registrados."""
# Certifique-se de que o timestamp está definido, se não estiver já
if not hasattr(event, 'timestamp') or event.timestamp is None:
event.timestamp = time.time()
# Despachar para manipuladores específicos do tipo de evento
for handler in self._handlers[type(event)]:
try:
handler(event)
except Exception as e:
print(f"Erro no manipulador {handler.__name__} para o evento {type(event).__name__}: {e}")
# Eventualmente despachar um evento de erro aqui para robustez
# Despachar também para manipuladores registrados para o tipo base AgentEvent
# Isso permite registrar ou monitorar genericamente
for handler in self._handlers[AgentEvent]:
try:
handler(event)
except Exception as e:
print(f"Erro no manipulador genérico {handler.__name__} para o evento {type(event).__name__}: {e}")
Etapa 3: Defina Seus Manipuladores
Cada manipulador é uma função simples que toma um objeto de evento. Ele realiza sua tarefa específica e, crucialmente, pode despachar novos eventos.
Vamos esboçar alguns manipuladores para nosso agente de redação:
“““html
# Suponhamos que 'dispatcher' seja uma instância de EventDispatcher
# --- Gerenciador para a consulta inicial do usuário ---
def handle_user_query(event: UserQueryReceived):
print(f"[{event.query_id}] Consulta do usuário recebida: {event.content[:50]}...")
# Aqui, normalmente usaríamos um LLM para decidir a intenção inicial
# Para simplificar, vamos supor que sempre passe pelo LLM para a redação
prompt = f"Você é um assistente útil para um escritor freelance. Redija uma resposta inicial, educada à seguinte consulta do cliente, e sugira uma ação a seguir (por exemplo, 'schedule_call', 'search_knowledge_base'):\n\n{event.content}\n\nSaída em JSON com os campos 'draft_response' e 'suggested_action'."
# Emitir um evento para enviar ao LLM
dispatcher.dispatch(LLMRequestSent(
query_id=event.query_id,
model_name="gpt-4",
prompt=prompt,
metadata={"previous_event": type(event).__name__}
))
# --- Gerenciador para as respostas LLM ---
def handle_llm_response(event: LLMResponseReceived):
print(f"[{event.query_id}] Resposta LLM recebida: {event.response_text[:50]}...")
# Analisar a resposta LLM (isso seria mais robusto com 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"Ação sugerida pelo LLM: {action}"
))
if draft:
dispatcher.dispatch(DraftResponseReady(
query_id=event.query_id,
response_content=draft,
action_taken="drafted_initial_response"
))
if action == "schedule_call":
# Suponhamos que o LLM também tenha fornecido detalhes sobre a chamada, se necessário
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="calendar_scheduler",
tool_args={"client_email": "[email protected]", "duration": "30min"} # Espaço reservado
))
elif action == "search_knowledge_base":
# Suponhamos que o LLM tenha fornecido uma consulta de pesquisa
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="knowledge_base_search",
tool_args={"query": "artigos relacionados sobre agentes IA"} # Espaço reservado
))
except json.JSONDecodeError:
print(f"[{event.query_id}] Resposta LLM não é um JSON válido. Enviando para revisão humana.")
dispatcher.dispatch(FinalResponseReady(
query_id=event.query_id,
response_content="A análise da saída LLM falhou, necessidade de um humano. Resposta LLM original: " + event.response_text,
action_taken="human_review_needed"
))
# --- Gerenciador para solicitações de chamada de ferramentas ---
def handle_tool_call_request(event: ToolCallRequested):
print(f"[{event.query_id}] Solicitação de chamada de ferramenta: {event.tool_name} com args {event.tool_args}")
# Simular a execução da ferramenta
if event.tool_name == "calendar_scheduler":
# Em um sistema real, isso chamaria uma API real
print(f"Agendando uma chamada para {event.tool_args.get('client_email')}...")
time.sleep(1) # Simular um atraso de rede
if random.random() > 0.1: # Taxa de sucesso de 90%
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="Erro na API do calendário ou ocupado"
))
# ... outras ferramentas ...
# --- Gerenciador genérico de log ---
def log_all_events(event: AgentEvent):
print(f"LOG: {type(event).__name__} - {event.query_id} - {event.timestamp}")
# --- Registro dos gerenciadores ---
dispatcher = EventDispatcher()
dispatcher.register_handler(UserQueryReceived, handle_user_query)
dispatcher.register_handler(LLMResponseReceived, handle_llm_response)
dispatcher.register_handler(ToolCallRequested, handle_tool_call_request)
# ... outros gerenciadores para ToolCallSucceeded, ToolCallFailed, etc.
dispatcher.register_handler(AgentEvent, log_all_events) # Gerenciador genérico para todos os eventos
É um exemplo muito simplificado, mas você pode ver como cada elemento é independente. O `handle_user_query` não sabe *como* a consulta LLM será enviada, apenas que deve emitir um evento `LLMRequestSent`. Da mesma forma, o `handle_llm_response` não se preocupa com quem enviou o prompt original; ele simplesmente processa a resposta e decide o que fazer a seguir.
Simulação das chamadas LLM e de ferramentas
Para um sistema real, `LLMRequestSent` acionaria um componente que realmente chamaria a API LLM, e então emitiria `LLMResponseReceived` quando o resultado retornasse. É aí que `asyncio` ou um simples pool de threads podem ser úteis para chamadas LLM concorrentes ou execução de ferramentas sem bloquear o loop de eventos.
“`
import asyncio
import json
import random
import time
# ... (Definições dos eventos e EventDispatcher acima) ...
# API LLM fictícia
async def mock_llm_call(prompt: str) -> str:
print(f" [Mock LLM] Processando o prompt: {prompt[:80]}...")
await asyncio.sleep(random.uniform(1.0, 3.0)) # Simular a latência LLM
# Lógica fictícia muito básica para nosso caso de uso
if "schedule_call" in prompt:
return json.dumps({
"draft_response": "Obrigado pelo seu pedido! Eu adoraria discutir mais. Que tal agendarmos uma rápida chamada na próxima semana?",
"suggested_action": "schedule_call"
})
elif "search_knowledge_base" in prompt:
return json.dumps({
"draft_response": "Excelente pergunta! Eu redigi uma resposta e também consultei artigos relevantes.",
"suggested_action": "search_knowledge_base"
})
else:
return json.dumps({
"draft_response": "Obrigado por entrar em contato! Eu examinei seu pedido e escrevi uma resposta inicial.",
"suggested_action": "none"
})
# Componente Agent LLM (ouve LLMRequestSent, emite LLMResponseReceived)
async def llm_agent_component(event: LLMRequestSent, dispatcher: EventDispatcher):
response_text = await mock_llm_call(event.prompt)
# Em um sistema real, você analisaria para chamadas de ferramentas na resposta LLM
tool_calls = [] # Espaço reservado
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}
))
# Inscrever o manipulador assíncrono
dispatcher.register_handler(LLMRequestSent, lambda e: asyncio.create_task(llm_agent_component(e, dispatcher)))
# ... (outros manipuladores acima) ...
# Para executar um exemplo:
async def main():
query_id = "user_email_123"
dispatcher.dispatch(UserQueryReceived(
query_id=query_id,
content="Preciso de um artigo sobre agentes de IA orientados a eventos e uma chamada de acompanhamento.",
timestamp=time.time()
))
# Dar tempo para que os eventos sejam processados
await asyncio.sleep(10)
print("\n--- Processamento concluído para user_email_123 ---\n")
query_id_2 = "user_email_456"
dispatcher.dispatch(UserQueryReceived(
query_id=query_id_2,
content="Você pode resumir meus antigos artigos sobre arquiteturas de aprendizado profundo?",
timestamp=time.time()
))
await asyncio.sleep(10)
print("\n--- Processamento concluído para user_email_456 ---\n")
if __name__ == "__main__":
asyncio.run(main())
Eu introduzi `asyncio.create_task` para permitir que o `llm_agent_component` seja executado em paralelo com outros manipuladores ou dispatches subsequentes. É aqui que as arquiteturas orientadas a eventos realmente se destacam em termos de desempenho e reatividade dos agentes de IA.
Lições concretas para seu próximo projeto de agente
- Comece simples, pense em eventos: Mesmo para um pequeno agente, esboce os principais eventos que ocorrem. O que desencadeia o quê? Que informação deve ser transmitida?
- Defina esquemas de eventos claros: Use `dataclasses` ou modelos Pydantic para seus eventos. Isso garante consistência e facilita a depuração. Inclua sempre um `query_id` ou `correlation_id`.
- Separe as preocupações: Cada manipulador deve fazer uma coisa bem. Não tente comprimir muita lógica em um único manipulador. Se um manipulador precisar fazer uma chamada externa, ele deve emitir um evento de solicitação e esperar por um evento de resposta correspondente.
- Adoção da assíncronicidade: As interações dos agentes de IA (chamadas LLM, execução de ferramentas) são intrinsecamente assíncronas. Use `asyncio` ou um quadro semelhante para gerenciar essas operações em paralelo sem bloquear seu loop de eventos.
- Integrar o observável: Um registrador de eventos genérico (como meu `log_all_events`) é incrivelmente valioso. Você pode facilmente redirecionar esses eventos para um sistema de monitoramento ou simplesmente imprimi-los para desenvolvimento. Esse fluxo de eventos se torna o log do “processo de pensamento” interno do seu agente.
- Gestão de erros com eventos: Ao invés de um `try/except` profundamente aninhado, emita eventos `ErrorEvent` ou `ToolCallFailed`. Outros manipuladores podem então ouvir especificamente esses eventos para implementar uma lógica de re-tentativa, soluções alternativas ou pedidos de intervenção humana.
Mudar para um modelo orientado a eventos mudou completamente a minha forma de pensar sobre a construção de agentes. Isso me afastou da tentativa de antecipar cada caminho possível em um fluxo linear e me levou a construir um sistema que reage de forma inteligente ao seu ambiente e às suas próprias operações internas. É uma maneira mais resiliente, escalável e, francamente, mais agradável de construir agentes de IA complexos.
Experimente para seu próximo projeto de agente. Você pode se surpreender desvendando esse código spaghetti mais rápido do que imagina!
🕒 Published: