12 - FinOps: Cost Optimization per Agenti e Token LLM

Il problema: la spesa cresce con le iterazioni, non con le richieste

Un'applicazione che chiama un modello una volta per richiesta ha un costo prevedibile: token in ingresso più token in uscita, per il numero di richieste. Un agente rompe questa proporzione perché dentro ogni compito c'è un ciclo, e a ogni giro il modello riceve di nuovo tutto ciò che è successo prima. Il costo di un compito dipende quindi dalla sua lunghezza, e cresce più in fretta di quanto la lunghezza faccia pensare. Controllare la spesa, in un agente, significa agire su quattro leve: quanto contesto si rimanda, quanto di quel contesto si può riusare a prezzo ridotto, quale modello risponde, e quando ci si ferma.

Un avvertimento sui numeri, che vale per tutto l'articolo. I prezzi dei modelli cambiano spesso e dipendono dal fornitore, e qui non c'è nessun listino: i listati usano prezzi inventati per l'esempio, dichiarati come tali nel codice, e i risultati sono l'aritmetica di quei parametri, non una misura di un servizio reale. Cambia i parametri con quelli del tuo contratto e riesegui. I listati sono Python con la sola libreria standard, eseguiti due volte con uscita identica; nessun modello è stato chiamato e nessuna chiamata a pagamento è stata fatta.

Per il contesto della serie: l'articolo precedente ha mostrato come raccogliere le metriche, compresi i token consumati; qui si decide che cosa farne. Il successivo mette insieme le parti in un esempio costruito.

Perché un compito lungo costa molto più di un compito corto

Il meccanismo è un'addizione. Alla prima iterazione il modello riceve il prompt di sistema e le definizioni degli strumenti (il prefisso). Alla seconda riceve il prefisso più ciò che l'iterazione uno ha aggiunto: la risposta del modello e il risultato dello strumento. Alla decima riceve nove aggiunte. Se ogni iterazione aggiunge lo stesso numero di token, i token d'ingresso complessivi crescono con il quadrato del numero di giri, perché la somma 1 + 2 + ... + n vale n(n+1)/2 e non n.

Il listato calcola il costo di un compito da 1, 5, 10 e 20 iterazioni con un prefisso di 2000 token, 700 token aggiunti a ogni giro e 300 generati (valori d'esempio). La terza colonna è il costo senza cache; la quarta usa la cache incrementale, in cui ogni iterazione rilegge a prezzo ridotto ciò che aveva già inviato e paga come scrittura solo la parte nuova. I moltiplicatori 0,10 per la lettura e 1,25 per la scrittura sono quelli documentati da Anthropic per la cache con scadenza di cinque minuti; altri fornitori hanno regole diverse.

# Prezzi INVENTATI per l'esempio, in dollari per milione di token: non sono un listino.
INPUT_PER_M = 3.00
OUTPUT_PER_M = 15.00
LETTURA_CACHE = 0.10   # moltiplicatore documentato da Anthropic per le letture
SCRITTURA_CACHE = 1.25 # moltiplicatore per la scrittura con scadenza di 5 minuti

PREFISSO = 2000   # prompt di sistema + definizioni degli strumenti
PASSO = 700       # token che ogni iterazione aggiunge alla cronologia
USCITA = 300      # token generati a ogni iterazione


def dollari(token_in, token_out, moltiplicatore=1.0):
    return token_in * INPUT_PER_M * moltiplicatore / 1e6 + token_out * OUTPUT_PER_M / 1e6


def senza_cache(n):
    tot_in = sum(PREFISSO + i * PASSO for i in range(n))
    return tot_in, dollari(tot_in, n * USCITA)


def con_cache_incrementale(n):
    """Ogni iterazione rilegge dalla cache ciò che aveva già e scrive solo il nuovo."""
    costo = n * USCITA * OUTPUT_PER_M / 1e6
    gia_in_cache = 0
    for i in range(n):
        ingresso = PREFISSO + i * PASSO
        letti = gia_in_cache
        nuovi = ingresso - letti
        costo += (letti * LETTURA_CACHE + nuovi * SCRITTURA_CACHE) * INPUT_PER_M / 1e6
        gia_in_cache = ingresso
    return costo


print(f"{'iter':>4} {'token in':>9} {'senza cache':>12} {'con cache':>10} {'rapporto vs 1 iter':>19}")
base = senza_cache(1)[1]
for n in (1, 5, 10, 20):
    tin, c = senza_cache(n)
    print(f"{n:>4} {tin:>9} {c:>11.4f}$ {con_cache_incrementale(n):>9.4f}$ {c / base:>19.1f}")

# Output:
# iter  token in  senza cache  con cache  rapporto vs 1 iter
#    1      2000      0.0105$    0.0120$                 1.0
#    5     17000      0.0735$    0.0442$                 7.0
#   10     51500      0.1995$    0.0891$                19.0
#   20    173000      0.6090$    0.1947$                58.0

L'ultima colonna è la lezione: venti iterazioni non costano venti volte una sola, ne costano 58. Lo scarto viene dai token d'ingresso, che passano da 2000 a 173.000, mentre i token generati crescono solo linearmente. Nota anche la prima riga: con un solo giro la cache costa di più (0,0120 contro 0,0105), perché si paga la scrittura a 1,25 volte e non si legge mai nulla. La cache ripaga solo se il prefisso viene riletto prima che scada.

Le condizioni in cui il prompt caching conviene sono precise, e conviene verificarle sulla documentazione del fornitore invece di fidarsi di una percentuale di risparmio:

• Prefisso identico: la cache copre il prefisso nell'ordine in cui il fornitore lo costruisce (per Anthropic strumenti, poi sistema, poi messaggi), quindi cambiare un'istruzione all'inizio invalida tutto ciò che segue.
• Riuso entro la scadenza: per impostazione predefinita cinque minuti; una scadenza di un'ora esiste, con scrittura a 2 volte il prezzo di ingresso.
• Lunghezza minima: sotto una soglia che dipende dal modello (la documentazione elenca valori da 512 a 4096 token) la richiesta viene elaborata senza cache e senza errore.
• Controllo sull'uso: i campi cache_creation_input_tokens e cache_read_input_tokens della risposta dicono se la cache ha funzionato; se sono entrambi zero, non è stata usata.

L'ultima voce è la più utile in pratica: una cache che non scatta non dà errori, dà solo una fattura più alta. Per questo il conteggio dei token letti dalla cache va nelle metriche accanto ai token normali, e il rapporto tra i due è l'indicatore da guardare dopo ogni modifica al prompt di sistema.

Un tetto per compito: il registro dei costi

Misurare non basta se il compito continua a girare mentre spende. Il rimedio è un registro che sa quanto ha già speso ogni compito e rifiuta la chiamata che supererebbe il tetto. L'agente a quel punto non fallisce: restituisce ciò che ha ottenuto fino a lì, dichiarato parziale. Il listato implementa il registro con un tetto di 0,50 dollari per compito (valore d'esempio) e tre compiti a copione.

from collections import defaultdict


class BudgetEsaurito(Exception):
    pass


class Registro:
    def __init__(self, tetto_per_compito):
        self.tetto = tetto_per_compito
        self.per_compito = defaultdict(float)
        self.per_agente = defaultdict(float)

    def addebita(self, agente, compito, dollari):
        if self.per_compito[compito] + dollari > self.tetto:
            raise BudgetEsaurito(f"{compito}: {self.per_compito[compito]:.2f}$ spesi, "
                                 f"la chiamata da {dollari:.2f}$ supererebbe {self.tetto:.2f}$")
        self.per_compito[compito] += dollari
        self.per_agente[agente] += dollari


def agente_finto(registro, nome, compito, costi_per_passo):
    risultati = []
    for n, costo in enumerate(costi_per_passo, start=1):
        try:
            registro.addebita(nome, compito, costo)
        except BudgetEsaurito as errore:
            return {"stato": "parziale", "passi": risultati, "motivo": str(errore)}
        risultati.append(f"passo {n}")
    return {"stato": "completo", "passi": risultati}


reg = Registro(tetto_per_compito=0.50)
print(agente_finto(reg, "ricerca", "t1", [0.05, 0.08, 0.10]))
print(agente_finto(reg, "ricerca", "t2", [0.10, 0.20, 0.30, 0.40]))
print(agente_finto(reg, "report", "t3", [0.02, 0.02]))
print({a: round(c, 2) for a, c in sorted(reg.per_agente.items())})

# Output:
# {'stato': 'completo', 'passi': ['passo 1', 'passo 2', 'passo 3']}
# {'stato': 'parziale', 'passi': ['passo 1', 'passo 2'], 'motivo': 't2: 0.30$ spesi, la chiamata da 0.30$ supererebbe 0.50$'}
# {'stato': 'completo', 'passi': ['passo 1', 'passo 2']}
# {'report': 0.04, 'ricerca': 0.53}

Il compito t2 si ferma al terzo passo perché la chiamata da 0,30 avrebbe portato il totale a 0,60: l'eccezione scatta prima di addebitare, quindi la spesa del compito resta sotto il tetto. Il risultato restituito è «parziale» con i passi fatti e il motivo, e spetta a chi chiama decidere se riprovare con più budget. La somma per agente (0,53 per la ricerca) è maggiore del tetto perché somma due compiti, non uno.

Due limiti del listato. Funziona in un solo processo: con più repliche il controllo e l'addebito devono essere un'unica operazione atomica su un archivio condiviso, altrimenti due chiamate simultanee passano entrambe il controllo. E il costo di una chiamata si conosce solo dopo, dai token che il fornitore dichiara nella risposta; il tetto per chiamata (max_tokens) è lo strumento che limita il caso peggiore prima di spendere. I livelli di avviso sul budget giornaliero e mensile, con soglie come il 50% o l'80%, sono una convenzione tua, non una norma: scegli quelle che ti danno il tempo di reagire.

Scegliere il modello: la cascata e il suo punto debole

Non tutte le richieste di un agente hanno la stessa difficoltà, e i modelli grandi costano molte volte i piccoli. La letteratura lo studia da tempo: gli autori di FrugalGPT (arXiv:2305.05176) osservano che i prezzi dei modelli a pagamento possono differire di due ordini di grandezza e descrivono tre famiglie di strategie, tra cui la cascata di modelli, in cui si interroga prima il modello economico e si passa a quello costoso solo se la risposta non convince. Nei loro esperimenti, riportati nell'abstract, una cascata appresa eguaglia il migliore dei modelli singoli con una riduzione di costo fino al 98%, o lo supera di 4 punti percentuali di accuratezza a parità di costo. È un risultato degli autori, su loro dataset e prezzi del 2023, non una promessa.

Tutto dipende da come si decide che la risposta del modello piccolo non convince. Nel listato la decisione è una confidenza dichiarata dal modello, con una soglia. I 20 casi, le difficoltà e il comportamento del modello piccolo sono un copione che ho scritto io: il piccolo risponde bene fino alla difficoltà 2, è onesto sui casi più difficili tranne uno (la difficoltà 3, dove si dichiara sicuro pur sbagliando) e il grande è sempre corretto. Il costo è in unità relative (1 contro 15).

# 20 richieste con una difficoltà da 1 a 4 (dati inventati per l'esempio, seme non necessario)
RICHIESTE = [1, 1, 1, 2, 1, 2, 1, 1, 3, 1, 2, 1, 1, 4, 1, 2, 3, 1, 1, 2]
COSTO = {"piccolo": 1, "grande": 15}   # unità relative, non dollari


def modello_piccolo(difficolta):
    """Copione: risponde bene fino alla difficoltà 2; la confidenza dichiarata è quasi sempre onesta,
    ma sul caso 3 si dichiara sicuro quando non dovrebbe."""
    giusta = difficolta <= 2
    confidenza = 0.9 if difficolta <= 2 else (0.85 if difficolta == 3 else 0.3)
    return giusta, confidenza


def tutto_grande():
    return len(RICHIESTE) * COSTO["grande"], 0


def a_cascata(soglia):
    costo = errori = 0
    for d in RICHIESTE:
        giusta, conf = modello_piccolo(d)
        costo += COSTO["piccolo"]
        if conf >= soglia:
            errori += 0 if giusta else 1   # risposta del piccolo spedita all'utente
        else:
            costo += COSTO["grande"]       # il grande risponde, e il grande è sempre giusto
    return costo, errori


c0, e0 = tutto_grande()
print(f"tutto sul grande      costo {c0:>4}  errori {e0}")
for soglia in (0.95, 0.80, 0.50):
    c, e = a_cascata(soglia)
    print(f"cascata soglia {soglia:.2f}  costo {c:>4}  errori {e}  risparmio {100 * (1 - c / c0):.0f}%")

# Output:
# tutto sul grande      costo  300  errori 0
# cascata soglia 0.95  costo  320  errori 0  risparmio -7%
# cascata soglia 0.80  costo   35  errori 2  risparmio 88%
# cascata soglia 0.50  costo   35  errori 2  risparmio 88%

Le tre righe raccontano tre cose diverse. Con una soglia troppo alta (0,95) il piccolo non convince mai e la cascata costa più che usare sempre il grande, perché si paga la prima chiamata inutile. Con 0,80 il costo scende dell'88%, ma restano due errori: sono i casi in cui il modello ha sbagliato con confidenza 0,85, e nessuna soglia sopra o sotto li prende senza buttare via il risparmio. Il valore dell'88% è l'aritmetica del copione, non un dato da riportare altrove.

La conclusione operativa è che il risparmio di una cascata non si stima, si misura sui tuoi compiti, e va letto insieme agli errori che introduce. Serve quindi un insieme di compiti con la risposta giusta nota e un controllo di qualità: ne parla l'articolo sulla valutazione degli agenti. Non usarla quando un errore costa molto più di una chiamata al modello grande, o quando non hai modo di accorgerti degli errori: in quel caso il risparmio è solo un costo spostato altrove. Le percentuali di una ripartizione «70-20-10» tra richieste semplici, medie e difficili non hanno una fonte generale: la tua distribuzione si ricava dal tuo traffico.

Meno contesto, stesso lavoro: finestra, riassunto, recupero

Se i token d'ingresso crescono con la cronologia, la leva più diretta è non rimandarla tutta. Le tecniche sono tre: tenere solo gli ultimi N scambi (finestra scorrevole), sostituire la parte vecchia con un riassunto, o recuperare dalla memoria solo i frammenti pertinenti alla domanda corrente. Questa terza strada è quella dell'articolo sui sistemi di memoria. Il listato confronta le prime due con la cronologia completa su 30 scambi da 120 token ciascuno (token e riassuntore sono finti: il riassunto ha sempre 150 token).

TURNI = 30
TOKEN_PER_TURNO = 120   # token (qui: parole) per scambio utente+risposta
SOGLIA = 1000
FINESTRA = 6
RIASSUNTO = 150         # lunghezza del riassunto prodotto dal riassuntore finto


def completo():
    return sum(TOKEN_PER_TURNO * t for t in range(TURNI))


def finestra():
    return sum(TOKEN_PER_TURNO * min(t, FINESTRA) for t in range(TURNI))


def con_riassunto():
    cronologia = 0       # token della cronologia grezza dopo l'ultimo riassunto
    riassunto = 0
    spesi = 0
    extra_riassuntore = 0
    for _ in range(TURNI):
        spesi += riassunto + cronologia
        cronologia += TOKEN_PER_TURNO
        if cronologia > SOGLIA:
            extra_riassuntore += riassunto + cronologia   # il riassuntore legge tutto
            riassunto, cronologia = RIASSUNTO, 0
    return spesi, extra_riassuntore


s, extra = con_riassunto()
print("token di ingresso sommati su 30 scambi")
print(f"  cronologia completa : {completo()}")
print(f"  finestra di {FINESTRA}        : {finestra()}")
print(f"  riassunto oltre {SOGLIA}  : {s} (+ {extra} letti dal riassuntore)")

# Output:
# token di ingresso sommati su 30 scambi
#   cronologia completa : 52200
#   finestra di 6        : 19080
#   riassunto oltre 1000  : 16470 (+ 3540 letti dal riassuntore)

La finestra riduce l'ingresso da 52.200 a 19.080 token, ma è cieca: tutto ciò che è uscito dalla finestra, comprese le istruzioni dell'utente date all'inizio, non esiste più. Il riassunto fa meglio (16.470) ma ha un costo nascosto, che il listato stampa a parte: il riassuntore deve leggere la cronologia (altri 3.540 token) e, se è un modello, il riassunto può omettere o alterare dettagli. Il compromesso è sempre quello tra spesa e fedeltà, e l'unico modo di dire quale perdita sia accettabile è provarlo sui tuoi compiti.

Lavori che possono aspettare: l'elaborazione in batch

Una parte del lavoro di un agente non ha un utente in attesa: classificare un archivio, arricchire un dataset, rigenerare descrizioni, valutare una suite di test. Per questi lavori alcuni fornitori offrono interfacce asincrone più economiche. La documentazione di OpenAI descrive la Batch API come invio di gruppi di richieste asincrone con costi inferiori del 50%, limiti di frequenza separati e un tempo di consegna di 24 ore; quella di Anthropic descrive i batch di messaggi con elaborazione entro 24 ore, molti batch completati prima, e risultati disponibili 29 giorni. Gli sconti cambiano: verifica la pagina prezzi del tuo fornitore, e controlla se si sommano a quello della cache.

• Adatto: lavori senza utente in attesa, con volumi alti e risultati che si leggono dopo ore (arricchimento di dati, classificazione, valutazioni notturne).
• Non adatto: conversazioni, strumenti interattivi, qualunque passo di un ciclo agentico in cui il passo successivo dipende dalla risposta.
• Da prevedere: richieste che scadono senza risultato, da ritentare, e la regola di quanto si aspetta prima di dichiarare il lavoro fallito.

Il criterio comune è la dipendenza dal risultato: un ciclo agentico ha bisogno della risposta per decidere il passo seguente, quindi il batch riguarda i lavori attorno all'agente, non il suo ciclo. Rivalutare con un modello mille risposte di una suite di test è il caso tipico in cui il batch conviene.

Quando un agente si ripaga, e quando il self-hosting ha senso

La domanda finale non è tecnica: l'agente costa meno di quel che sostituisce? Il calcolo è semplice e va fatto con tutte le voci, non solo con le chiamate al modello: costo mensile dell'agente (modello, hosting, manutenzione ammortizzata), lavoro risparmiato (ore per costo orario) e costo iniziale di sviluppo da recuperare. Il secondo listato lo applica a un caso inventato e confronta anche il costo fisso di un modello gestito in proprio con quello a consumo.

import math

# 1) ROI e pareggio di un agente (numeri INVENTATI per l'esempio)
ore_settimana = 10
costo_orario = 40
costo_agente_mese = 600          # API + hosting + manutenzione ammortizzata
sviluppo_iniziale = 12_000
risparmio_mese = ore_settimana * costo_orario * 52 / 12
netto = risparmio_mese - costo_agente_mese
print(f"risparmio lordo al mese: {risparmio_mese:.0f}")
print(f"ROI mensile: {100 * netto / costo_agente_mese:.0f}%")
print(f"pareggio dello sviluppo: {math.ceil(sviluppo_iniziale / netto)} mesi")

# 2) Self-hosting contro API: da quale volume conviene (prezzi INVENTATI)
prezzo_api_per_m = 6.0           # dollari per milione di token (media ingresso+uscita)
fisso_mese = 1500                # server + corrente + tempo di chi lo gestisce
volume_pareggio_m = fisso_mese / prezzo_api_per_m
print(f"pareggio: {volume_pareggio_m:.0f} milioni di token al mese, "
      f"{volume_pareggio_m / 30:.1f} al giorno")
for al_giorno in (1, 5, 10):
    api = al_giorno * 30 * prezzo_api_per_m
    print(f"  {al_giorno:>2} M/giorno: API {api:>6.0f}$  fisso {fisso_mese}$  -> {'self-hosting' if api > fisso_mese else 'API'}")

# Output:
# risparmio lordo al mese: 1733
# ROI mensile: 189%
# pareggio dello sviluppo: 11 mesi
# pareggio: 250 milioni di token al mese, 8.3 al giorno
#    1 M/giorno: API    180$  fisso 1500$  -> API
#    5 M/giorno: API    900$  fisso 1500$  -> API
#   10 M/giorno: API   1800$  fisso 1500$  -> self-hosting

Nell'esempio l'agente restituisce un ROI mensile del 189% ma il pareggio dello sviluppo arriva in 11 mesi: due modi di dire la stessa cosa che suonano molto diversi, quindi riportali entrambi. Per il self-hosting, il pareggio con un costo fisso di 1.500 dollari al mese e un prezzo a consumo di 6 dollari per milione è a 250 milioni di token al mese. La parte difficile non è la divisione ma l'elenco dei costi fissi: il noleggio o l'acquisto della macchina, la corrente, gli aggiornamenti e il tempo di chi se ne occupa. Un modello più piccolo gestito in proprio, inoltre, non è un sostituto neutro del modello grande: la qualità va misurata come per qualunque cascata.

Domande frequenti

Che cosa sai fare adesso

Sai spiegare perché la spesa di un agente segue la lunghezza del compito e calcolarla con i tuoi prezzi, riconoscere quando la cache aiuta e quando costa di più, mettere un tetto per compito che restituisce un risultato parziale invece di un errore, e valutare una cascata di modelli guardando insieme risparmio ed errori. Sai ridurre il contesto con finestra o riassunto sapendo che cosa si perde, riservare il batch ai lavori senza utente in attesa e riportare ROI e pareggio come due numeri distinti. Tutte le cifre dei listati sono d'esempio: sostituiscile con le tue misure.