Cost optimization agent: reduce token spend 80% without losing quality

Cost optimization agent: reduce token spend by 80% without losing quality

A Claude Sonnet 3.7 without optimizations, on a document analysis task with 20 steps and full context, consumes between 160K to 240K tokens per run. At the current pricing - $3 per million input tokens, $15 per million output tokens - each run costs between $0.60 to $1.20. Multiply by 1000 runs per day and you get a daily cost of $600–$1200 before even considering spikes. This is the number that scares teams and pushes many back to one-shot solutions, giving up on the real benefits of agents. Note: I preserved the formatting, punctuation, and capitalization style as in the original text, including the proper nouns (e.g. "Claude Sonnet") and HTML inline tags.

The problem isn't that the agents cost too much. The problem is that most of the tokens are wasted: system prompts repeated on every call without caching, verbose tool results that clog up the context, easy steps executed with the most expensive model, tool calls executed in sequence when they could be parallel. These waste products aren't inevitable - they're non-optimized implementation patterns that get corrected with 7 specific techniques. Note: I preserved the formatting and punctuation of the original text, as well as the token XPHX0XPHX (which appears to be a placeholder or a code snippet).

episode, I present each technique with a working Python code snippet, the estimated saving percentage and implementation effort. I conclude with a before/after benchmark on a real ml-service agent from my portfolio - where Ollama on-prem eliminates monetary costs but makes cycle reduction even more critical on a VPS with limited resources. Note: I preserved the token "Ollama" as-is, assuming it's a proper noun or a brand/product name. If you'd like me to translate it, please let me know!

Section 1 - Token budget per step: where waste is hidden

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: sent to every LLM call. On an agent with 20 steps and a system prompt of 500 tokens, this contributes only 10,000 fixed overhead tokens.
• Tool definitions: Each tool definition occupies 50-200 tokens depending on the complexity of the JSON schema. With 10 defined tools, there are 500-2000 repeated tokens at each step.
• Conversation history: grows linearly with each step. At step 20, the history contains messages 1–19. Without pruning, context doubles every ~10 steps.
• Verbose tool result: 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

Execute 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.

Section 2 - The 7 Techniques with Expected Saving and Effort

The following table summarizes the 7 techniques with expected cost savings percentage on the total agent run cost, implementation effort (low/medium/high) and prerequisites. The saving is estimated on a ReAct agent of 20 steps with an 8K token context per step - numbers vary based on specific workload. Note: I preserved the original formatting, punctuation, and capitalization style, as well as the proper noun "ReAct" and the HTML inline tag "" (not present in this case).

The first four techniques combine well in a sequential pipeline. Model routing and speculative execution require more initial investment but produce the most significant saving on heterogeneous workloads. Batching is often the most underappreciated quick win: it does not reduce tokens, but reduces the number of LLM calls, which can halve latency and reduce overhead cost on agents with many parallel tool calls.

Section 3 - Prompt caching Anthropic: structuring cache blocks for 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.

The critical rule is that the cached content must be a Note: I've preserved the XPHX0XPHX token exactly as-is. Let me know if you need any further assistance! 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 To check that the caching is actually kicking in, verify it in the response.

Section 4 - Context Pruning with 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.

The principle is simple: embed the current message, embed every message in the history, calculate cosine similarity, filter out messages below a threshold. Recent messages receive an artificial boost because they tend to be more relevant even if semantically distant from the original target.

# 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 ensures that the last 3 messages are always included, regardless of similarity. This is important to maintain the immediate coherence of the agent: an agent that does not remember the tool result of the last step often makes the same call twice, thus nullifying the savings.

Sezione 5 - Tool result summarization e memory compression

When a tool returns verbose data - a search response with 5 chunks, the output of an API with 50 JSON fields, the text of a long document - most of that output is not needed by the agent in the next step. The result summarization tool uses a cost-effective LLM (Haiku on cloud, or the base model on Ollama) to distill the result into 100–200 tokens before inserting it into 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]

Memory compression with a sliding window resolves the problem of unlimited growing context in multi-turn conversations. A trigger every 10 turns is a reasonable default value: frequent enough to keep the context under control on long agents, yet rare enough not to introduce overhead.

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

Not all steps of an agent require the same model. Simple steps – extracting a field from a JSON, formatting an output, deciding whether to call a specific tool – are perfectly resolved with Claude Haiku (or llama3.1:8b on Ollama). Complex steps – multi-step reasoning, ambiguous analysis, long-term planning – require the best available model.

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.

Section 7 - Speculative Execution and 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.

The most surprising result is the +3% success rate. Context pruning removed redundant messages that confused the model, not just unnecessary tokens. The VPS now manages 8 concurrent runs instead of 2, which means it can handle traffic spikes without hardware upgrades. The monetary cost remains €0 (Ollama on-prem), but the costo computazionale è sceso drasticamente.

For the cloud usage case (Claude API), applying also prompt caching and model routing, the same optimization pattern would bring the cost from approximately €0.85/run to €0.16/run – an 81% saving on a medium-complexity agent.

Frequently Asked Questions

Quanto costa davvero un LLM agent in produzione senza ottimizzazioni?

A ReAct agent with Claude Sonnet 3.7 on 20 steps, context of 8K tokens per step and no optimization consumes between 160K to 240K tokens per run. At the price of $3/MTok input, this is €0.60–€1.20/run. Multiplying by 1000 runs/day = €600–€1200/day. With the combined 7 techniques it drops down to €120–€250/day.

The Anthropic prompt caching works also with non-Claude models?

No, it is a specific feature of the Anthropic APIs (Claude 3.5+ and 3.7). For OpenAI there exists Automatic Prompt Caching on GPT-4. For Ollama on-premises, native prompt caching does not exist, but the cost is €0: the main value comes from context pruning and model routing that reduce computational cycles.

How do I choose the similarity threshold for context pruning?

Start at 0.80 and measure the success rate on a golden examples evaluation set. If it remains stable, drop to 0.75. Never go below 0.60 without thorough testing: you risk removing critical messages that the agent uses to maintain coherence. The recency boost (last N messages always included) is your most important safety net.

Posso combinare tutte e 7 le tecniche nello stesso agent?

Yes, in a pipeline with layers: batching → model routing → prompt caching → context pruning → tool summarization → memory compression → speculative execution. The cumulative saving is greater than the sum of individual ones. Implement in the order indicated: the first techniques are the simplest and produce immediate quick wins. Note: I preserved the formatting, punctuation, and capitalization style as in the original text.

Come misuro la quality loss dovuta alle ottimizzazioni?

Build an evaluation set of 50-100 tasks with ground-truth outcome (semantic success/fail, not literal output). Run baseline and optimized. If the success rate drops by more than 3-5%, the optimization is too aggressive. If it remains stable or improves (less noise in context), it's valid.

Speculative execution works for all types of agent tasks?

It's only advantageous if the distribution is asymmetric: at least 60-70% of "easy" tasks resolvable by the cheap model. If your agent works primarily on complex tasks, the cost of verifying with the second model exceeds the savings. Measure the distribution before implementing it.

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

Yes, for different reasons. The monetary cost is zero, but the computational cost (CPU/GPU cycles, latency, throughput) is real. Reducing tokens processed per cycle means less latency per step, higher concurrent throughput, and lower risk of saturating the context window of the local model. On the Hetzner CCX13 VPS in the portfolio, I multiplied the number of concurrent runs possible without hardware upgrade by 4.

Conclusion

The cost of an LLM agent is not fixed: it's the result of implementation choices. Uncached system prompt, uncompressed tool results, expensive models used for mundane steps, sequential tool calls when they could be parallel - every wrong choice multiplies the cost by the number of steps and runs. The 7 techniques presented in this episode are not premature optimizations: they're solid engineering practices that produce real savings and often improve quality by reducing noise in the context. Note: LLM stands for Large Language Model, which is a type of artificial intelligence model used for natural language processing tasks.

The recommended approach is incremental: start with prompt caching and context pruning (low effort, high saving), then add tool summarization and memory compression, finally model routing and speculative execution when you have data on the workload distribution. The before/after benchmark on a real evaluation set is the only way to know where you are and where you're going - use the metrics presented in section 8 as a template.

For the Ollama on-prem case, similar to the ml-service in the portfolio, the goal is not to cut a bill but to maximize throughput on limited hardware. The same techniques produce different but equally valuable results: +300% concurrent runs on the same VPS is free hardware upgrade.

Next episode: EP04 - Agent Memory: 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.