Introdução ao Middleware de Agentes
A ascensão de agentes de IA sofisticados deu início a uma nova era no desenvolvimento de software. Essas entidades autônomas, capazes de raciocínio complexo, tomada de decisões e interação, estão se tornando centrais em muitas aplicações. No entanto, orquestrar seu comportamento, gerenciar seu estado e garantir seu funcionamento sólido muitas vezes requer mais do que uma simples invocação direta. É aqui que os padrões de middleware de agentes entram em cena. Semelhante ao middleware web tradicional, o middleware de agentes intercepta e processa requisições e respostas, mas dentro do contexto único do ciclo de vida de um agente, percepção, ação e comunicação.
O middleware de agentes serve como uma camada crucial entre a lógica central do agente e seu ambiente, ou entre diferentes componentes de um sistema multiagente. Ele fornece uma maneira estruturada de injetar preocupações transversais, melhorar capacidades, gerenciar estado e impor políticas sem bagunçar o código de tomada de decisão principal do agente. Neste mergulho profundo, vamos explorar padrões comuns de middleware de agentes, entender suas aplicações práticas e ilustrá-los com exemplos concretos, focando principalmente em frameworks baseados em Python ou implementações conceituais.
A Necessidade de Middleware de Agentes
Antes de explorar os padrões, vamos entender por que o middleware de agentes é indispensável:
- Separação de Preocupações: Os agentes frequentemente têm inteligência central (por exemplo, planejamento, raciocínio) e preocupações periféricas (por exemplo, registro, monitoramento, autenticação, transformação de dados). O middleware permite que essas preocupações sejam tratadas externamente.
- Modularidade e Reutilização: Funcionalidades comuns podem ser encapsuladas em componentes de middleware reutilizáveis.
- Extensibilidade: Novos recursos ou comportamentos podem ser adicionados aos agentes sem modificar sua lógica central.
- Solidez e Resiliência: O middleware pode lidar com erros, tentativas e interrupções de circuito para interações externas.
- Observabilidade: Registro centralizado, coleta de métricas e rastreamento se tornam muito mais fáceis.
- Segurança e Imposto de Políticas: Autorização, limitação de taxa e validação de entrada podem ser aplicadas de maneira consistente.
Padrões Comuns de Middleware de Agentes
Vamos categorizar os padrões de middleware de agentes com base em sua função principal e como eles interagem com o ciclo de vida do agente.
1. O Padrão Interceptor
O padrão Interceptor é talvez o mais fundamental e amplamente utilizado. Ele permite que você intercepte chamadas aos métodos de um agente ou suas interações com serviços externos, realizando pré-processamento antes da chamada e pós-processamento depois dela. Isso é análogo à Programação Orientada a Aspectos (AOP) ou ao middleware tradicional de requisição/resposta.
Exemplo Prático: Interceptor de Registro e Métricas
Imagine um agente que realiza ações com base em comandos do usuário. Queremos registrar cada ação realizada e medir seu tempo de execução.
import time
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class AgentAction:
def __init__(self, name, payload):
self.name = name
self.payload = payload
def __repr__(self):
return f"Action(name='{self.name}', payload={self.payload})"
class AgentResponse:
def __init__(self, success, result=None, error=None):
self.success = success
self.result = result
self.error = error
def __repr__(self):
return f"Response(success={self.success}, result={self.result}, error={self.error})"
class AgentCore:
def execute_action(self, action: AgentAction) -> AgentResponse:
logging.info(f"AgentCore: Executando ação '{action.name}' com payload {action.payload}")
if action.name == "search_web":
# Simulando busca na web
time.sleep(0.5)
return AgentResponse(success=True, result=f"Resultados encontrados para '{action.payload}'")
elif action.name == "send_email":
# Simulando envio de email
time.sleep(0.2)
if "@" in str(action.payload): # Validação simples
return AgentResponse(success=True, result=f"Email enviado para '{action.payload}'")
else:
return AgentResponse(success=False, error="Formato de email inválido")
else:
return AgentResponse(success=False, error=f"Ação desconhecida: {action.name}")
class LoggingMetricsInterceptor:
def __init__(self, next_handler):
self.next_handler = next_handler
def execute_action(self, action: AgentAction) -> AgentResponse:
start_time = time.perf_counter()
logging.info(f"Interceptor: Pré-processando ação '{action.name}'")
try:
response = self.next_handler.execute_action(action)
except Exception as e:
logging.error(f"Interceptor: Erro durante a ação '{action.name}': {e}")
response = AgentResponse(success=False, error=str(e))
end_time = time.perf_counter()
duration = (end_time - start_time) * 1000 # milissegundos
logging.info(f"Interceptor: Pós-processando ação '{action.name}'. Duração: {duration:.2f}ms. Sucesso: {response.success}")
# Em um sistema real, você enviaria métricas para Prometheus/Grafana etc.
return response
# Conectando o agente com o middleware
agent = LoggingMetricsInterceptor(AgentCore())
# Testes
print("\n--- Teste 1: Busca na Web Bem-Sucedida ---")
response1 = agent.execute_action(AgentAction("search_web", "notícias recentes de IA"))
print(f"Resposta Final: {response1}")
print("\n--- Teste 2: Envio de Email Bem-Sucedido ---")
response2 = agent.execute_action(AgentAction("send_email", "[email protected]"))
print(f"Resposta Final: {response2}")
print("\n--- Teste 3: Falha no Envio de Email (Erro de Validação) ---")
response3 = agent.execute_action(AgentAction("send_email", "email-inválido"))
print(f"Resposta Final: {response3}")
print("\n--- Teste 4: Ação Desconhecida ---")
response4 = agent.execute_action(AgentAction("unknown_task", "dados"))
print(f"Resposta Final: {response4}")
Neste exemplo, LoggingMetricsInterceptor envolve AgentCore. Qualquer chamada para execute_action passa primeiro pelo interceptor, que registra, mede o tempo, depois passa o controle para o próximo manipulador (AgentCore), e finalmente processa a resposta.
2. O Padrão Cadeia de Responsabilidade
O padrão Cadeia de Responsabilidade permite que múltiplos manipuladores (componentes de middleware) processem uma requisição sequencialmente. Cada manipulador decide se deve processar a requisição, passá-la para o próximo manipulador na cadeia ou interromper o processamento. Isso é ideal para cenários onde múltiplas condições ou transformações podem se aplicar à entrada ou saída de um agente.
Exemplo Prático: Cadeia de Validação e Transformação da Entrada
Considere um agente que recebe comandos em linguagem natural. Antes que o agente central processe o comando, podemos querer validar a entrada, sanitizá-la ou traduzir para um formato estruturado.
class Command:
def __init__(self, original_text: str, processed_data: dict = None):
self.original_text = original_text
self.processed_data = processed_data if processed_data is not None else {}
self.is_valid = True
self.error_message = None
def __repr__(self):
return f"Command(original='{self.original_text}', processed={self.processed_data}, valid={self.is_valid}, error='{self.error_message}')"
class AgentRequestHandler:
def handle_command(self, command: Command) -> Command:
raise NotImplementedError
class InputSanitizer(AgentRequestHandler):
def __init__(self, next_handler: AgentRequestHandler = None):
self.next_handler = next_handler
def handle_command(self, command: Command) -> Command:
if not command.is_valid:
return command # Para se já estiver inválido
# Sanitização simples: remove espaços no início/fim, converte para minúsculas
command.processed_data['sanitized_text'] = command.original_text.strip().lower()
logging.info(f"Sanitizer: Sanitizado '{command.original_text}' para '{command.processed_data['sanitized_text']}'")
if self.next_handler:
return self.next_handler.handle_command(command)
return command
class CommandValidator(AgentRequestHandler):
def __init__(self, next_handler: AgentRequestHandler = None):
self.next_handler = next_handler
def handle_command(self, command: Command) -> Command:
if not command.is_valid:
return command
sanitized_text = command.processed_data.get('sanitized_text', command.original_text)
if len(sanitized_text) < 5:
command.is_valid = False
command.error_message = "O comando é muito curto."
logging.warning(f"Validator: Comando inválido '{sanitized_text}' - muito curto.")
return command # Para o processamento se inválido
logging.info(f"Validator: Comando '{sanitized_text}' passou a validação de comprimento.")
if self.next_handler:
return self.next_handler.handle_command(command)
return command
class IntentRecognizer(AgentRequestHandler):
def __init__(self, next_handler: AgentRequestHandler = None):
self.next_handler = next_handler
def handle_command(self, command: Command) -> Command:
if not command.is_valid:
return command
sanitized_text = command.processed_data.get('sanitized_text', command.original_text)
if "schedule" in sanitized_text or "book" in sanitized_text:
command.processed_data['intent'] = 'schedule_event'
command.processed_data['params'] = {'topic': 'meeting'}
elif "weather" in sanitized_text:
command.processed_data['intent'] = 'get_weather'
command.processed_data['params'] = {'location': 'current'}
else:
command.processed_data['intent'] = 'unknown'
logging.info(f"IntentRecognizer: Intenção detectada '{command.processed_data['intent']}' para '{sanitized_text}'")
if self.next_handler:
return self.next_handler.handle_command(command)
return command
class AgentCoreProcessor(AgentRequestHandler):
def handle_command(self, command: Command) -> Command:
if not command.is_valid:
logging.error(f"Core: Não é possível processar comando inválido: {command.error_message}")
return command
logging.info(f"Core: Processando comando com intenção '{command.processed_data.get('intent')}' e parâmetros {command.processed_data.get('params')}")
command.processed_data['core_result'] = f"Executado {command.processed_data.get('intent')} com {command.processed_data.get('params')}"
return command
# Construindo a cadeia
core_processor = AgentCoreProcessor()
intent_recognizer = IntentRecognizer(core_processor)
validator = CommandValidator(intent_recognizer)
sanitizer = InputSanitizer(validator)
# O ponto de entrada para comandos
agent_entry_point = sanitizer
# Testando comandos
print("\n--- Teste 1: Comando de agendamento válido ---")
cmd1 = Command(" Por favor, agende uma reunião para mim ")
processed_cmd1 = agent_entry_point.handle_command(cmd1)
print(f"Comando Processado Final: {processed_cmd1}")
print("\n--- Teste 2: Comando de clima válido ---")
cmd2 = Command("Como está o clima?")
processed_cmd2 = agent_entry_point.handle_command(cmd2)
print(f"Comando Processado Final: {processed_cmd2}")
print("\n--- Teste 3: Comando curto inválido ---")
cmd3 = Command("oi")
processed_cmd3 = agent_entry_point.handle_command(cmd3)
print(f"Comando Processado Final: {processed_cmd3}")
print("\n--- Teste 4: Comando desconhecido ---")
cmd4 = Command("me conte uma piada")
processed_cmd4 = agent_entry_point.handle_command(cmd4)
print(f"Comando Processado Final: {processed_cmd4}")
Aqui, um Command objeto passa por uma cadeia: InputSanitizer -> CommandValidator -> IntentRecognizer -> AgentCoreProcessor. Cada componente modifica o objeto Command ou define seu flag is_valid. Se um componente invalidar o comando, os componentes subsequentes podem parar o processamento de forma elegante.
3. O Padrão Adapter para Ferramentas/APIs Externas
Embora não seja estritamente um middleware no sentido de interceptação de requisições-respostas, o padrão Adapter é crucial para permitir que agentes interajam com diversas ferramentas externas e APIs de maneira padronizada. Um adaptador envolve um serviço de terceiros, fornecendo uma interface consistente para o agente usar, abstraindo os detalhes da API externa.
Exemplo Prático: Acesso Unificado a Ferramentas
Um agente pode precisar chamar uma API de clima, uma API de calendário e um motor de busca. Cada um tem uma interface diferente. Os adaptadores normalizam essas interações.
import requests
import json
class ToolAdapter:
def execute(self, tool_name: str, params: dict) -> dict:
raise NotImplementedError
class WeatherAPIAdapter(ToolAdapter):
BASE_URL = "https://api.weatherapi.com/v1"
API_KEY = "YOUR_WEATHER_API_KEY" # Substitua pela chave real
def execute(self, tool_name: str, params: dict) -> dict:
if tool_name == "get_current_weather":
location = params.get("location", "London")
try:
response = requests.get(f"{self.BASE_URL}/current.json?key={self.API_KEY}&q={location}")
response.raise_for_status() # Levanta um HTTPError para respostas ruins (4xx ou 5xx)
data = response.json()
return {
"temperature_c": data['current']['temp_c'],
"condition": data['current']['condition']['text'],
"location": data['location']['name']
}
except requests.exceptions.RequestException as e:
logging.error(f"Erro na API de clima: {e}")
return {"error": str(e)}
return {"error": f"Ferramenta de clima desconhecida: {tool_name}"}
class CalendarAPIAdapter(ToolAdapter):
def execute(self, tool_name: str, params: dict) -> dict:
if tool_name == "create_event":
title = params.get("title")
start_time = params.get("start_time")
end_time = params.get("end_time")
logging.info(f"Calendário: Criando evento '{title}' de {start_time} a {end_time}")
# Simula chamada da API
time.sleep(0.1)
return {"status": "success", "event_id": "cal_123", "title": title}
elif tool_name == "list_events":
date = params.get("date")
logging.info(f"Calendário: Listando eventos para {date}")
# Simula chamada da API
time.sleep(0.1)
return {"status": "success", "events": [{"title": "Reunião da Equipe", "time": "10:00"}]}
return {"error": f"Ferramenta de calendário desconhecida: {tool_name}"}
class AgentToolbox:
def __init__(self):
self._adapters = {}
def register_adapter(self, adapter_name: str, adapter: ToolAdapter):
self._adapters[adapter_name] = adapter
logging.info(f"Toolbox: Adaptador '{adapter_name}' registrado")
def use_tool(self, adapter_name: str, tool_name: str, params: dict) -> dict:
adapter = self._adapters.get(adapter_name)
if not adapter:
return {"error": f"Nenhum adaptador registrado para '{adapter_name}'"}
logging.info(f"Toolbox: Usando ferramenta '{tool_name}' via adaptador '{adapter_name}' com parâmetros {params}")
return adapter.execute(tool_name, params)
# Inicializando a caixa de ferramentas do agente
agent_toolbox = AgentToolbox()
agent_toolbox.register_adapter("weather", WeatherAPIAdapter())
agent_toolbox.register_adapter("calendar", CalendarAPIAdapter())
# Agente usando sua caixa de ferramentas
print("\n--- Agente usando Ferramenta de Clima ---")
weather_info = agent_toolbox.use_tool("weather", "get_current_weather", {"location": "New York"})
print(f"Informações do Clima: {weather_info}")
print("\n--- Agente usando Ferramenta de Calendário (Criar Evento) ---")
calendar_event = agent_toolbox.use_tool("calendar", "create_event", {"title": "Revisão do Projeto", "start_time": "2023-10-27 14:00", "end_time": "2023-10-27 15:00"})
print(f"Evento do Calendário: {calendar_event}")
print("\n--- Agente usando Ferramenta de Calendário (Listar Eventos) ---")
list_events = agent_toolbox.use_tool("calendar", "list_events", {"date": "2023-10-27"})
print(f"Eventos Listados: {list_events}")
print("\n--- Agente tentando usar ferramenta não registrada ---")
unknown_tool = agent_toolbox.use_tool("search_engine", "google_search", {"query": "tendências de IA"})
print(f"Resultado da Ferramenta Desconhecida: {unknown_tool}")
Aqui, AgentToolbox atua como um registro central para instâncias de ToolAdapter. O agente não precisa conhecer os detalhes de como chamar WeatherAPIAdapter ou CalendarAPIAdapter; ele apenas solicita uma ferramenta pelo nome e fornece os parâmetros. Cada adaptador, então, traduz esse pedido genérico nas chamadas de API específicas necessárias.
4. O Padrão Registry/Service Locator
O padrão Registry ou Service Locator é comumente utilizado para fornecer aos agentes acesso a diversos serviços, capacidades ou outros agentes dentro de um sistema multiagente. Em vez de codificar dependências, os agentes consultam um registro central para descobrir e obter referências aos componentes necessários durante a execução. Isso aumenta a flexibilidade e o desacoplamento.
Exemplo Prático: Descoberta Dinâmica de Capacidades do Agente
Imagine um agente precisando de uma capacidade específica, como sumarização de texto ou geração de imagens. Ele não deve precisar saber qual serviço específico fornece isso, apenas que a capacidade existe.
class Capability:
def execute(self, data: str) -> str:
raise NotImplementedError
class TextSummarizer(Capability):
def execute(self, text: str) -> str:
logging.info(f"Resumindo texto: '{text[:30]}...' ")
# Simula chamada LLM ou lógica de resumo
time.sleep(0.3)
return f"Resumo de '{text[:20]}...': Esta é uma versão concisa."
class ImageGenerator(Capability):
def execute(self, prompt: str) -> str:
logging.info(f"Gerando imagem para o prompt: '{prompt}'")
# Simula chamada de API para geração de imagem
time.sleep(0.7)
return f"URL da imagem para '{prompt}': https://image.gen/id-123"
class CapabilityRegistry:
def __init__(self):
self._capabilities = {}
def register_capability(self, name: str, capability: Capability):
if name in self._capabilities:
logging.warning(f"Capacidade '{name}' já registrada. Sobrescrevendo.")
self._capabilities[name] = capability
logging.info(f"Registro: Capacidade '{name}' registrada.")
def get_capability(self, name: str) -> Capability:
capability = self._capabilities.get(name)
if not capability:
logging.error(f"Registro: Capacidade '{name}' não encontrada.")
raise ValueError(f"Capacidade '{name}' não encontrada.")
return capability
class Agent:
def __init__(self, registry: CapabilityRegistry):
self.registry = registry
def process_request(self, request_type: str, data: str) -> str:
try:
capability = self.registry.get_capability(request_type)
result = capability.execute(data)
return f"Agente processou '{request_type}': {result}"
except ValueError as e:
return f"Agente falhou ao processar '{request_type}': {e}"
except Exception as e:
return f"Agente encontrou um erro inesperado para '{request_type}': {e}"
# Configurando o registro
registry = CapabilityRegistry()
registry.register_capability("summarize", TextSummarizer())
registry.register_capability("generate_image", ImageGenerator())
# Cria um agente com acesso ao registro
agent_app = Agent(registry)
# Agente usa as capacidades
print("\n--- Agente solicitando resumo ---")
summary_result = agent_app.process_request("summarize", "A rápida raposa marrom salta sobre o cachorro preguiçoso. Este é um pangrama clássico usado para exibir todas as letras do alfabeto.")
print(summary_result)
print("\n--- Agente solicitando geração de imagem ---")
image_result = agent_app.process_request("generate_image", "uma cidade futurista ao pôr do sol")
print(image_result)
print("\n--- Agente solicitando capacidade desconhecida ---")
unknown_result = agent_app.process_request("translate", "hello world")
print(unknown_result)
O CapabilityRegistry atua como um localizador de serviços. O Agent não instancia diretamente TextSummarizer ou ImageGenerator; ele solicita ao registro uma capacidade pelo seu nome lógico. Isso permite que as capacidades sejam trocadas, atualizadas ou adicionadas sem mudar a lógica central do agente.
Combinando Padrões de Middleware
Em sistemas de agentes do mundo real, esses padrões são frequentemente combinados. Por exemplo, um comando de usuário recebido pode primeiro passar por uma Cadeia de Responsabilidade para validação e reconhecimento de intenção. A intenção identificada pode, então, acionar uma ação que utiliza o Registro/Localizador de Serviço para encontrar um Adaptador apropriado para uma ferramenta externa. A execução dessa ferramenta pode ser envolvida por um Interceptor para registro e tratamento de erros.
Exemplo: Um Fluxo de Interação de Agente em Múltiplas Camadas
Vamos delinear brevemente como isso pode parecer:
# 1. Solicitação Recebida (por exemplo, de uma interface de chat do usuário)
user_input = "Por favor, agende uma reunião sobre os resultados do Q4 para amanhã às 15h."
# 2. Cadeia de Responsabilidade para Pré-processamento
# InputSanitizer -> CommandValidator -> IntentRecognizer
command_object = Command(user_input)
processed_command = agent_entry_point.handle_command(command_object) # Usa a cadeia do exemplo anterior
if processed_command.is_valid and processed_command.processed_data.get('intent') == 'schedule_event':
# 3. A lógica central do agente decide usar uma ferramenta
intent_params = processed_command.processed_data.get('params', {})
# 4. Usar Registro/Localizador de Serviço para obter o adaptador apropriado
# O agente sabe que precisa de um adaptador 'calendar' para 'schedule_event'
# 5. A execução da ferramenta em si é envolvida por um Interceptor
# (Imagine agent_toolbox.use_tool sendo envolvido por um genérico ToolCallInterceptor)
# Para simplicidade, chamaremos a caixa de ferramentas diretamente aqui, mas imagine que é feita através de proxy.
# Simula o parsing do horário a partir da entrada original
event_title = intent_params.get('topic', 'Reunião Genérica')
start_time_str = "2023-10-28 15:00" # Extraído de user_input por um IntentRecognizer mais sofisticado
end_time_str = "2023-10-28 16:00"
print("\n--- Agente orquestrando o uso da ferramenta ---")
tool_call_result = agent_toolbox.use_tool(
"calendar",
"create_event",
{"title": event_title, "start_time": start_time_str, "end_time": end_time_str}
)
print(f"Resultado da Chamada de Ferramenta: {tool_call_result}")
else:
print(f"Agente não conseguiu processar a solicitação: {processed_command.error_message or 'Intenção inválida'}")
Esse fluxo demonstra como diferentes padrões de middleware podem ser compostos para criar uma arquitetura de agente sólida e mantível.
Conclusão
Padrões de middleware para agentes são essenciais para construir sistemas de agentes de IA escaláveis, sólidos e mantíveis. Ao aplicar padrões como Interceptor, Cadeia de Responsabilidade, Adaptador e Registro/Localizador de Serviço, os desenvolvedores podem gerenciar de forma eficaz preocupações transversais, integrar funcionalidades diversas e abstrair complexidades. Esses padrões promovem modularidade, reutilização e extensibilidade, permitindo que os agentes evoluam e interajam com seus ambientes de maneira mais inteligente e confiável. À medida que os agentes de IA se tornam mais sofisticados e integrados em nossas vidas diárias, uma profunda compreensão e aplicação prática desses padrões de middleware será crucial para o sucesso.
🕒 Published:
Related Articles
- Guida alla selezione della libreria per agenti AI
- Bibliotecas Essenciais para Agentes de IA: Armadilhas Comuns e Soluções Práticas
- Muster der Middleware-Agenten im Jahr 2026: Praktische Architekturen für autonome Systeme
- CrewAI vs AutoGen : Un confronto completo tra i framework per i sistemi di IA multi-agente