💡 Custo-Otimização: Roteamento Heterogêneo e Cache Semântico

O erro mais caro em orquestradores de IA é usar o modelo mais potente para todas as tarefas. Claude Opus custa 60x mais que Claude Haiku. Para uma tarefa simples como "classifique este texto em 3 categorias", Haiku resolve tão bem quanto Opus — ao custo de $0.25/M tokens vs. $15/M tokens.

O classificador analisa cada tarefa antes da execução e decide qual modelo usar. Pode ser baseado em regras (heurísticas) para zero custo adicional, ou usar um LLM leve (Haiku) para classificação mais precisa.

from dataclasses import dataclass
from enum import Enum
from typing import Optional
import re

class Complexity(str, Enum):
    SIMPLE = "simple"      # → Haiku
    MEDIUM = "medium"      # → Sonnet
    COMPLEX = "complex"    # → Opus

@dataclass
class ClassificationResult:
    complexity: Complexity
    model: str
    confidence: float
    reasoning: str
    estimated_cost: float

class ComplexityClassifier:
    """
    Classifica tarefas em 3 níveis de complexidade
    para roteamento heterogêneo de modelos.

    Estratégia em 3 camadas (por ordem de custo):
    1. Regex/heurística: zero custo, 80% de cobertura
    2. Feature-based ML: zero custo, 95% de cobertura
    3. LLM classifier (Haiku): máxima precisão, custo mínimo
    """

    MODEL_MAP = {
        Complexity.SIMPLE:  "claude-3-haiku-20240307",
        Complexity.MEDIUM:  "claude-3-5-sonnet-20241022",
        Complexity.COMPLEX: "claude-opus-4-5",
    }

    MODEL_COSTS = {
        "claude-3-haiku-20240307":      {"input": 0.25,  "output": 1.25},
        "claude-3-5-sonnet-20241022":   {"input": 3.00,  "output": 15.00},
        "claude-opus-4-5":              {"input": 15.00, "output": 75.00},
    }

    # Palavras-chave que indicam alta complexidade
    COMPLEX_KEYWORDS = {
        "multi-step reasoning", "prove", "complex analysis",
        "compare and contrast", "synthesize", "comprehensive",
        "strategic", "evaluate trade-offs", "critique",
        "write production code", "design architecture",
        "raciocínio avançado", "análise estratégica",
    }

    # Palavras-chave que indicam baixa complexidade
    SIMPLE_KEYWORDS = {
        "classify", "extract", "list", "summarize briefly",
        "categorize", "format", "translate", "yes or no",
        "true or false", "count", "identifique", "classifique",
        "extraia", "liste", "formate", "sim ou não",
    }

    def classify_by_heuristics(self, task: dict) -> Optional[ClassificationResult]:
        """Classificação por regras — custo zero."""
        task_text = (
            task.get("description", "") + " " +
            task.get("type", "") + " " +
            str(task.get("input", ""))
        ).lower()

        # Verifica número de sub-tarefas
        num_subtasks = len(task.get("subtasks", []))
        if num_subtasks > 5:
            return ClassificationResult(
                complexity=Complexity.COMPLEX,
                model=self.MODEL_MAP[Complexity.COMPLEX],
                confidence=0.85,
                reasoning=f"Múltiplas sub-tarefas ({num_subtasks})",
                estimated_cost=self._estimate_cost(Complexity.COMPLEX)
            )

        # Verifica palavras-chave de complexidade alta
        complex_hits = sum(1 for kw in self.COMPLEX_KEYWORDS if kw in task_text)
        simple_hits = sum(1 for kw in self.SIMPLE_KEYWORDS if kw in task_text)

        # Comprimento da entrada como proxy de complexidade
        input_length = len(str(task.get("input", "")))

        if complex_hits >= 2 or input_length > 5000:
            return ClassificationResult(
                complexity=Complexity.COMPLEX,
                model=self.MODEL_MAP[Complexity.COMPLEX],
                confidence=0.80,
                reasoning=f"Keywords complexas: {complex_hits}, input: {input_length} chars",
                estimated_cost=self._estimate_cost(Complexity.COMPLEX)
            )

        if simple_hits >= 1 and complex_hits == 0 and input_length < 1000:
            return ClassificationResult(
                complexity=Complexity.SIMPLE,
                model=self.MODEL_MAP[Complexity.SIMPLE],
                confidence=0.85,
                reasoning=f"Keywords simples: {simple_hits}, input curto: {input_length} chars",
                estimated_cost=self._estimate_cost(Complexity.SIMPLE)
            )

        return None  # inconclusivo → vai para próxima camada

    def classify_by_type(self, task: dict) -> Optional[ClassificationResult]:
        """Classificação por tipo de tarefa — custo zero."""
        task_type = task.get("type", "").lower()

        SIMPLE_TYPES = {
            "classification", "extraction", "formatting",
            "translation", "routing", "validation", "parsing",
            "classificação", "extração", "formatação", "roteamento",
        }

        MEDIUM_TYPES = {
            "analysis", "summarization", "code_review", "research",
            "comparison", "writing", "report",
            "análise", "pesquisa", "redação", "relatório",
        }

        COMPLEX_TYPES = {
            "architecture_design", "complex_coding", "strategic_planning",
            "synthesis", "multi_agent_coordination", "reasoning",
            "planejamento_estratégico", "síntese", "raciocínio",
        }

        if task_type in SIMPLE_TYPES:
            return ClassificationResult(
                complexity=Complexity.SIMPLE,
                model=self.MODEL_MAP[Complexity.SIMPLE],
                confidence=0.90,
                reasoning=f"Tipo simples: {task_type}",
                estimated_cost=self._estimate_cost(Complexity.SIMPLE)
            )
        if task_type in MEDIUM_TYPES:
            return ClassificationResult(
                complexity=Complexity.MEDIUM,
                model=self.MODEL_MAP[Complexity.MEDIUM],
                confidence=0.85,
                reasoning=f"Tipo médio: {task_type}",
                estimated_cost=self._estimate_cost(Complexity.MEDIUM)
            )
        if task_type in COMPLEX_TYPES:
            return ClassificationResult(
                complexity=Complexity.COMPLEX,
                model=self.MODEL_MAP[Complexity.COMPLEX],
                confidence=0.90,
                reasoning=f"Tipo complexo: {task_type}",
                estimated_cost=self._estimate_cost(Complexity.COMPLEX)
            )

        return None

    async def classify_by_llm(self, task: dict) -> ClassificationResult:
        """Classificação por LLM (Haiku) — fallback de alta precisão."""
        import anthropic, json

        client = anthropic.AsyncAnthropic()
        response = await client.messages.create(
            model="claude-3-haiku-20240307",
            max_tokens=100,
            messages=[{
                "role": "user",
                "content": f"""Classify this AI task complexity.
Task: {str(task)[:500]}

Reply ONLY with JSON: {{"complexity": "simple|medium|complex", "reason": "..."}}"""
            }]
        )

        try:
            result = json.loads(response.content[0].text)
            complexity = Complexity(result["complexity"])
            return ClassificationResult(
                complexity=complexity,
                model=self.MODEL_MAP[complexity],
                confidence=0.92,
                reasoning=result.get("reason", "LLM classified"),
                estimated_cost=self._estimate_cost(complexity)
            )
        except Exception:
            # Default seguro
            return ClassificationResult(
                complexity=Complexity.MEDIUM,
                model=self.MODEL_MAP[Complexity.MEDIUM],
                confidence=0.50,
                reasoning="Fallback para médio",
                estimated_cost=self._estimate_cost(Complexity.MEDIUM)
            )

    async def classify(self, task: dict) -> ClassificationResult:
        """Pipeline de classificação em cascata."""
        # Camada 1: por tipo (zero custo)
        result = self.classify_by_type(task)
        if result and result.confidence >= 0.85:
            return result

        # Camada 2: por heurística (zero custo)
        result = self.classify_by_heuristics(task)
        if result and result.confidence >= 0.80:
            return result

        # Camada 3: LLM (custo mínimo, máxima precisão)
        return await self.classify_by_llm(task)

    def _estimate_cost(self, complexity: Complexity, avg_tokens: int = 1000) -> float:
        model = self.MODEL_MAP[complexity]
        costs = self.MODEL_COSTS[model]
        # Assume 70% input, 30% output
        return (avg_tokens * 0.7 * costs["input"] + avg_tokens * 0.3 * costs["output"]) / 1_000_000

Cache semântico vai além do cache exato: se uma tarefa é semanticamente similar a outra já processada (mesmo que com palavras diferentes), reutilizamos o resultado. Hit rate típico: 30-60% em workflows repetitivos.

# pip install anthropic redis numpy

import numpy as np
import json
import hashlib
import time
from typing import Optional
import redis.asyncio as aioredis

class SemanticCache:
    """
    Cache semântico usando embeddings de texto.
    Queries similares (cosine similarity > threshold) retornam
    resultado cacheado sem chamar o LLM.

    Custo de embedding (Voyage-3): $0.18/M tokens
    Custo evitado (Sonnet):        $18/M tokens (100x mais barato)
    """

    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        similarity_threshold: float = 0.92,
        ttl_seconds: int = 3600 * 24,  # 24 horas
        max_cache_size: int = 10_000,
    ):
        self.redis = aioredis.from_url(redis_url, decode_responses=True)
        self.threshold = similarity_threshold
        self.ttl = ttl_seconds
        self.max_size = max_cache_size

    async def get_embedding(self, text: str) -> np.ndarray:
        """Gera embedding via Voyage API (ou Anthropic)."""
        import anthropic

        client = anthropic.Anthropic()
        # Voyage-3 é o modelo de embedding da Anthropic
        response = client.messages.create(
            model="claude-3-haiku-20240307",
            max_tokens=10,
            messages=[{"role": "user", "content": text}]
        )
        # Em produção, use a API de embeddings do Voyage:
        # import voyageai
        # vo = voyageai.Client()
        # result = vo.embed([text], model="voyage-3")
        # return np.array(result.embeddings[0])

        # Simulação para exemplo (usar Voyage em produção)
        np.random.seed(hash(text) % 2**32)
        return np.random.randn(1024)

    @staticmethod
    def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
        """Similaridade de cosseno entre dois vetores."""
        a_norm = np.linalg.norm(a)
        b_norm = np.linalg.norm(b)
        if a_norm == 0 or b_norm == 0:
            return 0.0
        return float(np.dot(a, b) / (a_norm * b_norm))

    def _make_key(self, prompt: str) -> str:
        """Chave de lookup no Redis."""
        return f"semantic_cache:{hashlib.sha256(prompt.encode()).hexdigest()[:16]}"

    async def get(self, prompt: str, model: str) -> Optional[dict]:
        """
        Tenta encontrar resultado similar no cache.
        Retorna None se não houver hit.
        """
        query_embedding = await self.get_embedding(prompt)

        # Busca todos os embeddings cached (em prod: use Redis Vector Search)
        pattern = "semantic_cache:*"
        keys = await self.redis.keys(pattern)

        best_similarity = 0.0
        best_result = None

        for key in keys[:500]:  # limita busca para performance
            cached_raw = await self.redis.get(key)
            if not cached_raw:
                continue

            cached = json.loads(cached_raw)

            # Só usa cache do mesmo modelo
            if cached.get("model") != model:
                continue

            cached_embedding = np.array(cached["embedding"])
            similarity = self.cosine_similarity(query_embedding, cached_embedding)

            if similarity > best_similarity:
                best_similarity = similarity
                best_result = cached

        if best_similarity >= self.threshold and best_result:
            # Registra hit no Redis para métricas
            await self.redis.incr("semantic_cache:hits")
            return {
                "result": best_result["result"],
                "similarity": best_similarity,
                "cache_age_seconds": int(time.time()) - best_result["timestamp"],
                "cache_hit": True,
            }

        await self.redis.incr("semantic_cache:misses")
        return None

    async def set(self, prompt: str, model: str, result: str) -> None:
        """Armazena resultado no cache com embedding."""
        embedding = await self.get_embedding(prompt)
        key = self._make_key(prompt)

        cached_data = {
            "prompt_preview": prompt[:200],
            "model": model,
            "embedding": embedding.tolist(),
            "result": result,
            "timestamp": int(time.time()),
        }

        await self.redis.setex(key, self.ttl, json.dumps(cached_data))
        await self.redis.incr("semantic_cache:stored")

    async def get_stats(self) -> dict:
        """Estatísticas de performance do cache."""
        hits = int(await self.redis.get("semantic_cache:hits") or 0)
        misses = int(await self.redis.get("semantic_cache:misses") or 0)
        total = hits + misses
        hit_rate = hits / total if total > 0 else 0

        keys = await self.redis.keys("semantic_cache:*")
        cache_size = len([k for k in keys if not k.startswith("semantic_cache:h") and
                         not k.startswith("semantic_cache:m") and
                         not k.startswith("semantic_cache:s")])

        return {
            "hit_rate": hit_rate,
            "hits": hits,
            "misses": misses,
            "cache_size": cache_size,
            "estimated_savings_usd": hits * 0.003,  # ~$0.003 por hit evitado (Sonnet)
        }


# ============================================================
# INTEGRAÇÃO NO ORQUESTRADOR
# ============================================================

class CacheAwareLLMClient:
    """Cliente LLM que verifica cache semântico antes de chamar a API."""

    def __init__(self, cache: SemanticCache):
        self.cache = cache
        import anthropic
        self.client = anthropic.AsyncAnthropic()

    async def complete(
        self, prompt: str, model: str, max_tokens: int = 2048
    ) -> tuple[str, bool]:
        """
        Retorna (resposta, cache_hit).
        Se cache_hit=True, não houve custo de LLM.
        """
        # Tenta cache primeiro
        cached = await self.cache.get(prompt, model)
        if cached:
            return cached["result"], True

        # Cache miss → chama LLM
        response = await self.client.messages.create(
            model=model,
            max_tokens=max_tokens,
            messages=[{"role": "user", "content": prompt}]
        )
        result = response.content[0].text

        # Armazena para próximas queries similares
        await self.cache.set(prompt, model, result)
        return result, False

Em workflows longos, o contexto acumula. Após 20 iterações, você pode estar enviando 8.000 tokens de histórico quando apenas 1.500 são relevantes. Context compression reduz custo sem perder qualidade.

from dataclasses import dataclass
from typing import List, Dict, Any
import anthropic

@dataclass
class Message:
    role: str
    content: str
    tokens: int
    relevance_score: float = 1.0

class ContextCompressor:
    """
    Reduz o contexto mantendo apenas informações relevantes.
    Estratégias implementadas:
    1. Token budget: mantém últimas N mensagens
    2. Relevance filtering: remove mensagens de baixa relevância
    3. Summarization: comprime histórico antigo em sumário
    """

    def __init__(
        self,
        max_tokens: int = 4000,       # budget máximo de contexto
        summary_threshold: int = 3000, # quando comprime
        min_recency_messages: int = 5  # sempre mantém últimas N mensagens
    ):
        self.max_tokens = max_tokens
        self.summary_threshold = summary_threshold
        self.min_recency = min_recency_messages
        self.client = anthropic.AsyncAnthropic()

    def count_tokens(self, messages: List[Message]) -> int:
        """Conta tokens aproximados (4 chars ≈ 1 token)."""
        return sum(len(m.content) // 4 for m in messages)

    def filter_by_relevance(
        self, messages: List[Message], query: str, top_k: int = 10
    ) -> List[Message]:
        """
        Mantém mensagens mais relevantes para a query atual.
        Em prod: use embedding similarity para score mais preciso.
        """
        query_words = set(query.lower().split())

        for msg in messages:
            content_words = set(msg.content.lower().split())
            overlap = len(query_words & content_words)
            msg.relevance_score = overlap / max(len(query_words), 1)

        # Sempre mantém mensagens mais recentes
        recent = messages[-self.min_recency:]
        older = messages[:-self.min_recency]

        # Filtra antigas por relevância
        relevant_older = sorted(
            older, key=lambda m: m.relevance_score, reverse=True
        )[:max(0, top_k - self.min_recency)]

        # Reconstrói na ordem cronológica
        relevant_older_set = set(id(m) for m in relevant_older)
        chronological = [m for m in messages
                        if id(m) in relevant_older_set or m in recent]
        return chronological

    async def summarize_history(
        self, messages: List[Message]
    ) -> Message:
        """Comprime histórico antigo em um sumário conciso."""
        history_text = "\n".join([
            f"{m.role}: {m.content[:300]}..."
            for m in messages
        ])

        response = await self.client.messages.create(
            model="claude-3-haiku-20240307",  # modelo barato para sumário
            max_tokens=500,
            messages=[{
                "role": "user",
                "content": f"""Crie um sumário conciso das principais informações:

{history_text}

Inclua: decisões tomadas, resultados obtidos, contexto essencial.
Máximo: 400 tokens."""
            }]
        )

        summary = response.content[0].text
        return Message(
            role="system",
            content=f"[HISTÓRICO COMPRIMIDO] {summary}",
            tokens=len(summary) // 4,
            relevance_score=1.0  # sumários são sempre relevantes
        )

    async def compress(
        self,
        messages: List[Message],
        current_query: str
    ) -> List[Message]:
        """
        Pipeline de compressão completo.
        1. Filtra por relevância
        2. Se ainda grande demais, sumariza histórico antigo
        """
        total_tokens = self.count_tokens(messages)

        # Abaixo do threshold: não precisa comprimir
        if total_tokens <= self.summary_threshold:
            return messages

        # Passo 1: filtra por relevância
        filtered = self.filter_by_relevance(messages, current_query)
        filtered_tokens = self.count_tokens(filtered)

        if filtered_tokens <= self.max_tokens:
            return filtered

        # Passo 2: sumariza histórico antigo
        split_point = max(0, len(filtered) - self.min_recency)
        to_summarize = filtered[:split_point]
        to_keep = filtered[split_point:]

        if to_summarize:
            summary_msg = await self.summarize_history(to_summarize)
            result = [summary_msg] + to_keep
        else:
            result = to_keep

        final_tokens = self.count_tokens(result)
        reduction_pct = (1 - final_tokens / total_tokens) * 100

        print(f"[ContextCompressor] {total_tokens} → {final_tokens} tokens "
              f"({reduction_pct:.0f}% redução)")

        return result

Plan-and-Execute é um padrão onde um modelo potente (Opus) planeja e modelos baratos (Haiku/Sonnet) executam. O planner vê o objetivo completo uma vez; os executores recebem apenas a sub-tarefa relevante.

from dataclasses import dataclass
from typing import List, Dict
import asyncio
import anthropic
import json

@dataclass
class ExecutionPlan:
    tasks: List[dict]
    strategy: str
    estimated_cost: float
    parallel_groups: List[List[str]]  # tarefas que podem rodar em paralelo

class PlanAndExecuteOrchestrator:
    """
    Padrão Plan-and-Execute:
    - Planner (Opus/Sonnet): vê o objetivo completo, cria plano detalhado
    - Executors (Haiku/Sonnet): recebem apenas 1 sub-tarefa, sem contexto global
    - Resultado: qualidade de Opus pelo custo de Haiku (para a maior parte)
    """

    def __init__(self):
        self.client = anthropic.AsyncAnthropic()
        self.classifier = ComplexityClassifier()

    async def plan(self, objective: str) -> ExecutionPlan:
        """Usa Sonnet para planejar (balança custo vs. qualidade de planejamento)."""
        response = await self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"""Decomponha em tarefas atômicas e independentes.
Objetivo: {objective}

Responda em JSON:
{{
  "strategy": "parallel|sequential|mixed",
  "parallel_groups": [["t1", "t2"], ["t3"]],
  "tasks": [
    {{
      "id": "t1",
      "description": "...",
      "type": "research|analysis|writing|classification",
      "depends_on": [],
      "estimated_complexity": "simple|medium|complex",
      "context_needed": "brief summary of what context this task needs"
    }}
  ]
}}"""
            }]
        )

        raw = response.content[0].text
        match = __import__('re').search(r'\{.*\}', raw, __import__('re').DOTALL)
        data = json.loads(match.group()) if match else {}
        tasks = data.get("tasks", [])

        # Estima custo baseado na complexidade declarada no plano
        total_cost = sum(
            self.classifier._estimate_cost(
                Complexity(t.get("estimated_complexity", "medium"))
            )
            for t in tasks
        )

        return ExecutionPlan(
            tasks=tasks,
            strategy=data.get("strategy", "parallel"),
            estimated_cost=total_cost,
            parallel_groups=data.get("parallel_groups", [[t["id"] for t in tasks]])
        )

    async def execute_task(self, task: dict) -> dict:
        """
        Executa uma única tarefa com o modelo apropriado.
        O executor recebe APENAS a sub-tarefa — sem o objetivo global.
        Isso economiza tokens e evita "distração" do modelo.
        """
        # Classifica complexidade para escolher modelo
        classification = await self.classifier.classify(task)

        # Monta prompt mínimo — só o necessário
        prompt = f"""Execute esta tarefa específica:

Tipo: {task['type']}
Descrição: {task['description']}
Contexto necessário: {task.get('context_needed', 'N/A')}

Produza apenas o resultado da tarefa, sem explicações extras."""

        response = await self.client.messages.create(
            model=classification.model,
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )

        return {
            "task_id": task["id"],
            "result": response.content[0].text,
            "model_used": classification.model,
            "complexity": classification.complexity.value,
            "cost": classification.estimated_cost,
        }

    async def execute_parallel_group(self, task_ids: List[str], tasks_map: dict) -> List[dict]:
        """Executa grupo de tarefas em paralelo."""
        group_tasks = [tasks_map[tid] for tid in task_ids if tid in tasks_map]
        results = await asyncio.gather(*[
            self.execute_task(task) for task in group_tasks
        ])
        return list(results)

    async def run(self, objective: str) -> dict:
        """Executa o workflow completo Plan-and-Execute."""
        # 1. Planejamento (uma vez, com modelo médio)
        plan = await self.plan(objective)
        print(f"Plano criado: {len(plan.tasks)} tarefas, estratégia: {plan.strategy}")

        # 2. Execução em grupos paralelos (cada tarefa com modelo adequado)
        tasks_map = {t["id"]: t for t in plan.tasks}
        all_results = []

        for group in plan.parallel_groups:
            group_results = await self.execute_parallel_group(group, tasks_map)
            all_results.extend(group_results)

        # 3. Agrega resultados
        total_cost = sum(r.get("cost", 0) for r in all_results)
        models_used = {}
        for r in all_results:
            m = r.get("model_used", "unknown")
            models_used[m] = models_used.get(m, 0) + 1

        return {
            "results": all_results,
            "total_cost_usd": total_cost,
            "models_distribution": models_used,
            "tasks_completed": len(all_results),
        }

Dados reais de um workflow de análise competitiva com 50 empresas, medindo custo, latência e qualidade com e sem as otimizações.