Cost optimization agent: ridurre token spend 80% senza perdere quality

Cost optimization agent: ridurre token spend 80% senza perdere quality

Un agent Claude Sonnet 3.7 senza ottimizzazioni, su un task di analisi documentale con 20 step e context pieno, consuma tra 160K e 240K token per run. Al pricing attuale - $3 per milione di token input, $15 per milione output - ogni run vale tra €0.60 e €1.20. Moltiplica per 1000 run al giorno e ottieni €600–€1200 di costo quotidiano prima ancora di considerare i picchi. Questo è il numero che spaventa i team e spinge molti a tornare su soluzioni one-shot, rinunciando ai vantaggi reali degli agent.

Il problema non è che gli agent costino troppo. Il problema è che la maggior parte dei token viene sprecata: system prompt ripetuto a ogni chiamata senza caching, tool result verbosi che intasano il context, step facili eseguiti con il modello più costoso, tool call eseguite in sequenza quando potrebbero essere parallele. Questi sprechi non sono inevitabili - sono pattern di implementazione non ottimizzati che si correggono con 7 tecniche specifiche.

In questo episodio presento ogni tecnica con codice Python funzionante, percentuale di saving attesa e sforzo di implementazione. Chiudo con un benchmark before/after su un agent reale del ml-service del mio portfolio - dove Ollama on-prem elimina il costo monetario ma rende ancora più critica la riduzione dei cicli computazionali su un VPS con risorse limitate.

Sezione 1 - Token budget per step: dove si nasconde lo spreco

Prima di ottimizzare occorre misurare. La maggior parte dei team guarda il costo totale del run senza scomporne i contribuenti. Un agent ReAct tipico ha questa struttura di costo per ogni iterazione:

• System prompt: inviato a ogni chiamata LLM. Su un agent con 20 step e system prompt da 500 token, questo contribuisce con 10.000 token solo di overhead fisso.
• Tool definitions: ogni tool definition occupa 50–200 token a seconda della complessità dello schema JSON. Con 10 tool definiti, sono 500–2000 token ripetuti a ogni step.
• Conversation history: cresce linearmente con ogni step. Al passo 20, la history contiene messaggi 1–19. Senza potatura, il context raddoppia ogni ~10 step.
• Tool result verbosi: un risultato di ricerca vettoriale può restituire 5 chunk da 400 token ciascuno = 2000 token per tool call. Se l'agent fa 3 ricerche per step, sono 6000 token solo di tool output.

Il seguente snippet calcola la distribuzione del costo per categoria su un singolo step, in modo da avere una baseline prima di applicare ottimizzazioni.

# token_budget_analyzer.py - analizza la distribuzione del costo per step
# Installa: pip install anthropic tiktoken

import tiktoken
from dataclasses import dataclass, field
from typing import Any

# Tokenizer compatibile con la famiglia Claude (Anthropic usa tokenizer proprietario,
# cl100k_base è una buona approssimazione per stime di costo)
_enc = tiktoken.get_encoding("cl100k_base")

def count_tokens(text: str) -> int:
    """Conta token approssimativi per una stringa."""
    return len(_enc.encode(text))

@dataclass
class StepCostBreakdown:
    system_prompt_tokens: int = 0
    tool_definitions_tokens: int = 0
    conversation_history_tokens: int = 0
    tool_results_tokens: int = 0
    current_message_tokens: int = 0

    @property
    def total(self) -> int:
        return (
            self.system_prompt_tokens
            + self.tool_definitions_tokens
            + self.conversation_history_tokens
            + self.tool_results_tokens
            + self.current_message_tokens
        )

    def report(self) -> dict[str, Any]:
        """Restituisce breakdown percentuale - usato per identificare dove ottimizzare."""
        t = self.total or 1  # evita divisione per zero
        return {
            "total_tokens": t,
            "system_prompt": f"{self.system_prompt_tokens} ({100*self.system_prompt_tokens//t}%)",
            "tool_definitions": f"{self.tool_definitions_tokens} ({100*self.tool_definitions_tokens//t}%)",
            "conversation_history": f"{self.conversation_history_tokens} ({100*self.conversation_history_tokens//t}%)",
            "tool_results": f"{self.tool_results_tokens} ({100*self.tool_results_tokens//t}%)",
            "current_message": f"{self.current_message_tokens} ({100*self.current_message_tokens//t}%)",
        }

def analyze_step_cost(
    system_prompt: str,
    tool_definitions: list[str],
    conversation_history: list[dict],
    tool_results: list[str],
    current_message: str,
) -> StepCostBreakdown:
    """
    Analizza il costo token di un singolo step dell'agent.
    conversation_history: lista di messaggi {"role": ..., "content": ...}
    """
    history_text = " ".join(
        f"{m['role']}: {m['content']}" for m in conversation_history
    )
    breakdown = StepCostBreakdown(
        system_prompt_tokens=count_tokens(system_prompt),
        tool_definitions_tokens=sum(count_tokens(td) for td in tool_definitions),
        conversation_history_tokens=count_tokens(history_text),
        tool_results_tokens=sum(count_tokens(tr) for tr in tool_results),
        current_message_tokens=count_tokens(current_message),
    )
    return breakdown

Esegui analyze_step_cost su 5–10 step reali del tuo agent e osserva le percentuali. In quasi tutti i casi il pattern è: system prompt + tool definitions = 20–35% del costo fisso, conversation history = 30–50% al passo 15+, tool results = 20–40%. Sapere dove si concentra il costo guida la scelta delle tecniche da applicare per prime.

Sezione 2 - Le 7 tecniche con saving atteso e sforzo

La tabella seguente riassume le 7 tecniche con percentuale di saving attesa sul costo totale dell'agent run, sforzo di implementazione (basso/medio/alto) e prerequisiti. Il saving è stimato su un agent ReAct da 20 step con context di 8K token per step - i numeri variano in base al workload specifico.

Le prime quattro tecniche si combinano bene in pipeline sequenziale. Model routing e speculative execution richiedono più investimento iniziale ma producono il saving più significativo su workload eterogenei. Il batching è spesso il quick win più sottovalutato: non riduce i token ma riduce il numero di chiamate LLM, che su agent con molte tool call parallele può dimezzare la latenza e ridurre il costo di overhead.

Sezione 3 - Prompt caching Anthropic: come strutturare i blocchi cache_control

Il prompt caching Anthropic consente di marcare blocchi di contenuto come "cache_control": {"type": "ephemeral"}. Anthropic memorizza il KV-cache di questi blocchi per 5 minuti. Le chiamate successive che presentano lo stesso prefisso cached pagano solo 10% del costo input sui token cached. Su un system prompt da 2000 token ripetuto 20 volte, il saving è 20 × 2000 × 0.90 = 36.000 token di sconto.

La regola critica è che il contenuto cached deve essere un prefisso stabile: non può variare tra chiamate. Il system prompt e le tool definitions sono candidati ideali. Il conversation history non lo è, perché cambia a ogni step.

# prompt_caching_agent.py - implementa prompt caching Anthropic
# Installa: pip install anthropic>=0.40.0

import anthropic
from typing import Any

client = anthropic.Anthropic()

# System prompt lungo: candidato ideale per caching
# Deve essere stabile (identico) tra le chiamate nello stesso agent run
SYSTEM_PROMPT = """Sei un agent di analisi documentale specializzato in contratti legali.
Il tuo compito è estrarre clausole rilevanti, identificare obbligazioni e scadenze,
e produrre un summary strutturato in JSON.

Regole operative:
1. Usa sempre il tool `search_clauses` prima di rispondere su contenuto specifico.
2. Se una clausola è ambigua, usa `request_clarification` prima di procedere.
3. Non inferire obbligazioni non esplicitamente scritte nel documento.
4. Ogni risposta deve includere il riferimento alla sezione del documento.
[... altre 1500 token di istruzioni operative dettagliate ...]
"""

# Tool definitions: stabili, buoni candidati per caching
TOOL_DEFINITIONS = [
    {
        "name": "search_clauses",
        "description": "Cerca clausole nel documento per keyword o tipo.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Query di ricerca"},
                "clause_type": {
                    "type": "string",
                    "enum": ["obligation", "deadline", "penalty", "termination"],
                    "description": "Tipo di clausola da cercare",
                },
            },
            "required": ["query"],
        },
    },
    # ... altri tool
]

def call_agent_step_with_caching(
    conversation_history: list[dict],
    current_message: str,
) -> dict[str, Any]:
    """
    Esegue un singolo step agent con prompt caching attivo.
    Struttura: system (cached) + tools (cached) + history (non cached) + messaggio corrente.
    """
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=4096,
        # System prompt con cache_control: viene cachato dopo la prima chiamata
        system=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"},  # marker caching
            }
        ],
        # Tool definitions con cache_control sul blocco finale della lista
        tools=[
            *TOOL_DEFINITIONS[:-1],  # tool intermedi senza cache_control
            {
                **TOOL_DEFINITIONS[-1],
                "cache_control": {"type": "ephemeral"},  # marker sull'ultimo tool
            },
        ],
        messages=[
            *conversation_history,     # history NON cachata: cambia ogni step
            {"role": "user", "content": current_message},
        ],
    )

    # Leggi usage per monitorare cache hit/miss
    usage = response.usage
    cache_hit_tokens = getattr(usage, "cache_read_input_tokens", 0)
    cache_miss_tokens = getattr(usage, "cache_creation_input_tokens", 0)

    return {
        "response": response,
        "cache_hit_tokens": cache_hit_tokens,    # token letti dalla cache (90% sconto)
        "cache_miss_tokens": cache_miss_tokens,  # token scritti in cache (prima volta)
        "regular_tokens": usage.input_tokens,    # token non cached
    }

Attenzione: il caching ha senso solo se le chiamate successive arrivano entro 5 minuti dalla creazione del cache entry. Su agent con step lenti (più di qualche minuto ciascuno), il TTL scade e non ottieni il risparmio. Monitora sempre il campo cache_read_input_tokens nella response per verificare che il caching stia effettivamente scattando.

Sezione 4 - Context pruning con embedding similarity

Il context di un agent cresce linearmente con ogni step. Al passo 20 di un agent con messaggi da 400 token l'uno, la history pesa 8000 token prima ancora del messaggio corrente. Ma non tutti quei messaggi sono ugualmente rilevanti per il passo corrente. Il context pruning usa l'embedding similarity per tenere solo i messaggi semanticamente vicini all'obiettivo attuale.

Il principio è semplice: embed il messaggio corrente, embed ogni messaggio della history, calcola cosine similarity, filtra via i messaggi sotto una soglia. I messaggi recenti ricevono un boost artificiale perché tendono a essere più rilevanti anche se semanticamente distanti dall'obiettivo originale.

# context_pruner.py - pruning semantico della conversation history
# Installa: pip install sentence-transformers numpy

import numpy as np
from sentence_transformers import SentenceTransformer
from typing import Any

# Modello leggero e veloce: 384 dimensioni, ~80MB, latenza <10ms su CPU
_model = SentenceTransformer("all-MiniLM-L6-v2")

def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    """Calcola cosine similarity tra due vettori normalizzati."""
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-8))

def prune_context(
    conversation_history: list[dict[str, Any]],
    current_objective: str,
    similarity_threshold: float = 0.75,
    recency_boost_last_n: int = 3,
    max_tokens_budget: int = 4000,
) -> list[dict[str, Any]]:
    """
    Filtra la conversation history mantenendo solo i messaggi rilevanti.

    Args:
        conversation_history: lista di messaggi {"role", "content"}
        current_objective: descrizione dell'obiettivo corrente dell'agent
        similarity_threshold: soglia cosine similarity (0.75 = bilanciato)
        recency_boost_last_n: gli ultimi N messaggi vengono sempre mantenuti
        max_tokens_budget: limite token approssimativo per la history potata

    Returns:
        History filtrata, ordinata cronologicamente.
    """
    if not conversation_history:
        return []

    # Embed l'obiettivo corrente
    objective_emb = _model.encode(current_objective, normalize_embeddings=True)

    pruned = []
    total_tokens_approx = 0

    for i, message in enumerate(conversation_history):
        content = message.get("content", "")
        if not isinstance(content, str):
            # Gestione content strutturato (tool use/result): mantieni sempre
            pruned.append(message)
            continue

        # Gli ultimi N messaggi: mantieni sempre (recency bias)
        is_recent = i >= len(conversation_history) - recency_boost_last_n
        if is_recent:
            pruned.append(message)
            total_tokens_approx += len(content.split()) * 1.3  # stima token
            continue

        # Calcola similarità semantica con l'obiettivo
        msg_emb = _model.encode(content, normalize_embeddings=True)
        similarity = cosine_similarity(objective_emb, msg_emb)

        if similarity >= similarity_threshold:
            # Controlla budget token prima di aggiungere
            msg_tokens_approx = len(content.split()) * 1.3
            if total_tokens_approx + msg_tokens_approx <= max_tokens_budget:
                pruned.append(message)
                total_tokens_approx += msg_tokens_approx

    return pruned

Il parametro recency_boost_last_n=3 garantisce che gli ultimi 3 messaggi siano sempre inclusi, indipendentemente dalla similarity. Questo è importante per mantenere la coerenza immediata dell'agent: un agent che non ricorda il tool result dell'ultimo step spesso fa la stessa chiamata due volte, vanificando il risparmio.

Sezione 5 - Tool result summarization e memory compression

Quando un tool restituisce dati verbosi - una risposta di ricerca con 5 chunk, l'output di un'API con 50 campi JSON, il testo di un documento lungo - la maggior parte di quell'output non serve all'agent nel passo successivo. La tool result summarization usa un LLM economico (Haiku su cloud, o il modello base su Ollama) per distillare il risultato in 100–200 token prima di inserirlo nel context.

# tool_result_summarizer.py - comprime output verbosi dei tool
# Compatibile con Ollama on-prem e API Anthropic

import httpx
import json
from typing import Any

# Configurazione: usa Ollama locale se disponibile, fallback a nessuna compressione
OLLAMA_BASE = "http://localhost:11434"
SUMMARIZER_MODEL = "llama3.1:8b"  # modello economico per summarization

async def summarize_tool_result(
    tool_name: str,
    tool_result: Any,
    agent_current_goal: str,
    max_output_tokens: int = 150,
) -> str:
    """
    Distilla un tool result verboso in un summary compatto.
    Usa il LLM locale Ollama per evitare costi aggiuntivi.

    Args:
        tool_name: nome del tool (usato nel prompt di summarization)
        tool_result: output raw del tool (stringa o oggetto JSON)
        agent_current_goal: obiettivo corrente dell'agent (guida il summary)
        max_output_tokens: lunghezza massima del summary in token approssimativi

    Returns:
        Summary compatto del tool result.
    """
    result_text = (
        json.dumps(tool_result, ensure_ascii=False)
        if not isinstance(tool_result, str)
        else tool_result
    )

    # Non comprimere se già breve (sotto i 200 token approssimativi)
    if len(result_text.split()) < 150:
        return result_text

    prompt = f"""Stai aiutando un AI agent. Il tool '{tool_name}' ha restituito il seguente output.
L'obiettivo corrente dell'agent è: {agent_current_goal}

Output del tool:
{result_text[:3000]}  # tronca a 3000 char per sicurezza

Estrai SOLO le informazioni rilevanti per l'obiettivo dell'agent in massimo {max_output_tokens} parole.
Mantieni numeri, nomi, date e valori specifici. Ometti testo boilerplate e dettagli non pertinenti.
Rispondi SOLO con il summary, senza introduzioni."""

    try:
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.post(
                f"{OLLAMA_BASE}/api/generate",
                json={
                    "model": SUMMARIZER_MODEL,
                    "prompt": prompt,
                    "stream": False,
                    "options": {"num_predict": max_output_tokens + 20},
                },
            )
            data = response.json()
            return data.get("response", result_text).strip()
    except Exception:
        # Fallback: restituisce il result originale se Ollama non è disponibile
        return result_text[:500] + "... [troncato]"


def compress_conversation_memory(
    conversation_history: list[dict[str, Any]],
    window_size: int = 10,
    keep_recent: int = 4,
) -> list[dict[str, Any]]:
    """
    Comprime i turn di conversazione più vecchi in un summary JSON strutturato.
    Applica sliding window: ogni 'window_size' turn, comprime i più vecchi.

    Args:
        conversation_history: history completa della conversazione
        window_size: ogni quanti turn attivare la compressione
        keep_recent: quanti turn recenti mantenere espansi

    Returns:
        History compressa: summary JSON + turn recenti.
    """
    if len(conversation_history) < window_size:
        return conversation_history  # non ancora al trigger

    # Messaggi da comprimere: tutti tranne gli ultimi 'keep_recent'
    to_compress = conversation_history[:-keep_recent]
    to_keep = conversation_history[-keep_recent:]

    # Costruisci summary strutturato dei turn più vecchi
    summary_lines = []
    for msg in to_compress:
        role = msg.get("role", "unknown")
        content = msg.get("content", "")
        if isinstance(content, str) and content:
            # Tronca a 100 char per entry nel summary
            summary_lines.append(f"[{role}]: {content[:100]}")

    summary = {
        "type": "compressed_history",
        "turns_compressed": len(to_compress),
        "summary": "; ".join(summary_lines[:20]),  # max 20 entry nel summary
    }

    # Inserisci il summary come primo messaggio sintetico
    compressed_entry = {
        "role": "user",
        "content": f"[HISTORY SUMMARY]: {json.dumps(summary, ensure_ascii=False)}",
    }

    return [compressed_entry, *to_keep]

La memory compression con sliding window risolve il problema del context che cresce senza limiti nelle conversazioni multi-turn. Il trigger ogni 10 turn è un valore di default ragionevole: abbastanza raro da non introdurre overhead, abbastanza frequente da tenere il context sotto controllo su agent lunghi.

Sezione 6 - Model routing: Haiku per facili, Opus solo per hard

Non tutti gli step di un agent richiedono lo stesso modello. Step semplici - estrarre un campo da un JSON, formattare un output, decidere se chiamare un tool specifico - si risolvono perfettamente con Claude Haiku (o llama3.1:8b su Ollama). Step complessi - ragionamento multi-step, analisi ambigua, pianificazione a lungo termine - richiedono il modello migliore disponibile.

Il model router classifica ogni step prima di eseguirlo e sceglie il modello appropriato. Il classificatore stesso è un LLM leggero - usare il modello costoso per decidere quale modello usare vanifica l'ottimizzazione.

# model_router.py - routing Haiku/Opus per step agent
# Classifica la complessità dello step e sceglie il modello

import anthropic
from enum import Enum
from typing import Any

class TaskComplexity(Enum):
    SIMPLE = "simple"    # Haiku: €0.0025/1K token input
    MEDIUM = "medium"    # Sonnet: €0.003/1K token input
    HARD = "hard"        # Opus: €0.015/1K token input

# Mappa complessità → modello (configurabile)
MODEL_MAP = {
    TaskComplexity.SIMPLE: "claude-haiku-4-5",
    TaskComplexity.MEDIUM: "claude-sonnet-4-5",
    TaskComplexity.HARD: "claude-opus-4-5",
}

# Classificatore: usa Haiku per classificare (costo minimo)
CLASSIFIER_PROMPT = """Analizza il seguente step di un AI agent e classifica la sua complessità.

Step da classificare: {step_description}

Contesto disponibile: {context_summary}

Rispondi con UNA sola parola:
- "simple": estrazione dati, formattazione, decisione binaria, lookup diretto
- "medium": analisi moderata, comparazione 2-3 opzioni, sintesi breve
- "hard": ragionamento multi-step, ambiguità alta, pianificazione complessa, creatività richiesta

Risposta:"""

_client = anthropic.Anthropic()

def classify_step_complexity(
    step_description: str,
    context_summary: str = "",
) -> TaskComplexity:
    """
    Classifica la complessità di uno step usando Haiku (costo minimo).
    Fallback a MEDIUM in caso di errore classificazione.
    """
    prompt = CLASSIFIER_PROMPT.format(
        step_description=step_description[:500],  # limita input classificatore
        context_summary=context_summary[:200],
    )
    try:
        response = _client.messages.create(
            model="claude-haiku-4-5",  # sempre Haiku per classificare
            max_tokens=10,  # risposta brevissima: "simple"/"medium"/"hard"
            messages=[{"role": "user", "content": prompt}],
        )
        label = response.content[0].text.strip().lower()
        return TaskComplexity(label) if label in TaskComplexity._value2member_map_ else TaskComplexity.MEDIUM
    except Exception:
        return TaskComplexity.MEDIUM  # fallback sicuro

def get_model_for_step(
    step_description: str,
    context_summary: str = "",
    force_complexity: TaskComplexity | None = None,
) -> tuple[str, TaskComplexity]:
    """
    Ritorna (model_id, complexity) per il prossimo step.
    force_complexity bypassa la classificazione (utile per override manuali).
    """
    complexity = force_complexity or classify_step_complexity(step_description, context_summary)
    model = MODEL_MAP[complexity]
    return model, complexity

Il saving di model routing su un agent reale dipende dalla distribuzione dei task. Se il 70% degli step sono classificati come "simple" o "medium", il risparmio rispetto all'uso esclusivo di Opus è significativo. Nel benchmark di fine articolo mostro i numeri concreti sul ml-service del portfolio.

Sezione 7 - Speculative execution e batching tool calls asyncio

La speculative execution è un pattern mutuato dai processori hardware: esegui prima con il modello cheap (speculazione), verifica con il modello costoso solo se il risultato sembra incerto. Se la confidence della speculazione è alta, accetti la risposta cheap senza invocare il modello costoso. Il saving è massimo su workload dove la maggioranza dei task è "facile".

# speculative_execution.py - cheap model first, verify solo se necessario

import anthropic
from dataclasses import dataclass

_client = anthropic.Anthropic()

@dataclass
class SpeculativeResult:
    answer: str
    used_cheap: bool         # True se accettata la risposta cheap
    confidence: float        # 0.0–1.0, stimata dal cheap model
    verification_skipped: bool

CHEAP_MODEL = "claude-haiku-4-5"
VERIFY_MODEL = "claude-sonnet-4-5"
CONFIDENCE_THRESHOLD = 0.85  # sopra questa soglia, accetta senza verifica

def run_speculative_step(
    prompt: str,
    confidence_threshold: float = CONFIDENCE_THRESHOLD,
) -> SpeculativeResult:
    """
    Esegue uno step con speculative execution.
    1. Chiama il modello cheap
    2. Stima la confidence della risposta
    3. Se confidence >= threshold, accetta senza verifica
    4. Altrimenti, verifica con il modello costoso

    La confidence è stimata chiedendo al cheap model stesso di valutarsi.
    """
    # Step 1: speculazione con modello cheap
    cheap_response = _client.messages.create(
        model=CHEAP_MODEL,
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )
    cheap_answer = cheap_response.content[0].text

    # Step 2: stima confidence (auto-valutazione del cheap model)
    confidence_prompt = f"""Hai appena risposto a questa domanda:
Domanda: {prompt[:300]}
Risposta: {cheap_answer[:300]}

Su una scala da 0.0 a 1.0, quanto sei sicuro di questa risposta?
Considera: chiarezza della domanda, disponibilità di informazioni nel context,
assenza di ambiguità. Rispondi SOLO con un numero (es: 0.87)."""

    conf_response = _client.messages.create(
        model=CHEAP_MODEL,
        max_tokens=10,
        messages=[{"role": "user", "content": confidence_prompt}],
    )
    try:
        confidence = float(conf_response.content[0].text.strip())
        confidence = max(0.0, min(1.0, confidence))  # clamp [0,1]
    except ValueError:
        confidence = 0.5  # fallback neutro

    # Step 3: accetta speculazione o verifica
    if confidence >= confidence_threshold:
        return SpeculativeResult(
            answer=cheap_answer,
            used_cheap=True,
            confidence=confidence,
            verification_skipped=True,
        )

    # Step 4: verifica con modello costoso
    verify_response = _client.messages.create(
        model=VERIFY_MODEL,
        max_tokens=1024,
        messages=[
            {"role": "user", "content": prompt},
            {"role": "assistant", "content": cheap_answer},
            {
                "role": "user",
                "content": "Rivedi e correggi la risposta precedente se necessario. "
                           "Se è corretta, confermala senza modifiche.",
            },
        ],
    )
    return SpeculativeResult(
        answer=verify_response.content[0].text,
        used_cheap=False,
        confidence=confidence,
        verification_skipped=False,
    )


# batching_tool_calls.py - esecuzione parallela di tool call indipendenti
import asyncio
from typing import Callable, Awaitable

async def batch_tool_calls(
    tool_calls: list[dict],
    tool_executor: Callable[[str, dict], Awaitable[str]],
    max_concurrent: int = 5,
) -> list[str]:
    """
    Esegue tool call indipendenti in parallelo con asyncio.Semaphore.
    Riduce il round-trip totale da N×latency a max(latencies).

    Args:
        tool_calls: lista di {"name": str, "input": dict}
        tool_executor: funzione async che esegue un singolo tool call
        max_concurrent: limite di concorrenza (evita rate limit)

    Returns:
        Lista ordinata dei risultati (stesso ordine dei tool_calls in input).
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    async def execute_with_semaphore(call: dict) -> str:
        async with semaphore:
            return await tool_executor(call["name"], call["input"])

    # Esegui tutti in parallelo, mantieni ordine
    results = await asyncio.gather(
        *[execute_with_semaphore(call) for call in tool_calls],
        return_exceptions=True,
    )

    # Converti exception in messaggio di errore strutturato
    return [
        str(r) if isinstance(r, Exception) else r
        for r in results
    ]

Il batching è spesso il quick win più sottovalutato. Se un agent ReAct chiama 4 tool in sequenza - ciascuno con 200ms di latenza - la durata totale è 800ms. Con batching parallelo, sono 200ms. Per tool con latenza più alta (query DB, chiamate API esterne), il risparmio in latenza si traduce anche in costo se il provider LLM addebita su durata sessione.

Sezione 8 - Benchmark before/after: agent ml-service portfolio

Ho applicato queste tecniche all'agent RAG del ml-service del mio portfolio - il sistema che risponde a query sui contenuti del sito usando Ollama on-prem con llama3.1:8b. Qui il costo monetario è €0, ma l'ottimizzazione è ugualmente critica: su un VPS Hetzner CCX13 (4 vCPU, 8GB RAM), ogni ciclo agent che processa token inutili è latenza in più e throughput in meno.

Il risultato più sorprendente è il +3% sul success rate. Context pruning ha rimosso messaggi ridondanti che confondevano il modello, non solo token inutili. Il VPS ora gestisce 8 run concorrenti invece di 2, il che significa poter servire picchi di traffico senza upgrade hardware. Il costo monetario rimane €0 (Ollama on-prem), ma il costo computazionale è sceso drasticamente.

Per il caso d'uso cloud (Claude API), applicando anche prompt caching e model routing, lo stesso pattern di ottimizzazione porterebbe il costo da circa €0.85/run a €0.16/run - un saving dell'81% su un agent di media complessità.

FAQ

Quanto costa davvero un LLM agent in produzione senza ottimizzazioni?

Un agent ReAct con Claude Sonnet 3.7 su 20 step, context 8K token per step e nessuna ottimizzazione consuma tra 160K e 240K token per run. Al prezzo di $3/MTok input, sono €0.60–€1.20/run. Moltiplicato per 1000 run/giorno = €600–€1200/giorno. Con le 7 tecniche combinate si scende a €120–€250/giorno.

Il prompt caching Anthropic funziona anche con modelli non Claude?

No, è una feature specifica delle API Anthropic (Claude 3.5+ e 3.7). Per OpenAI esiste Automatic Prompt Caching su GPT-4o. Per Ollama on-prem non esiste prompt caching nativo, ma il costo è €0: il valore principale viene da context pruning e model routing che riducono i cicli computazionali.

Come scelgo la soglia di similarità per il context pruning?

Inizia a 0.80 e misura il success rate su un evaluation set di golden examples. Se rimane stabile, scendi a 0.75. Non scendere mai sotto 0.60 senza test approfonditi: rischi di rimuovere messaggi critici che l'agent usa per mantenere coerenza. Il recency boost (ultimi N messaggi sempre inclusi) è il tuo safety net più importante.

Posso combinare tutte e 7 le tecniche nello stesso agent?

Sì, in pipeline stratificata: batching → model routing → prompt caching → context pruning → tool summarization → memory compression → speculative execution. Il saving cumulativo è maggiore della somma dei singoli. Implementa nell'ordine indicato: le prime tecniche sono le più semplici e producono quick win immediati.

Come misuro la quality loss dovuta alle ottimizzazioni?

Costruisci un evaluation set di 50-100 task con ground-truth esito (success/fail semantico, non output letterale). Esegui baseline e ottimizzato. Se il success rate cala di più del 3-5%, l'ottimizzazione è troppo aggressiva. Se rimane stabile o migliora (meno rumore nel context), è valida.

La speculative execution funziona per tutti i tipi di task agent?

No. È vantaggiosa solo se la distribuzione è asimmetrica: almeno 60-70% di task "facili" risolvibili dal cheap model. Se il tuo agent lavora prevalentemente su task complessi, il costo di verificare con il secondo modello supera il risparmio. Misura la distribuzione prima di implementarla.

Con Ollama on-prem a costo zero, queste ottimizzazioni hanno senso?

Sì, per ragioni diverse. Il costo monetario è zero, ma il costo computazionale (CPU/GPU cycles, latency, throughput) è reale. Ridurre i token processati per ciclo significa meno latenza per step, maggiore throughput concorrente e minor rischio di saturare la context window del modello locale. Sul VPS Hetzner CCX13 del portfolio, ho moltiplicato per 4 il numero di run concorrenti possibili senza upgrade hardware.

Conclusione

Il costo di un LLM agent non è fisso: è il risultato delle scelte di implementazione. System prompt non cachato, tool result non compressi, model costoso usato per step banali, tool call in sequenza quando potrebbero essere parallele - ogni scelta sbagliata moltiplica il costo per il numero di step e di run. Le 7 tecniche presentate in questo episodio non sono ottimizzazioni premature: sono pratiche di engineering solide che producono saving reali e, spesso, migliorano anche la quality riducendo il rumore nel context.

L'approccio raccomandato è incrementale: inizia da prompt caching e context pruning (basso sforzo, alto saving), poi aggiungi tool summarization e memory compression, infine model routing e speculative execution quando hai dati sulla distribuzione del workload. Il benchmark before/after su evaluation set reali è l'unico modo per sapere dove sei e dove arrivi - usa le metriche presentate nella sezione 8 come template.

Per il caso Ollama on-prem come nel ml-service del portfolio, l'obiettivo non è tagliare una bolletta ma massimizzare il throughput su hardware limitato. Le stesse tecniche producono risultati diversi ma ugualmente preziosi: +300% di run concorrenti sullo stesso VPS è un upgrade hardware gratuito.

Prossimo episodio: EP04 - Memoria agent: vector store vs conversation buffer vs summary. Come costruire i 3 layer di memoria di un agent (working, episodic, semantic) e quando scegliere ciascuno. Approfondimento su observability delle ottimizzazioni di costo: come tracciare cache hit rate e token savings con OpenTelemetry.

Se ti interessa come questi pattern di AI engineering si traducono in workflow quotidiano su un codebase reale, ho documentato il mio setup (skill design + planning fasato + sub-agent orchestration) in Claude Code Workflow: Superpowers + GSD framework.