Se lavori con l’API di Claude e hai mai dovuto elaborare migliaia di documenti, classificare grandi dataset o generare contenuti in batch, conosci bene il problema: i costi si accumulano rapidamente e le richieste sincrone rallentano il tuo sistema. La Batch API di Claude risolve entrambi i problemi in un colpo solo, offrendo elaborazione asincrona a meta’ del prezzo rispetto alle chiamate standard.
In questa guida vedremo come funziona la Batch API, quando conviene usarla, e come integrarla in un workflow Python reale. Se hai gia’ lavorato con la Prompt Caching di Claude per ridurre i costi, la Batch API e’ il passo successivo per ottimizzare le elaborazioni su larga scala.
Cos’e’ la Batch API di Claude
La Batch API (endpoint POST /v1/messages/batches) e’ un sistema di elaborazione asincrona che permette di inviare fino a 10.000 richieste in un singolo batch. Invece di aspettare ogni risposta in tempo reale, invii tutte le richieste insieme e recuperi i risultati quando il processing e’ completato, tipicamente entro un’ora ma con un massimo di 24 ore.
Il vantaggio economico e’ immediato: 50% di sconto su input e output tokens rispetto alle chiamate Messages API standard. Su carichi di lavoro grandi, questa differenza si traduce in risparmi significativi ogni mese.
Quando usare la Batch API
La Batch API e’ ideale quando la latenza non e’ un vincolo critico. Casi d’uso perfetti:
- Classificazione di testi — analizzare migliaia di recensioni, ticket di supporto o email
- Generazione di contenuti — creare descrizioni prodotto, summary o traduzioni in bulk
- Analisi dati non strutturati — estrarre entita’, sentiment o metadati da documenti
- Valutazione di modelli — eseguire benchmark e test su dataset grandi
- Pipeline ETL con AI — arricchire dati esistenti con insight generati da LLM
Non e’ invece adatta per applicazioni interattive, chatbot o qualsiasi caso in cui l’utente aspetta una risposta in tempo reale. Per quei casi, l’API standard con streaming rimane la scelta giusta.
Come funziona il ciclo di vita di un batch
Il workflow della Batch API si articola in quattro fasi principali:
- Creazione — invii un array di richieste con
POST /v1/messages/batches - Processing — Anthropic elabora il batch in modo asincrono (stato:
in_progress) - Completamento — il batch raggiunge lo stato
endedcon tutti i risultati disponibili - Recupero — scarichi i risultati con
GET /v1/messages/batches/{id}/results
Ogni richiesta nel batch include un custom_id che ti permette di associare ogni risultato alla richiesta originale. I risultati vengono restituiti in formato JSONL (una riga per ogni risultato), indipendentemente dall’ordine di completamento.
Implementazione Python: da zero a produzione
Vediamo come implementare la Batch API in Python. Useremo la libreria ufficiale anthropic che semplifica notevolmente la gestione del ciclo di vita del batch.
Setup e installazione
pip install anthropic
export ANTHROPIC_API_KEY="sk-ant-..."Creare e inviare un batch
Il codice seguente mostra come creare un batch per classificare 1.000 recensioni di prodotti. Ogni richiesta e’ indipendente e include il testo da analizzare:
import anthropic
import json
client = anthropic.Anthropic()
# Prepara le richieste — in produzione le carichi da DB o CSV
reviews = [
{"id": "rev_001", "text": "Prodotto eccellente, consegna rapida!"},
{"id": "rev_002", "text": "Qualita’ deludente, non corrisponde alle foto."},
{"id": "rev_003", "text": "Nella media, niente di speciale."},
# ... fino a 10.000 elementi
]
# Costruisci le richieste batch
requests_list = []
for review in reviews:
requests_list.append(
anthropic.types.message_create_params.Request(
custom_id=review["id"],
params=anthropic.types.MessageCreateParamsNonStreaming(
model="claude-opus-4-5",
max_tokens=100,
messages=[{
"role": "user",
"content": f"""Classifica questa recensione in una categoria:
POSITIVA, NEGATIVA o NEUTRALE.
Rispondi con una sola parola.
Recensione: {review["text"]}"""
}]
)
)
)
# Crea il batch
batch = client.messages.batches.create(requests=requests_list)
print(f"Batch creato: {batch.id}")
print(f"Stato: {batch.processing_status}")
print(f"Richieste totali: {batch.request_counts.processing}")Monitorare lo stato del batch
Dopo aver creato il batch, devi attendere il completamento. La strategia migliore e’ il polling con backoff esponenziale per non sovraccaricare l’API con richieste continue:
import time
def wait_for_batch(client: anthropic.Anthropic, batch_id: str,
poll_interval: int = 60) -> anthropic.types.MessageBatch:
"""Attende il completamento del batch con polling intelligente."""
while True:
batch = client.messages.batches.retrieve(batch_id)
status = batch.processing_status
counts = batch.request_counts
print(f"[{batch_id}] Stato: {status} | "
f"Completate: {counts.succeeded}/{counts.processing + counts.succeeded + counts.errored}")
if status == "ended":
print(f"Batch completato!")
print(f" Successi: {counts.succeeded}")
print(f" Errori: {counts.errored}")
print(f" Scadute: {counts.expired}")
return batch
# Backoff: aumenta progressivamente l’intervallo
time.sleep(poll_interval)
poll_interval = min(poll_interval * 1.5, 300) # max 5 minuti
# Uso
batch_id = "msgbatch_01HkcTjaV5uDC8jWR3IT96xy"
completed_batch = wait_for_batch(client, batch_id)Recuperare e processare i risultati
Una volta completato il batch, recuperi i risultati in streaming JSONL. La libreria Python gestisce automaticamente il parsing:
def process_batch_results(client: anthropic.Anthropic, batch_id: str) -> dict:
"""Recupera e organizza i risultati del batch."""
results = {}
errors = []
for result in client.messages.batches.results(batch_id):
custom_id = result.custom_id
if result.result.type == "succeeded":
# Estrai il testo dalla risposta
message = result.result.message
text_content = next(
(block.text for block in message.content
if hasattr(block, "text")),
""
)
results[custom_id] = {
"status": "ok",
"response": text_content.strip(),
"input_tokens": message.usage.input_tokens,
"output_tokens": message.usage.output_tokens,
}
elif result.result.type == "errored":
error = result.result.error
errors.append({
"id": custom_id,
"type": error.type,
"message": str(error)
})
results[custom_id] = {"status": "error"}
elif result.result.type == "expired":
results[custom_id] = {"status": "expired"}
print(f"Risultati elaborati: {len(results)}")
print(f"Errori: {len(errors)}")
# Salva su file per analisi successiva
with open("batch_results.json", "w", encoding="utf-8") as f:
json.dump(results, f, ensure_ascii=False, indent=2)
return results
# Pipeline completa
results = process_batch_results(client, batch_id)
# Analisi dei risultati
categories = {"POSITIVA": 0, "NEGATIVA": 0, "NEUTRALE": 0}
for rid, data in results.items():
if data["status"] == "ok":
cat = data["response"].upper()
if cat in categories:
categories[cat] += 1
print(f"
Distribuzione sentiment:")
for cat, count in categories.items():
print(f" {cat}: {count}")Gestione avanzata: cancellazione e scadenza
I batch hanno una scadenza di 24 ore. Se il batch non viene completato entro questo limite, le richieste non ancora elaborate vengono marcate come expired. Puoi anche cancellare manualmente un batch in corso:
# Cancella un batch in corso
cancelled = client.messages.batches.cancel(batch_id)
print(f"Stato dopo cancellazione: {cancelled.processing_status}")
# Output: canceling (poi diventa ended con le richieste elaborate)
# Lista tutti i batch recenti
batches = client.messages.batches.list(limit=10)
for b in batches.data:
print(f"{b.id}: {b.processing_status} | "
f"Creato: {b.created_at} | "
f"Successi: {b.request_counts.succeeded}")Ottimizzazione dei costi: calcolo del risparmio
Con la Batch API paghi esattamente meta’ del costo normale per token. Facciamo i conti su un caso reale:
Supponiamo di dover classificare 10.000 recensioni, ognuna con circa 200 token in input e 10 token in output, usando Claude Haiku 3.5:
- API standard: (10.000 x 200 x $0.80/Mtok) + (10.000 x 10 x $4/Mtok) = $1.60 + $0.40 = $2.00
- Batch API: ($1.60 + $0.40) x 0.50 = $1.00
Su 1 milione di recensioni il risparmio sale a $100. Ma il vantaggio non e’ solo economico: la Batch API non conta contro i rate limit standard, permettendoti di elaborare volumi enormi senza preoccuparti dei limiti per minuto.
Se usi gia’ tecniche come la RAG con Claude API per recuperare contesto rilevante, puoi combinare le due ottimizzazioni: recuperi i chunk via RAG, costruisci i prompt e li elabori in batch per massimizzare il risparmio.
Pattern architetturali per pipeline di produzione
In produzione, la Batch API si integra naturalmente in pipeline asincrone. Ecco i pattern piu’ efficaci:
Pattern 1: Worker asincrono con coda
Il pattern piu’ comune prevede un worker che accumula richieste in una coda (Redis, SQS, etc.), le raggruppa in batch e le invia all’API. I risultati vengono poi scritti nel database o inviati via webhook.
Best practice: raggruppa le richieste per dimensione simile dei prompt. Batch omogenei completano piu’ velocemente e sono piu’ facili da monitorare.
Pattern 2: Scheduled batch notturno
Per pipeline non urgenti (report giornalieri, analisi periodiche), un job schedulato notturno raccoglie tutti i dati del giorno, crea un batch e processa i risultati al mattino. Massimo risparmio, zero impatto sulle operazioni diurne.
Pattern 3: Retry intelligente per errori
Le richieste fallite (errored) vanno gestite con retry esponenziale. Crea un nuovo batch con sole le richieste fallite, con un contatore di tentativi per evitare loop infiniti su errori permanenti.
Limiti e considerazioni pratiche
Prima di adottare la Batch API in produzione, tieni a mente questi vincoli:
- Massimo 10.000 richieste per batch — per dataset piu’ grandi, dividi in piu’ batch
- Payload massimo: 32 MB — prompt molto lunghi potrebbero richiedere batch piu’ piccoli
- Scadenza 24 ore — implementa sempre la gestione dei risultati
expired - No streaming — i risultati sono disponibili solo a batch completato
- Nessuna prioritizzazione — non puoi marcare alcune richieste come urgenti
- Tool use supportato — puoi usare tool use e agenti nelle richieste batch
Integrazione con workflow esistenti
Se stai costruendo agenti AI complessi, la Batch API si combina bene con architetture multi-agente. Puoi usarla per le fasi di analisi o preprocessing che non richiedono risposta immediata, mentre il loop principale dell’agente gestisce le interazioni in tempo reale.
Un pattern interessante e’ il dual-track processing: le richieste urgenti vanno sull’API standard con streaming, mentre le richieste non urgenti vengono accumulate e processate in batch. Il sistema ottimizza automaticamente il costo totale mantenendo la reattivita’ dove serve.
Conclusioni
La Batch API di Claude e’ uno strumento potente per chiunque lavori con elaborazioni massive. Il risparmio del 50% non e’ marketing: e’ reale e significativo su volumi alti. La chiave e’ identificare nel tuo workflow le parti che non richiedono risposta sincrona e migrarle su batch.
Inizia con un caso d’uso semplice — la classificazione di testi e’ perfetta — implementa il polling con backoff, e misura il risparmio reale dopo il primo mese. Nella maggior parte dei casi, il ROI giustifica l’investimento nell’integrazione gia’ dalla prima settimana di utilizzo.
Per approfondire le ottimizzazioni dell’API, leggi anche la guida sulla Prompt Caching: combinata con la Batch API, permette di ridurre ulteriormente i costi sui prompt con contesto ricorrente.
