Ottimizzazione dei costi per i sistemi LLM: dove vanno davvero i soldi

Usa i token dove contano davvero.

Indice

I costi degli LLM scalano linearmente con l’utilizzo. Un sistema che elabora 10.000 richieste al giorno a $0,01 per richiesta costa $100 al giorno — $365 all’anno. A livello enterprise, si superano i $10.000.

L’ottimizzazione dei costi non significa tagliare gli angoli. Significa spendere i token dove contano davvero.

Ogni token che sprechi è un token che avresti potuto spendere per una risposta migliore.

Strategie di ottimizzazione dei costi LLM

Gestione del budget dei token

Il modo più semplice per controllare i costi è impostare dei limiti. Per sessione, per compito o per giorno.

Strategia 1: Budget per sessione

I budget per sessione sono diretti:

class SessionBudget:
    def __init__(self, budget_tokens: int = 10000):
        self.budget = budget_tokens
        self.used = 0

    def allocate(self, tokens: int) -> bool:
        if self.used + tokens <= self.budget:
            self.used += tokens
            return True
        return False

    def remaining(self) -> int:
        return self.budget - self.used

Strategia 2: Budget per compito

I budget per compito sono più utili. Diversi compiti richiedono quantità diverse di contesto:

task_budgets:
  classify:
    max_tokens: 100
    model: qwen2.5-1.5b
  summarize:
    max_tokens: 500
    model: qwen2.5-7b
  code_review:
    max_tokens: 2000
    model: qwen2.5-coder-7b
  reason:
    max_tokens: 4000
    model: qwen2.5-32b

Strategia 3: Budget adattivi

I budget adattivi si regolano in base a ciò che accade realmente. Se i compiti di classificazione utilizzano costantemente 80 token, smetti di allocarne 100:

class AdaptiveBudget:
    def __init__(self):
        self.task_history = {}

    def allocate(self, task_type: str) -> int:
        if task_type in self.task_history:
            return int(self.task_history[task_type] * 1.5)
        return 1000

    def record(self, task_type: str, tokens_used: int):
        if task_type not in self.task_history:
            self.task_history[task_type] = tokens_used
        else:
            self.task_history[task_type] = (
                0.9 * self.task_history[task_type] + 0.1 * tokens_used
            )

La media mobile esponenziale (peso 0,9) significa che l’utilizzo recente conta più della storia. Regola il peso in base a quanto sono volatili i tuoi carichi di lavoro.

Inferenza API vs locale

L’inferenza locale è più economica su larga scala. Il punto di pareggio dipende dal tuo hardware e dalle tariffe API.

Modello API ($/M token) Costo locale/ora Pareggio
GPT-4o $2,50 / $10,00 N/D
Claude Sonnet 4 $3,00 / $15,00 N/D
Qwen2.5-72B $0,50 / $2,00 ~$0,50 ~4 ore/giorno
Qwen2.5-32B $0,30 / $1,20 ~$0,20 ~2 ore/giorno
Qwen2.5-7B $0,10 / $0,40 ~$0,05 ~1 ora/giorno

I calcoli sull’hardware:

Hardware Costo iniziale Elettricità mensile Pareggio vs API
RTX 3090 (usata) $600 $15 ~4 mesi
RTX 4090 $1.500 $20 ~6 mesi
RTX 5080 $1.000 $18 ~5 mesi
DGX Spark $2.000 $30 ~8 mesi

Con un utilizzo moderato — un’ora o più al giorno — l’inferenza locale si ripaga da sola. Con un alto utilizzo, i risparmi sono drammatici. Il rovescio della medaglia è il capitale iniziale. Una RTX 5080 costa $1.000. Una fattura API puoi metterla in pausa. L’hardware no.

Strategie di fallback

Quando il modello preferito è troppo costoso o troppo lento, passa a qualcosa di più economico. La chiave è sapere quando la qualità è “abbastanza buona”.

Strategia 1: Fallback basato sulla qualità

Il fallback basato sulla qualità prova i modelli fino a quando l’output non soddisfa una soglia:

class QualityFallback:
    def __init__(self, quality_threshold: float = 0.8):
        self.threshold = quality_threshold
        self.models = [
            {"model": "claude-sonnet-4", "cost": 0.015},
            {"model": "qwen2.5-72b", "cost": 0.002},
            {"model": "qwen2.5-32b", "cost": 0.001},
            {"model": "qwen2.5-7b", "cost": 0.0004},
        ]

    def route(self, prompt: str) -> str:
        for model_config in self.models:
            result = self.call_model(model_config["model"], prompt)
            if self.evaluate_quality(result) >= self.threshold:
                return result
        return self.call_model(self.models[0]["model"], prompt)

Il problema è la valutazione stessa. Come misuri la qualità senza chiamare un altro modello? Alcuni sistemi usano un classificatore piccolo. Altri usano controlli euristici — lunghezza, struttura, presenza di parole chiave. Nessuno di questi è perfetto.

Strategia 2: Fallback basato sulla latenza

Il fallback basato sulla latenza è più semplice. Instrada verso il modello più veloce che soddisfa il tuo budget di tempo:

class LatencyFallback:
    def __init__(self, max_latency: float = 5.0):
        self.max_latency = max_latency
        self.models = [
            {"model": "qwen2.5-1.5b", "latency": 0.5},
            {"model": "qwen2.5-7b", "latency": 2.0},
            {"model": "qwen2.5-32b", "latency": 10.0},
            {"model": "claude-sonnet-4", "latency": 5.0},
        ]

    def route(self, prompt: str) -> str:
        for model_config in sorted(self.models, key=lambda x: x["latency"]):
            if model_config["latency"] <= self.max_latency:
                return self.call_model(model_config["model"], prompt)
        return self.call_model(self.models[0]["model"], prompt)

Caching

La memorizzazione nella cache è l’ottimizzazione dei costi più sottovalutata. I prompt identici accadono più spesso di quanto pensi — richieste di classificazione, query stile FAQ, chiamate agli strumenti ripetute.

Strategia 1: Cache dei prompt

La cache esatta dei prompt è semplice:

import hashlib

class PromptCache:
    def __init__(self, max_size: int = 1000):
        self.cache = {}
        self.max_size = max_size

    def get(self, prompt: str) -> str | None:
        key = hashlib.sha256(prompt.encode()).hexdigest()
        return self.cache.get(key)

    def set(self, prompt: str, response: str):
        key = hashlib.sha256(prompt.encode()).hexdigest()
        if len(self.cache) >= self.max_size:
            self.cache.pop(next(iter(self.cache)))
        self.cache[key] = response

Strategia 2: Cache semantica

La cache semantica è più utile. Cattura i prompt che sono diversi ma significano la stessa cosa:

from sentence_transformers import SentenceTransformer

class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.95):
        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.cache = {}
        self.threshold = similarity_threshold

    def get(self, prompt: str) -> str | None:
        prompt_embedding = self.model.encode([prompt])[0]
        for cached_prompt, cached_response in self.cache.items():
            cached_embedding = self.model.encode([cached_prompt])[0]
            similarity = self.cosine_similarity(
                prompt_embedding, cached_embedding
            )
            if similarity >= self.threshold:
                return cached_response
        return None

    def set(self, prompt: str, response: str):
        self.cache[prompt] = response

La soglia è importante. 0,95 è aggressiva — solo prompt molto simili corrispondono. 0,85 è più permissiva ma rischia di restituire risposte errate. Misura il tuo tasso di mancata corrispondenza e regola.

La memorizzazione nella cache delle risposte per le query comuni vale comunque la pena. Se gli utenti chiedono ripetutamente “com’è il meteo” o “che ora è”, memorizza nella cache il modello, non solo il prompt esatto:

class ResponseCache:
    def __init__(self):
        self.common_queries = {
            "what is the weather": "Check weather API",
            "what is the time": "Check system time",
            "who is the president": "Check current president",
        }

    def get(self, query: str) -> str | None:
        query_lower = query.lower()
        for common_query, response in self.common_queries.items():
            if common_query in query_lower:
                return response
        return None

Non è sofisticato, ma funziona. Le query comuni sono comuni per un motivo.

Quando l’ottimizzazione aiuta

L’ottimizzazione è importante quando si elaborano grandi volumi, si eseguono carichi di lavoro misti o si pagano costi API che si accumulano.

Non è importante quando si sta prototipando, si usa un singolo modello o si elaborano piccoli volumi. La complessità della gestione del budget, del fallback e della cache non ne vale la pena per un sistema che effettiva 100 richieste al giorno.

Fai prima funzionare il flusso di base. Aggiungi l’ottimizzazione quando arriva la fattura.

Compromessi

Strategia Costo Qualità Complessità
Nessuna ottimizzazione Più alto Costante Più bassa
Gestione budget token Moderato Variabile Media
Modelli di fallback Basso-Medio Variabile Media
Caching Più basso Alta (per i colpi di cache) Media
Ibrido Ottimizzato Ottimizzato Più alta

I sistemi di produzione di solito sono ibridi. Budget per sessione, fallback basato su qualità o latenza, cache di tutto ciò che puoi. La complessità è reale, ma anche i risparmi.

Correlati

Iscriviti

Ricevi nuovi articoli su sistemi, infrastruttura e ingegneria AI.