Introdução ao Agente Middleware
O surgimento de agentes AI sofisticados marcou o início de uma nova era no desenvolvimento de software. Essas entidades autônomas, capazes de raciocínio complexo, tomada de decisão e interação, estão se tornando centrais em muitas aplicações. No entanto, orquestrar seu comportamento, gerenciar seu estado e garantir seu bom funcionamento muitas vezes requer mais do que uma simples invocação direta. É aqui que entram os modelos de agente middleware. Semelhante aos middlewares web tradicionais, o agente middleware intercepta e processa as solicitações e respostas, mas no contexto único do ciclo de vida de um agente, desde sua percepção, ação até sua comunicação.
O agente middleware serve como uma camada crucial entre a lógica central do agente e seu ambiente, ou entre diferentes componentes de um sistema multi-agentes. Ele fornece uma forma estruturada de injetar preocupações transversais, melhorar as capacidades, gerenciar o estado e fazer valer políticas sem sobrecarregar o código principal de tomada de decisão do agente. Nesta exploração aprofundada, vamos examinar os modelos de agente middleware comuns, entender suas aplicações práticas e ilustrá-los com exemplos concretos, focando principalmente em frameworks ou implementações conceituais baseados em Python.
A Necessidade do Agente Middleware
Antes de explorarmos os modelos, vamos entender por que o agente middleware é imprescindível:
- Separação de Preocupações: Os agentes muitas vezes possuem uma 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 de forma externa.
- 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, re-tentativas e circuit breaking para interações externas.
- Observabilidade: O registro centralizado, a coleta de métricas e o rastreamento se tornam muito mais fáceis.
- Segurança e Aplicação de Políticas: A autorização, limitação de taxa e validação de entradas podem ser aplicadas de maneira consistente.
Modelos Comuns de Agente Middleware
Vamos classificar os modelos de agente middleware com base em sua função principal e na interação com o ciclo de vida do agente.
1. O Modelo do Interceptor
O modelo do Interceptor é talvez o mais fundamental e amplamente utilizado. Ele permite que você intercepte as chamadas aos métodos de um agente ou suas interações com serviços externos, realizando um pré-processamento antes da chamada e um pós-processamento depois. Isso é análogo à Programação Orientada a Aspecto (POA) ou ao middleware tradicional de solicitação/resposta.
Exemplo Prático: Interceptor de Registro e Métricas
Imagine um agente que executa 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":
# Simular busca na web
time.sleep(0.5)
return AgentResponse(success=True, result=f"Resultados encontrados para '{action.payload}'")
elif action.name == "send_email":
# Simular 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 o Prometheus/Grafana etc.
return response
# Conectando o agente com o middleware
agent = LoggingMetricsInterceptor(AgentCore())
# Casos de teste
print("\n--- Teste 1: Busca na Web Bem-Sucedida ---")
response1 = agent.execute_action(AgentAction("search_web", "últimas notícias AI"))
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", "data"))
print(f"Resposta Final: {response4}")
Neste exemplo, LoggingMetricsInterceptor encapsula AgentCore. Qualquer chamada a execute_action passa primeiro pelo interceptor, que registra, mede o tempo e então passa o controle para o próximo manipulador (AgentCore), e finalmente processa a resposta.
2. O Modelo da Cadeia de Responsabilidade
O modelo da Cadeia de Responsabilidade permite que vários manipuladores (componentes middleware) tratem uma solicitação sequencialmente. 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 várias 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 de Entradas
Consideremos um agente que recebe comandos em linguagem natural. Antes que o agente principal processe o comando, poderíamos querer validar a entrada, limpá-la ou traduzi-la em 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: remover espaços antes/depois, converter 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 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: Processamento do comando com a intenção '{command.processed_data.get('intent')}' e os 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 os comandos
agent_entry_point = sanitizer
# Comandos de teste
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 tratado 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 tratado final: {processed_cmd2}")
print("\n--- Teste 3: Comando inválido curto ---")
cmd3 = Command("oi")
processed_cmd3 = agent_entry_point.handle_command(cmd3)
print(f"Comando tratado 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 tratado final: {processed_cmd4}")
Aqui, um Command objeto percorre uma cadeia: InputSanitizer -> CommandValidator -> IntentRecognizer -> AgentCoreProcessor. Cada componente modifica o objeto Command ou define seu indicador is_valid. Se um componente invalidar o comando, os componentes seguintes podem interromper o processamento de forma suave.
3. O modelo Adaptador para ferramentas/APIs externas
Embora não seja estritamente um middleware no sentido de interceptação de requisições-respostas, o modelo Adaptador é crucial para permitir que agentes interajam com diversas ferramentas e APIs externas de forma padronizada. Um adaptador encapsula um serviço de terceiros, fornecendo uma interface consistente para que o agente o utilize, abstraindo as especificidades 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 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" # Substituir 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() # Lança um HTTPError para más respostas (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"Calendar: Criando o evento '{title}' de {start_time} a {end_time}")
# Simular 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"Calendar: Lista de eventos para {date}")
# Simular uma chamada API
time.sleep(0.1)
return {"status": "success", "events": [{"title": "Sync de 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 a ferramenta '{tool_name}' através do adaptador '{adapter_name}' com os parâmetros {params}")
return adapter.execute(tool_name, params)
# Inicializar 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 a ferramenta 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 a ferramenta Calendário (Criar Evento) ---")
calendar_event = agent_toolbox.use_tool("calendar", "create_event", {"title": "Revisão de 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 a ferramenta 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 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 especificidades de como chamar WeatherAPIAdapter ou CalendarAPIAdapter; basta solicitar uma ferramenta pelo nome e fornecer parâmetros. Cada adaptador traduz então essa solicitação genérica em chamadas de API específicas exigidas.
4. O modelo Registro/Localizador de serviço
O modelo de Registro ou Localizador de serviço é comumente usado para fornecer aos agentes acesso a diversos serviços, capacidades ou outros agentes em um sistema multi-agentes. Em vez de codificar dependências de forma rígida, os agentes consultam um registro central para descobrir e obter referências aos componentes necessários em tempo real. Isso melhora a flexibilidade e o desacoplamento.
Exemplo prático: Descoberta dinâmica das capacidades do agente
Imagine um agente que precisa de uma capacidade específica, como a síntese de texto ou a geração de imagens. Ele 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"Resumo do texto: '{text[:30]}...' ")
# Simular chamada LLM ou lógica de resumo
time.sleep(0.3)
return f"Resumo de '{text[:20]}...': É uma versão concisa."
class ImageGenerator(Capability):
def execute(self, prompt: str) -> str:
logging.info(f"Geração de imagem para o prompt: '{prompt}'")
# Simular chamada API de 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"A capacidade '{name}' já está registrada. Substituindo.")
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 falhou ao processar '{request_type}': {e}"
except Exception as e:
return f"O Agente encontrou um erro inesperado para '{request_type}': {e}"
# Configuração do registro
registry = CapabilityRegistry()
registry.register_capability("summarize", TextSummarizer())
registry.register_capability("generate_image", ImageGenerator())
# Criar um agente com acesso ao registro
agent_app = Agent(registry)
# O agente utiliza capacidades
print("\n--- Agente solicitando um resumo ---")
summary_result = agent_app.process_request("summarize", "A rápida raposa marrom salta sobre o cão preguiçoso. É um pangrama clássico usado para exibir todas as letras do alfabeto.")
print(summary_result)
print("\n--- Agente solicitando a 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 uma capacidade desconhecida ---")
unknown_result = agent_app.process_request("translate", "hello world")
print(unknown_result)
O CapabilityRegistry serve como localizador de serviço. O Agent não instancia diretamente TextSummarizer ou ImageGenerator; ele solicita ao registro uma capacidade pelo seu nome lógico. Isso permite substituir, atualizar ou adicionar capacidades sem modificar a lógica fundamental do agente.
Combinação de modelos Middleware
Nos sistemas de agentes reais, esses modelos são frequentemente combinados. Por exemplo, um comando de usuário que chega 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 usa o Registro/Localizador de Serviço para encontrar um Adaptador apropriado para uma ferramenta externa. A execução dessa ferramenta pode então ser envolvida por um Interceptor para registro e gerenciamento de erros.
Exemplo: Um Fluxo de Interação Agente Multicamadas
Vamos esboçar brevemente como isso poderia parecer:
# 1. Solicitação de entrada (por exemplo, a partir de uma interface de chat do usuário)
user_input = "Por favor, agende uma reunião sobre os resultados do T4 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 fundamental do agente decide usar uma ferramenta
intent_params = processed_command.processed_data.get('params', {})
# 4. Usar o 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 envolto por um Interceptor genérico ToolCallInterceptor)
# Para simplificar, chamaremos diretamente a caixa de ferramentas aqui, mas imagine que ela seja um proxy.
# Simular 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" # Analisado a partir de user_input por um IntentRecognizer mais sofisticado
end_time_str = "2023-10-28 16:00"
print("\n--- Agente orquestrando o uso de 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 não conseguiu processar a solicitação: {processed_command.error_message or 'Intenção inválida'}")
Esse fluxo demonstra como diferentes modelos middleware podem ser compostos para criar uma arquitetura de agente sólida e mantível.
Conclusão
Os modelos middleware de agentes são essenciais para construir sistemas de agentes AI escaláveis, bem estruturados e mantíveis. Ao aplicar modelos como Interceptor, Cadeia de Responsabilidade, Adaptador e Registro/Localizador de Serviço, os desenvolvedores podem gerenciar eficazmente preocupações transversais, integrar diversas funcionalidades e abstrair complexidades. Esses modelos promovem modularidade, reutilização e extensibilidade, permitindo que os agentes evoluam e interajam com seu ambiente de forma mais inteligente e confiável. À medida que os agentes AI se tornam mais sofisticados e integrados em nosso cotidiano, uma compreensão profunda e uma aplicação prática desses modelos middleware serão essenciais para o sucesso.
🕒 Published: