Oi pessoal, aqui é o Alex do agntai.net! É março de 2026, e tenho passado tempo demais pensando em como construímos agentes de IA. Especificamente, tenho lutado com o “código colante” – as coisas que conectam todas as saídas sofisticadas de LLM, chamadas de ferramentas e gerenciamento de estado. Todos nós já vimos as demonstrações impressionantes, certo? 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 callbacks, lógica condicional e atualizações de estado. Parece menos que construir um sistema inteligente e mais que gerenciar uma fábrica de espaguete muito complexa.
Então, hoje, quero falar sobre algo que tem ganhado força silenciosamente e, francamente, salvando minha sanidade: Arquiteturas Orientadas a Eventos para Agentes de IA. Não é um conceito novo em engenharia de software, de forma alguma, mas aplicá-lo com cuidado a agentes de IA, especialmente aqueles que orquestram múltiplas interações com LLM e ferramentas externas, parece um sopro de ar fresco. Esqueça o pensamento linear, passo a passo por um momento. Vamos pensar em sistemas reativos.
Minha Luta Pessoal com Monólitos de Agentes
Alguns meses atrás, estava trabalhando em um agente projetado para me ajudar a gerenciar meu fluxo de trabalho de redação freelance. A ideia era simples: ele monitoraria minha caixa de entrada em busca de novas consultas, redigiria respostas iniciais, sugeriria artigos relevantes do meu banco de dados de conhecimento e até ajudaria a agendar chamadas de acompanhamento. Parecia simples o suficiente.
Minha abordagem inicial era bastante típica: um loop principal. Receber email. Analisar email. Decidir ação (redigir, agendar, pesquisar). Chamar LLM. Processar saída do LLM. Chamar ferramenta (API de calendário, API de email, API de banco de dados de conhecimento). Atualizar estado interno. Repetir.
Começou bem, mas à medida que adicionava mais “inteligência” e mais ferramentas, tornou-se um pesadelo. E se a chamada da API de calendário falhasse? E se o LLM alucinasse um contato que não existia? E se eu precisasse pausar e pedir input humano para uma decisão crítica? Meu script de agente monolítico rapidamente se transformou em um labirinto aninhado de `if/else` com blocos de `try/except` em toda parte. Debugar era um pesadelo. Modificar uma parte frequentemente quebrava outra. Parecia que eu estava constantemente tampando vazamentos em um navio afundando.
Lembro de uma noite tarde, tentando descobrir por que meu agente continuava redigindo respostas para emails que já tinha processado. Descobriu-se que a atualização de estado para “email processado” estava acontecendo *depois* de uma possível nova execução do LLM em um caminho de falha. Era uma condição de corrida clássica em um sistema que não foi projetado para lidar com operações assíncronas e não determinísticas de forma elegante. Foi quando comecei a procurar uma maneira melhor.
Por Que Agentes Orientados a Eventos Fazem Sentido
Pense em como os humanos trabalham. Geralmente, não seguimos um script rígido e pré-definido para cada interação. Nós reagimos a coisas. Alguém faz uma pergunta – esse é um evento. Nós processamos e respondemos – esse é 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 (EDA) para agentes de IA reflete esse padrão de interação natural. Em vez de um fluxo de controle rígido, os componentes emitem eventos quando algo significativo acontece. Outros componentes (ouvintes, manipuladores) reagem a esses eventos. Isso traz vários benefícios principais:
- Modularidade: Os componentes tornam-se fracamente acoplados. Um executor de ferramenta não precisa saber quem o chamou ou o que acontecerá a seguir; ele apenas emite um evento como “tool_call_succeeded” ou “tool_call_failed.”
- Flexibilidade: É muito mais fácil adicionar novos recursos ou modificar os existentes. Quer uma nova ferramenta? Basta adicionar um manipulador que escuta por um evento de intenção específico. Precisa registrar cada chamada de LLM? Adicione um registrador que escuta por “llm_response_received.”
- Resiliência: Se um componente falhar, é menos provável que derrube todo o sistema. Um evento pode ser refeito ou um manipulador alternativo pode pegá-lo. Você pode incorporar filas de mensagens mortas para eventos que não podem ser processados.
- Concorrência: Muitos eventos podem ser processados em paralelo, seja por diferentes manipuladores ou pelo mesmo manipulador em diferentes instâncias de eventos. Isso é crucial para agentes que precisam gerenciar várias tarefas em andamento.
- Observabilidade: O fluxo de eventos fornece um registro claro e auditável de tudo o que o agente está fazendo. Você pode facilmente rastrear o fluxo de informações e decisões.
A Ideia Central: Eventos, Distribuidores e Manipuladores
No coração, uma EDA precisa de três coisas:
- Eventos: Estruturas de dados simples que descrevem algo que aconteceu (por exemplo, `ToolCalled`, `LLMResponseReceived`, `UserQueryReceived`).
- Um Disparador de Eventos: Um mecanismo central que pega um evento e o roteia para todas as partes interessadas.
- Manipuladores de Eventos: Funções ou classes que “ouvem” tipos específicos de eventos e executam alguma lógica quando recebem um.
Vejamos um exemplo simplificado. Imagine nosso agente de pipeline de escrita. Em vez de uma grande função, temos:
- Um evento `UserQueryReceived` (quando um novo e-mail chega).
- Um evento `LLMInputGenerated` (quando criamos um prompt para o LLM).
- Um evento `LLMResponseReceived` (quando o LLM envia sua saída).
- Um evento `ToolCallRequested` (quando o LLM sugere o uso de uma ferramenta).
- Um evento `ToolCallSucceeded` / `ToolCallFailed` (após uma interação com a ferramenta).
- Um evento `DraftResponseReady` (quando um rascunho está pronto para revisão).
Cada um desses eventos carrega dados relevantes – o conteúdo do e-mail, o prompt/resposta do LLM, nome da ferramenta e argumentos, etc.
Blocos de Construção: Uma Abordagem Pythonica
Você não precisa de uma fila de mensagens pesada como o Kafka para sistemas de agente simples (embora para produção, agentes distribuídos, você definitivamente possa precisar!). Para um agente de processo único, um simples disparador de eventos em memória funciona maravilhas.
Passo 1: Defina Seus Eventos
Gosto de usar `dataclasses` para eventos porque são limpos e explícitos.
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 timestamp 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! Ele nos permite correlacionar eventos pertencentes à mesma interação ou tarefa geral do usuário. Sem ele, seu fluxo de eventos se torna uma bagunça caótica.
Passo 2: Crie um Disparador de Eventos
É aqui que os eventos são roteados. Um dicionário simples que mapeia tipos de eventos para 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]):
"""Registre uma função para manipular um tipo específico de evento."""
self._handlers[event_type].append(handler)
def dispatch(self, event: AgentEvent):
"""Envie um evento para todos os manipuladores registrados."""
# Garantir que o timestamp esteja definido, se ainda não estiver
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}")
# Potencialmente despachar um evento de erro aqui para solidez
# Também despachar para manipuladores registrados para o tipo base AgentEvent
# Isso permite log ou monitoramento genérico
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}")
Passo 3: Defina Seus Manipuladores
Cada manipulador é uma função simples que recebe 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 escrita:
# Supondo que 'dispatcher' seja uma instância de EventDispatcher
# --- Manipulador para 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, tipicamente usaríamos um LLM para decidir a intenção inicial
# Para simplicidade, vamos assumir que sempre vai para o LLM para redação
prompt = f"Você é um assistente útil para um escritor freelancer. Redija uma resposta inicial e educada para a seguinte consulta do cliente e sugira uma ação de acompanhamento (por exemplo, 'schedule_call', 'search_knowledge_base'):\n\n{event.content}\n\nSaída em JSON com os campos 'draft_response' e 'suggested_action'."
# Despachar 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__}
))
# --- Manipulador para respostas do LLM ---
def handle_llm_response(event: LLMResponseReceived):
print(f"[{event.query_id}] Resposta do LLM recebida: {event.response_text[:50]}...")
# Analisar resposta do LLM (isso seria mais sólido 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":
# Suponha que o LLM também forneceu detalhes da chamada, se necessário
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="calendar_scheduler",
tool_args={"client_email": "[email protected]", "duration": "30min"} # Marcador
))
elif action == "search_knowledge_base":
# Suponha que o LLM forneceu a consulta de pesquisa
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="knowledge_base_search",
tool_args={"query": "artigos relacionados a agentes de IA"} # Marcador
))
except json.JSONDecodeError:
print(f"[{event.query_id}] Resposta do LLM não é um JSON válido. Enviando para revisão humana.")
dispatcher.dispatch(FinalResponseReady(
query_id=event.query_id,
response_content="Falha ao analisar a saída do LLM, precisa de humano. Resposta original do LLM: " + event.response_text,
action_taken="human_review_needed"
))
# --- Manipulador para solicitações de ferramentas ---
def handle_tool_call_request(event: ToolCallRequested):
print(f"[{event.query_id}] Chamada de ferramenta solicitada: {event.tool_name} com args {event.tool_args}")
# Simular execução da ferramenta
if event.tool_name == "calendar_scheduler":
# Em um sistema real, isso chamaria uma API real
print(f"Agendando chamada para {event.tool_args.get('client_email')}...")
time.sleep(1) # Simular 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 ...
# --- Manipulador genérico de logs ---
def log_all_events(event: AgentEvent):
print(f"LOG: {type(event).__name__} - {event.query_id} - {event.timestamp}")
# --- Registrar manipuladores ---
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 manipuladores para ToolCallSucceeded, ToolCallFailed, etc.
dispatcher.register_handler(AgentEvent, log_all_events) # Manipulador genérico para todos os eventos
Este é um exemplo muito simplificado, mas você pode ver como cada peça é independente. O `handle_user_query` não sabe *como* a solicitação do LLM será enviada, apenas que precisa emitir um evento `LLMRequestSent`. Da mesma forma, o `handle_llm_response` não se importa quem enviou o prompt original; ele apenas processa a resposta e decide o que fazer em seguida.
Simulando Chamadas de LLM e Ferramentas
Para um sistema real, `LLMRequestSent` acionaria um componente que realmente chama a API do LLM, e então despacharia `LLMResponseReceived` quando o resultado retornasse. É aqui que `asyncio` ou um simples pool de threads pode ser útil para chamadas concorrentes de LLM ou execuções de ferramentas sem bloquear o loop de eventos.
“`html
import asyncio
import json
import random
import time
# ... (Event definitions and EventDispatcher from above) ...
# Mock LLM API
async def mock_llm_call(prompt: str) -> str:
print(f" [Mock LLM] Processando prompt: {prompt[:80]}...")
await asyncio.sleep(random.uniform(1.0, 3.0)) # Simula latência do LLM
# Lógica de simulação muito básica para nosso caso de uso
if "schedule_call" in prompt:
return json.dumps({
"draft_response": "Obrigado pela sua consulta! Adoraria conversar mais. Que tal agendarmos uma ligação rápida na próxima semana?",
"suggested_action": "schedule_call"
})
elif "search_knowledge_base" in prompt:
return json.dumps({
"draft_response": "Ótima pergunta! Eu redigi uma resposta e também procurei alguns artigos relevantes.",
"suggested_action": "search_knowledge_base"
})
else:
return json.dumps({
"draft_response": "Obrigado por entrar em contato! Eu revisei seu pedido e redigi uma resposta inicial.",
"suggested_action": "none"
})
# Componente do Agente LLM (escuta LLMRequestSent, despacha LLMResponseReceived)
async def llm_agent_component(event: LLMRequestSent, dispatcher: EventDispatcher):
response_text = await mock_llm_call(event.prompt)
# Em um sistema real, você faria o parsing para chamadas de ferramenta da resposta do 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}
))
# Registre o manipulador assíncrono
dispatcher.register_handler(LLMRequestSent, lambda e: asyncio.create_task(llm_agent_component(e, dispatcher)))
# ... (outros manipuladores de cima) ...
# 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 ligação de acompanhamento.",
timestamp=time.time()
))
# Dê um tempo para que os eventos sejam processados
await asyncio.sleep(10)
print("\n--- Processamento completo para user_email_123 ---\n")
query_id_2 = "user_email_456"
dispatcher.dispatch(UserQueryReceived(
query_id=query_id_2,
content="Você pode resumir meus artigos anteriores sobre arquiteturas de aprendizado profundo?",
timestamp=time.time()
))
await asyncio.sleep(10)
print("\n--- Processamento completo para user_email_456 ---\n")
if __name__ == "__main__":
asyncio.run(main())
Introduzi `asyncio.create_task` para permitir que o `llm_agent_component` funcione em conjunto com outros manipuladores ou despachos subsequentes. É aqui que as arquiteturas orientadas a eventos realmente se destacam em desempenho e capacidade de resposta em agentes de IA.
Liçöes Ações para seu Próximo Projeto de Agente
- Comece Simples, Pense em Eventos: Mesmo para um pequeno agente, esboce os eventos chave que acontecem. O que desencadeia o quê? Que informações precisam ser passadas?
- Defina Esquemas de Evento Claros: Use `dataclasses` ou modelos Pydantic para seus eventos. Isso garante consistência e facilita a depuração. Sempre inclua um `query_id` ou `correlation_id`.
- Separe as Preocupações: Cada manipulador deve fazer uma coisa bem. Não tente enfiar muita lógica em um único manipulador. Se um manipulador precisa fazer uma chamada externa, ele deve despachar um evento de solicitação e esperar por um evento de resposta correspondente.
- Abrace a Assincronicidade: As interações do agente de IA (chamadas LLM, execução de ferramentas) são inerentemente assíncronas. Use `asyncio` ou uma estrutura similar para lidar com isso de forma concorrente, sem bloquear seu loop de eventos.
- Construa Observabilidade: Um logger de eventos genérico (como meu `log_all_events`) é incrivelmente valioso. Você pode facilmente enviar esses eventos para um sistema de monitoramento ou simplesmente imprimi-los para desenvolvimento. Este fluxo de eventos se torna o log do “processo de pensamento” interno do seu agente.
- Tratamento de Erros com Eventos: Em vez de profundamente aninhados `try/except`, despache `ErrorEvent` ou eventos `ToolCallFailed`. Outros manipuladores podem então ouvir especificamente esses eventos para implementar lógica de repetição, fallback ou solicitações de intervenção humana.
Mudar para um modelo orientado a eventos mudou completamente como eu penso sobre a construção de agentes. Isso me afastou de tentar antecipar cada caminho possível em um fluxo linear e em direção à construção de um sistema que reage inteligentemente ao seu ambiente e suas próprias operações internas. É uma maneira mais resiliente, escalável e, francamente, mais agradável de construir agentes de IA complexos.
Experimente em seu próximo projeto de agente. Você pode se surpreender ao desfazer aquele código espaguete mais rápido do que pensa!
🕒 Published: