🔀 Camada 2 — Roteamento Multi-Agente

Implementação de roteamento estático

class StaticRouter:
    """Roteador estático: tabela fixa tipo_tarefa → agente."""

    ROUTING_TABLE = {
        "pesquisa":     "research_agent",
        "analise":      "analysis_agent",
        "redacao":      "writing_agent",
        "juridico":     "legal_agent",
        "financeiro":   "finance_agent",
        "tecnico":      "tech_agent",
        "formatacao":   "formatting_agent",
    }

    FALLBACK_AGENT = "general_agent"

    def route(self, task: dict) -> str:
        task_type = task.get("agente_tipo", "").lower()
        agent = self.ROUTING_TABLE.get(task_type, self.FALLBACK_AGENT)

        print(f"[StaticRouter] Task '{task['id']}' → Agent '{agent}'")
        return agent

    def add_route(self, task_type: str, agent_id: str):
        """Adicionar rota dinamicamente (ainda estático, mas configurável)."""
        self.ROUTING_TABLE[task_type] = agent_id

Implementação: LLMRouter

import anthropic
import json

client = anthropic.Anthropic()

class LLMRouter:
    """Roteador dinâmico: LLM seleciona agente por semântica da tarefa."""

    def __init__(self, agent_registry: dict):
        """
        agent_registry: dict com agent_id → descrição de capacidades
        Exemplo:
        {
          "research_agent": "Especializado em pesquisa web, coleta de dados públicos...",
          "legal_agent": "Análise jurídica, contratos, compliance LGPD, litígios...",
          "finance_agent": "Análise financeira, modelagem, valuation, DCF...",
        }
        """
        self.registry = agent_registry

    def route(self, task: dict) -> tuple[str, float]:
        """Retorna (agent_id, confidence_score)."""

        agents_desc = "\n".join([
            f"- {agent_id}: {desc}"
            for agent_id, desc in self.registry.items()
        ])

        response = client.messages.create(
            model="claude-haiku-4-5",  # Modelo barato para classificação
            max_tokens=256,
            messages=[{
                "role": "user",
                "content": f"""Selecione o agente mais adequado para esta tarefa.

TAREFA:
Título: {task['titulo']}
Descrição: {task['descricao']}

AGENTES DISPONÍVEIS:
{agents_desc}

Retorne JSON com:
{{"agent_id": "id_do_agente", "confidence": 0.0-1.0, "reasoning": "justificativa curta"}}"""
            }]
        )

        result = json.loads(response.content[0].text)

        # Fallback se confiança baixa
        if result["confidence"] < 0.6:
            result["agent_id"] = "general_agent"

        return result["agent_id"], result["confidence"]

    def add_agent(self, agent_id: str, capabilities: str):
        """Adicionar agente ao pool sem alterar código de roteamento."""
        self.registry[agent_id] = capabilities

Estratégias de load balancing

from collections import deque
import time

class AgentPool:
    """Pool de instâncias de um tipo de agente com load balancing."""

    def __init__(self, agent_type: str, instances: list[str]):
        self.agent_type = agent_type
        self.instances = instances
        self.queue_depth = {inst: 0 for inst in instances}
        self.last_used = {inst: 0.0 for inst in instances}
        self._round_robin = deque(instances)

    def get_instance_round_robin(self) -> str:
        """Rotação simples — ignora estado."""
        instance = self._round_robin[0]
        self._round_robin.rotate(-1)
        return instance

    def get_instance_least_loaded(self) -> str:
        """Escolhe instância com menor fila — ótimo para tarefas longas."""
        return min(self.queue_depth, key=self.queue_depth.get)

    def get_instance_health_aware(self, healthy_instances: set) -> str:
        """Escolhe instância saudável com menor carga."""
        healthy = {k: v for k, v in self.queue_depth.items()
                   if k in healthy_instances}
        if not healthy:
            raise RuntimeError(f"Nenhuma instância saudável de {self.agent_type}")
        return min(healthy, key=healthy.get)

    def mark_busy(self, instance_id: str):
        self.queue_depth[instance_id] += 1
        self.last_used[instance_id] = time.time()

    def mark_free(self, instance_id: str):
        self.queue_depth[instance_id] = max(0, self.queue_depth[instance_id] - 1)

    def status(self) -> dict:
        return {
            "type": self.agent_type,
            "instances": len(self.instances),
            "queue_depths": dict(self.queue_depth),
            "total_load": sum(self.queue_depth.values())
        }

import time
import json
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class RoutingDecision:
    task_id: str
    agent_id: str
    model: str
    complexity: str
    confidence: float
    timestamp: float = field(default_factory=time.time)
    duration_ms: float = 0.0

class AgentRouter:
    """Router completo com pool, load balancing, custo e logging."""

    def __init__(self, agent_registry: dict, pools: dict[str, AgentPool]):
        self.registry = agent_registry  # agent_type → capabilities
        self.pools = pools              # agent_type → AgentPool
        self.llm_router = LLMRouter(agent_registry)
        self.cost_router = CostAwareRouter()
        self.decisions_log: list[RoutingDecision] = []

    def route(self, task: dict) -> RoutingDecision:
        start = time.time()

        # 1. Classificar por complexidade (custo)
        cost_result = self.cost_router.route(task)

        # 2. Selecionar tipo de agente (capacidade)
        agent_type, confidence = self.llm_router.route(task)

        # 3. Selecionar instância (disponibilidade)
        pool = self.pools.get(agent_type)
        if pool:
            instance_id = pool.get_instance_least_loaded()
            pool.mark_busy(instance_id)
        else:
            instance_id = f"{agent_type}_default"

        decision = RoutingDecision(
            task_id=task["id"],
            agent_id=instance_id,
            model=cost_result["model"],
            complexity=cost_result["complexity"],
            confidence=confidence,
            duration_ms=(time.time() - start) * 1000
        )

        self.decisions_log.append(decision)
        print(f"[Router] {task['id']} → {instance_id} ({cost_result['model']}, "
              f"conf={confidence:.2f}, {decision.duration_ms:.0f}ms)")

        return decision

    def get_routing_metrics(self) -> dict:
        """Métricas de acerto do roteador."""
        if not self.decisions_log:
            return {}
        avg_confidence = sum(d.confidence for d in self.decisions_log) / len(self.decisions_log)
        avg_latency = sum(d.duration_ms for d in self.decisions_log) / len(self.decisions_log)
        complexity_dist = {}
        for d in self.decisions_log:
            complexity_dist[d.complexity] = complexity_dist.get(d.complexity, 0) + 1

        return {
            "total_decisions": len(self.decisions_log),
            "avg_confidence": round(avg_confidence, 3),
            "avg_routing_latency_ms": round(avg_latency, 1),
            "complexity_distribution": complexity_dist,
        }