Introdução ao Agent Middleware
A crescente sofisticação dos agentes de IA inaugurou uma nova era no desenvolvimento de software. Essas entidades autônomas, capazes de raciocínios complexos, decisões e interações, estão se tornando centrais em muitas aplicações. No entanto, orquestrar seu comportamento, gerenciar seu estado e garantir um funcionamento sólido frequentemente requer mais do que uma simples invocação direta. É aqui que entram em cena os modelos de agent middleware. Semelhante ao middleware web tradicional, o agent middleware intercepta e trata solicitações e respostas, mas no contexto único do ciclo de vida de um agente, da percepção, da ação e da comunicação.
O agent middleware desempenha um papel crucial entre a lógica principal do agente e seu ambiente, ou entre os diferentes componentes de um sistema multi-agente. Ele fornece uma maneira estruturada de injetar preocupações transversais, melhorar capacidades, gerenciar estado e fazer cumprir políticas sem sobrecarregar o código decisional principal do agente. Neste aprofundamento, exploraremos os modelos comuns de agent middleware, entenderemos suas aplicações práticas e os ilustraremos com exemplos concretos, concentrando-nos principalmente em frameworks baseados em Python ou implementações conceituais.
A Necessidade do Agent Middleware
Antes de explorarmos os modelos, vamos entender por que o agent middleware é 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 gerenciar essas preocupações externamente.
- Modularidade e Reutilização: Funcionalidades comuns podem ser encapsuladas em componentes middleware reutilizáveis.
- Extensibilidade: Novas funcionalidades ou comportamentos podem ser adicionados aos agentes sem modificar sua lógica central.
- Solidez e Resiliência: O middleware pode gerenciar erros, tentativas de reexecução e interrupção de circuito para interações externas.
- Observabilidade: O registro centralizado, a coleta de métricas e o rastreamento se tornam muito mais simples.
- Segurança e Aplicação de Políticas: Autorização, limitação de taxa e validação de entrada podem ser aplicadas de forma consistente.
Modelos Comuns de Agent Middleware
Classificaremos os modelos de agent middleware com base em sua função principal e em como interagem com o ciclo de vida do agente.
1. O Modelo Interceptor
O modelo Interceptor é talvez o mais fundamental e amplamente utilizado. Ele permite interceptar chamadas aos métodos de um agente ou suas interações com serviços externos, executando pré-processamento antes da chamada e pós-processamento depois dela. Isso é análogo à Programação Orientada a Aspectos (AOP) ou ao middleware tradicional para requisição/resposta.
Exemplo Prático: Interceptor de Registro e Métricas
Imagine um agente que executa ações com base em solicitações dos usuários. Queremos registrar cada ação executada e medir seu tempo de execução.
“`html
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 ação '{action.name}' com payload {action.payload}")
if action.name == "search_web":
# Simula pesquisa na web
time.sleep(0.5)
return AgentResponse(success=True, result=f"Resultados encontrados para '{action.payload}'")
elif action.name == "send_email":
# Simula 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é-processamento da 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-processamento da 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())
# Casos de teste
print("\n--- Teste 1: Pesquisa na web bem-sucedida ---")
response1 = agent.execute_action(AgentAction("search_web", "últimas notícias 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: Envio de email não bem-sucedido (Erro de validação) ---")
response3 = agent.execute_action(AgentAction("send_email", "bad-email"))
print(f"Resposta final: {response3}")
print("\n--- Teste 4: Ação desconhecida ---")
response4 = agent.execute_action(AgentAction("unknown_task", "data"))
print(f"Resposta final: {response4}")
Neste exemplo, LoggingMetricsInterceptor envolve AgentCore. Qualquer chamada para execute_action passa primeiro pelo interceptor, que registra, mede o tempo, então passa o controle para o próximo manipulador (AgentCore), e finalmente processa a resposta.
2. O Modelo Chain of Responsibility
O modelo Chain of Responsibility permite que múltiplos manipuladores (componentes middleware) processem uma solicitação em sequência. Cada manipulador decide se deve processar a solicitaçã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 ser aplicadas à 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 core processe o comando, poderíamos querer validar a entrada, sanitizá-la ou traduzi-la em um formato estruturado.
“““html
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á não for válido
# Sanitização simples: remove espaços iniciais/finais, converte para minúsculas
command.processed_data['sanitized_text'] = command.original_text.strip().lower()
logging.info(f"Sanitizer: Sanificado '{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 não válido '{sanitized_text}' - muito curto.")
return command # Para o processamento se não for válido
logging.info(f"Validator: Comando '{sanitized_text}' passou na 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: Impossível processar o comando inválido: {command.error_message}")
return command
logging.info(f"Core: Processando o 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
# Construção da cadeia
core_processor = AgentCoreProcessor()
intent_recognizer = IntentRecognizer(core_processor)
validator = CommandValidator(intent_recognizer)
sanitizer = InputSanitizer(validator)
# O ponto de entrada para os comandos
agent_entry_point = sanitizer
# Comandos de teste
print("\n--- Teste 1: Comando de agendamento válido ---")
cmd1 = Command(" Por favor, programe 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 tempo?")
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("conte-me uma piada")
processed_cmd4 = agent_entry_point.handle_command(cmd4)
print(f"Comando Processado Final: {processed_cmd4}")
Aqui, um Command objeto atravessa 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 seguintes podem interromper o processamento sem problemas.
3. O Padrão Adapter para Ferramentas/API Externas
“`
Embora não seja estritamente middleware no sentido de interceptação de requisições e respostas, o padrão Adapter é crucial para permitir que agentes interajam com diferentes ferramentas externas e APIs de forma padronizada. Um adaptador encapsula um serviço de terceiros, fornecendo uma interface coerente para o agente utilizar, abstraindo as especificações da API externa.
Exemplo Prático: Acesso Unificado às Ferramentas
Um agente pode precisar chamar uma API de clima, uma API de calendário e um mecanismo de busca. Cada uma 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() # Gera uma exceção HTTPError para respostas inválidas (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 API 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: Criação do evento '{title}' de {start_time} a {end_time}")
# Simula uma chamada 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: Lista de eventos para {date}")
# Simula uma chamada API
time.sleep(0.1)
return {"status": "success", "events": [{"title": "Sincronizaçã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 registrado '{adapter_name}'")
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: Utilizando a ferramenta '{tool_name}' via adaptador '{adapter_name}' com parâmetros {params}")
return adapter.execute(tool_name, params)
# Inicializa a caixa de ferramentas do agente
agent_toolbox = AgentToolbox()
agent_toolbox.register_adapter("weather", WeatherAPIAdapter())
agent_toolbox.register_adapter("calendar", CalendarAPIAdapter())
# O agente utiliza sua caixa de ferramentas
print("\n--- Agente utilizando a Ferramenta de Clima ---")
weather_info = agent_toolbox.use_tool("weather", "get_current_weather", {"location": "New York"})
print(f"Informações de Clima: {weather_info}")
print("\n--- Agente utilizando a 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 de Calendário: {calendar_event}")
print("\n--- Agente utilizando a Ferramenta de Calendário (Lista de Eventos) ---")
list_events = agent_toolbox.use_tool("calendar", "list_events", {"date": "2023-10-27"})
print(f"Eventos Listados: {list_events}")
print("\n--- Agente tentando utilizar uma ferramenta não registrada ---")
unknown_tool = agent_toolbox.use_tool("search_engine", "google_search", {"query": "tendências IA"})
print(f"Resultado da Ferramenta Desconhecida: {unknown_tool}")
Aqui, AgentToolbox atua como um registro central para as instâncias de ToolAdapter. O agente não precisa conhecer as especificações sobre como chamar WeatherAPIAdapter ou CalendarAPIAdapter; ele deve apenas solicitar uma ferramenta pelo nome e fornecer parâmetros. Cada adaptador, então, traduz essa solicitação genérica nas chamadas API específicas requisitadas.
4. O Padrão Registry/Service Locator
O padrão Registry ou Service Locator é comumente utilizado para fornecer aos agentes acesso a vários serviços, capacidades ou outros agentes dentro de um sistema multi-agente. Em vez de codificar as dependências, os agentes consultam um registro central para descobrir e obter referências aos componentes necessários em tempo de execução. Isso aumenta a flexibilidade e o desacoplamento.
Exemplo Prático: Descoberta Dinâmica das Capacidades do Agente
Imagine um agente que necessite de uma capacidade específica, como o resumo de texto ou a geração de imagens. Não deveria 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 o texto: '{text[:30]}...' ")
# Simulate LLM call or summarization logic
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}'")
# Simulate image generation API call
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"O agente processou '{request_type}': {result}"
except ValueError as e:
return f"O agente não conseguiu processar '{request_type}': {e}"
except Exception as e:
return f"O agente encontrou um erro inesperado para '{request_type}': {e}"
# Configura 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)
# O agente utiliza as capacidades
print("\n--- O agente solicita um resumo ---")
summary_result = agent_app.process_request("summarize", "A rápida raposa marrom salta sobre o cão preguiçoso. Este é um clássico pangrama usado para mostrar todas as letras do alfabeto.")
print(summary_result)
print("\n--- O agente solicita a geração de uma imagem ---")
image_result = agent_app.process_request("generate_image", "uma cidade futurista ao pôr do sol")
print(image_result)
print("\n--- O agente solicita uma capacidade desconhecida ---")
unknown_result = agent_app.process_request("translate", "hello world")
print(unknown_result)
O CapabilityRegistry atua como um serviço de localização. O Agent não instancia diretamente o TextSummarizer ou ImageGenerator; ele solicita ao registro uma capacidade através de seu nome lógico. Isso permite substituir, atualizar ou adicionar capacidades sem modificar a lógica central do agente.
Combinação de Padrões Middleware
Nos agentes reais, esses padrões são frequentemente combinados. Por exemplo, um comando de usuário que chega pode passar primeiro por uma Cadeia de Responsabilidade para validação e reconhecimento da intenção. A intenção identificada pode então acionar uma ação que usa o Registro/Service Locator para encontrar um Adaptador apropriado para uma ferramenta externa. A execução dessa ferramenta pode então ser envolvida por um Interceptor para logging e gerenciamento de erros.
Exemplo: Um Fluxo de Interação do Agente em Múltiplos Níveis
Esboçamos brevemente como poderia parecer:
# 1. Solicitação Recebida (ex. de uma interface de chat do usuário)
user_input = "Por favor, programe uma reunião sobre os resultados do Q4 para amanhã às 15:00."
# 2. Cadeia de Responsabilidade para o 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. Usa o Registro/Service Locator 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 é envolta por um Interceptor
# (Imagine agent_toolbox.use_tool sendo envolto por um genérico ToolCallInterceptor)
# Para simplicidade, chamaremos diretamente a toolbox aqui, mas imagine que seja proxyado.
# Simula a análise do tempo 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--- O agente orquestra 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 da Ferramenta: {tool_call_result}")
else:
print(f"O agente falhou em processar a solicitação: {processed_command.error_message or 'Intenção não válida'}")
Este fluxo demonstra como diferentes padrões de middleware podem ser compostos para criar uma arquitetura para agentes sólida e manutenível.
Conclusão
Os padrões de middleware para agentes são essenciais para construir sistemas de agentes de IA escaláveis, robustos e manuteníveis. Aplicando padrões como Interceptor, Cadeia de Responsabilidade, Adaptador e Registro/Service Locator, os desenvolvedores podem gerenciar efetivamente preocupações transversais, integrar funcionalidades diversas e abstrair complexidade. 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 cotidianas, uma compreensão profunda e aplicação prática desses padrões de middleware será fundamental para o sucesso.
🕒 Published: