Módulo 2.2 — Middleware e execução | DeerFlow 2.0

💡 Pré-requisito mental

Revise o módulo 2.1 se "node act" ainda soa abstrato. Todo o middleware vive dentro desse node — a esteira é executada cada vez que o grafo decide rodar uma tool.

🔗 Fluxo do middleware

Middleware, no DeerFlow, é uma camada plugável que intercepta cada tool call antes e depois da execução real. Pense em onion layers: o pedido desce por N middlewares até chegar ao handler da tool, e a resposta sobe pela mesma pilha em ordem inversa. Qualquer camada pode logar, transformar, bloquear ou substituir.

🎯 Conceito Central

Um middleware é uma função async (call, next) -> result. Recebe o call (nome da tool + args + contexto), pode alterá-lo, chama next(call) para descer na pilha, recebe o result, pode alterá-lo, e devolve. É praticamente o mesmo padrão de Express/Koa — e funciona pelos mesmos motivos.

•Antes do next — validação de args, rewrite, deny-list, medição de latência inicial.
•next(call) — passa adiante; se nada interceptou, vai até o handler real.
•Depois do next — log de resultado, redaction, métricas, retry, conversão de erro.
•Short-circuit — middleware pode devolver resultado sem chamar next (ex: cache hit, deny).

💡 Ordem importa

A ordem de registro define a ordem de execução. Guardrail deny-list sempre primeiro (short-circuit barato). Observabilidade/logging depois, para ver mesmo o que foi bloqueado. Retry e cache vão no meio, perto do handler. Inverter essa ordem transforma o middleware de ativo protetor em contador de bala perdida.

📨 Despacho via agent_name

Quando o lead_agent decide acionar um custom agent, ele não faz import Python direto. Ele emite uma tool call especial com agent_name no payload e deixa o middleware resolver. Isso parece cerimonial, mas é o que permite swap de agentes sem reboot e A/B testing de versões.

lead_agent emite dispatch

tool_call = {"name": "dispatch_agent", "args": {"agent_name": "security-auditor", ...}}

O plano diz "isso é trabalho do security-auditor". Em vez de importar a classe, o lead_agent produz uma tool call estruturada. A instrução vai no campo args.task.

Middleware de agents resolve

Lookup no registry, aplica policy, injeta contexto isolado

Um middleware dedicado captura calls com name == "dispatch_agent", consulta o registry (onde agentes estão registrados por nome), monta o subgrafo do custom agent e executa.

Custom agent roda em sub-state

Contexto próprio, memória própria, tools restritas

O custom agent começa com state isolado: só o task e as memórias pertinentes a ele. Ao terminar, devolve um resumo. O lead_agent vê esse resumo — não vê o histórico interno.

Resumo retorna como tool result

lead_agent segue o grafo normal como se fosse qualquer tool

Do ponto de vista do lead_agent, custom agent é uma tool. Isso simplifica o grafo e explica por que acoplar um novo agente não exige mudar o lead.

🔄 Ciclo de vida de uma tool call

Aqui está a sequência completa — do momento em que o modelo produz o JSON da tool_call até a volta ao loop. São sete etapas, todas relevantes quando algo não funciona.

Parse & validação de schema

tool_call do LLM vira um objeto tipado. Args são validados contra o JSON schema registrado da tool. Erro aqui vira mensagem de sistema de volta ao modelo — ele reformata sozinho.

Guardrails (allow/deny)

Middleware de guardrail checa se essa tool está permitida para esse usuário/skill/canal. Deny vira erro estruturado — o modelo aprende que aquela rota está fechada.

Pré-processamento

Injeção de credenciais, rewrite de args (ex: normalizar paths), quota check. O modelo não vê credenciais — elas entram aqui, fora do prompt.

Handler real

Executa a tool: HTTP request para MCP server, chamada Python local, spawn em sandbox. Timeout configurável — raro ser infinito.

Pós-processamento

Redaction de PII, truncation de outputs grandes, sumarização por sub-agent quando o resultado estoura o threshold configurado.

State update

Resultado é anexado ao histórico de mensagens do state, memórias "worth remembering" são detectadas, métricas são publicadas.

Volta ao grafo

Control flow volta ao node plan. O modelo lê o novo histórico e decide o próximo passo: outra tool, agent dispatch ou respond.

💥 Recuperação de erros

APIs externas falham. Tools alucinam args. Sandbox estoura memória. A pergunta não é "se" mas "quando" — e o middleware é o lugar certo para tratar isso sem que o grafo inteiro caia.

✓ Fazer

✓Retries com backoff exponencial para erros transientes (5xx, timeout).
✓Converter erros em mensagens legíveis pelo modelo (ele re-planeja).
✓Circuit breaker por tool — falha em série desliga a tool por N segundos.
✓Fallback tool quando disponível (ex: search B se search A cai).
✓Observabilidade: sempre logar a tentativa e o resultado.

✗ Evitar

✗Retry infinito em 4xx — são erros de args, insistir piora.
✗Swallowing de exceção: retornar "ok" quando deu erro.
✗Crash do grafo inteiro por falha de tool opcional.
✗Expor stacktraces de Python direto no prompt do modelo.
✗Fail-fast global — bom é fail-fast por tool, não por sessão.

📊 Heurísticas úteis

3 retries com backoff 1s, 2s, 4s cobre 95% das falhas transientes em APIs públicas.
Circuit breaker abre após 5 falhas consecutivas e reabre após 60s (half-open probe).
Budget por turno: máximo N tool calls por turno (default 20) evita loops custosos.
Timeout padrão de 30s por tool; IO realmente lenta opta-in explicitamente.

🧪 Lab: middleware que loga tools

Escrever o primeiro middleware. Objetivo simples: interceptar cada tool call, logar nome + duração + sucesso, e devolver. Trinta linhas de Python. Depois dessa base você adiciona retry, quota e guardrail sem esforço.

Passos do lab

1.
Criar arquivo em deerflow/app/middleware/tool_logger.py. A app importa do harness, nunca o contrário.

Escrever o middleware:

import time, logging
from deerflow.harness.middleware import ToolMiddleware, ToolCall, ToolResult

log = logging.getLogger("deerflow.tools")

class ToolLogger(ToolMiddleware):
    async def __call__(self, call: ToolCall, next) -> ToolResult:
        t0 = time.monotonic()
        try:
            result = await next(call)
            elapsed = (time.monotonic() - t0) * 1000
            log.info("tool_ok name=%s ms=%.1f", call.name, elapsed)
            return result
        except Exception as e:
            elapsed = (time.monotonic() - t0) * 1000
            log.warning("tool_err name=%s ms=%.1f err=%s", call.name, elapsed, e)
            raise

3.
Registrar no bootstrap. Em deerflow/app/bootstrap.py, adicione harness.middleware.add(ToolLogger()) antes dos middlewares já existentes.
4.
Reiniciar: docker compose restart deerflow. Confirme no log startup a linha "middleware registered: tool_logger".
5.
Rodar query que use pelo menos uma tool (ex: "pesquise sobre LangGraph e me dê 3 referências"). Em seguida docker compose logs --tail=50 deerflow e procure linhas tool_ok.
6.
Validar no LangSmith: cruze as durações do log com o trace — devem bater em ±10ms. Se não baterem, provavelmente há middleware extra entre o logger e o handler.

💡 Entregável

Snippet do log mostrando 3+ linhas tool_ok com durações reais + screenshot do trace correspondente no LangSmith. Com isso você prova que o middleware está na posição certa da pilha.

📝 Resumo do Módulo

✓

Middleware é onion: antes → next → depois — o mesmo padrão de Express/Koa aplicado a tool calls.

✓

Custom agent é uma tool call via agent_name — lead_agent não importa classes, despacha via registry.

✓

Ciclo de vida tem 7 etapas — parse, guardrail, pré-proc, handler, pós-proc, state, volta.

✓

Erro não derruba o grafo — retry com backoff para transiente, mensagem legível para args errados, circuit breaker para falha persistente.

✓

Ordem de middlewares importa — guardrail antes, observabilidade cobrindo tudo, cache/retry perto do handler.

✓

Middleware custom é trivial — 30 linhas de Python, registro no bootstrap, e a app ganha observabilidade sem tocar no harness.

Próximo Módulo:

2.3 — 🎨 Criando a sua primeira skill

Anatomia, trigger accuracy, testes e lab de skill para cruzar IBGE + Portal da Transparência.

← Módulo Anterior Voltar para Trilha Próximo Módulo →