Introduzione: Dall'Esperimento alla Produzione
Usare un LLM nel playground e semplice. Portarlo in produzione e una sfida ingegneristica completamente diversa. In produzione devi gestire rate limiting, retry su errori, caching per ridurre i costi, monitoring per tracciare latenza e qualità, fallback quando un provider e down, e budget mensili che possono esplodere senza controllo.
Questo articolo copre l'intero percorso: dalle API dei principali provider (OpenAI, Anthropic) al deploy di modelli open source, con pattern architetturali collaudati per applicazioni LLM robuste e scalabili.
Cosa Imparerai in Questo Articolo
- Le API di OpenAI e Anthropic: setup, modelli e pricing
- Deploy di modelli open source con Ollama e vLLM
- Pattern di produzione: retry, caching, rate limiting
- Streaming per una UX reattiva
- Strategie di fallback e multi-provider
- Monitoring, logging e gestione dei costi
API OpenAI: Il Leader di Mercato
OpenAI offre l'ecosistema API più maturo e diffuso. I modelli GPT-4 e GPT-4o rappresentano lo standard de facto per molte applicazioni, con un'ampia documentazione e una community attiva.
# Setup completo dell'API OpenAI con gestione errori
from openai import OpenAI, APIError, RateLimitError, APITimeoutError
import time
client = OpenAI(
api_key="sk-...", # Meglio da variabile d'ambiente
timeout=30.0, # Timeout in secondi
max_retries=3 # Retry automatici
)
def call_openai_with_retry(
messages: list,
model: str = "gpt-4o",
max_retries: int = 3,
base_delay: float = 1.0
) -> str:
"""Chiama OpenAI con exponential backoff su errori."""
for attempt in range(max_retries):
try:
response = client.chat.completions.create(
model=model,
messages=messages,
temperature=0.7,
max_tokens=1000
)
return response.choices[0].message.content
except RateLimitError:
delay = base_delay * (2 ** attempt)
print(f"Rate limited. Retry in {delay}s...")
time.sleep(delay)
except APITimeoutError:
print(f"Timeout. Attempt {attempt + 1}/{max_retries}")
time.sleep(base_delay)
except APIError as e:
print(f"API error: {e.status_code} - {e.message}")
if e.status_code >= 500:
time.sleep(base_delay * (2 ** attempt))
else:
raise
raise Exception("Max retries exceeded")
# Utilizzo
result = call_openai_with_retry(
messages=[{"role": "user", "content": "Spiega il pattern Repository"}]
)
API Anthropic: Sicurezza e Affidabilità
Anthropic offre la famiglia di modelli Claude, con un focus su sicurezza, riduzione delle allucinazioni e lunghe context window (fino a 200K token). L'API e simile nella struttura ma con alcune differenze chiave.
# Setup API Anthropic con streaming
from anthropic import Anthropic
client = Anthropic(api_key="sk-ant-...")
# Chiamata base
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1000,
system="Sei un esperto di architettura software. Rispondi in modo conciso.",
messages=[
{"role": "user", "content": "Confronta monolite vs microservizi"}
]
)
print(response.content[0].text)
print(f"Token usati: {response.usage.input_tokens} in + {response.usage.output_tokens} out")
# Streaming per UX reattiva
def stream_claude_response(prompt: str) -> str:
"""Stream la risposta token per token per una UX reattiva."""
full_response = ""
with client.messages.stream(
model="claude-3-5-sonnet-20241022",
max_tokens=1000,
messages=[{"role": "user", "content": prompt}]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
full_response += text
print() # newline finale
return full_response
response = stream_claude_response("Scrivi una guida rapida al testing in Python")
Modelli Open Source: Liberta e Controllo
I modelli open source come Llama 3 e Mistral offrono il controllo totale sui dati e sull'infrastruttura. Nessun dato lascia il tuo ambiente, nessun costo per token, ma serve gestire l'infrastruttura GPU.
Ollama: Il Modo Più Semplice
Ollama e il modo più rapido per eseguire modelli open source localmente. Un singolo comando scarica e avvia il modello, esponendo un'API compatibile con OpenAI.
# Utilizzo di Ollama con API compatibile OpenAI
from openai import OpenAI
# Ollama espone un'API compatibile OpenAI su localhost
ollama_client = OpenAI(
base_url="http://localhost:11434/v1",
api_key="ollama" # Ollama non richiede API key
)
# Usa esattamente la stessa interfaccia di OpenAI!
response = ollama_client.chat.completions.create(
model="llama3.1:8b", # Modello locale
messages=[
{"role": "system", "content": "Sei un assistente tecnico."},
{"role": "user", "content": "Spiega Docker in 3 punti"}
],
temperature=0.7,
max_tokens=500
)
print(response.choices[0].message.content)
# Costo: $0 (solo elettricita e hardware)
Pattern di Produzione: Caching
Il caching e la strategia più efficace per ridurre i costi in produzione. Se la stessa domanda (o una simile) viene posta ripetutamente, non serve chiamare l'LLM ogni volta.
# Sistema di caching per risposte LLM
import hashlib
import json
from datetime import datetime, timedelta
class LLMCache:
"""Cache semplice per risposte LLM con TTL."""
def __init__(self, ttl_hours: int = 24):
self.cache: dict = {}
self.ttl = timedelta(hours=ttl_hours)
self.hits = 0
self.misses = 0
def _make_key(self, model: str, messages: list, temperature: float) -> str:
"""Genera una chiave di cache deterministica."""
content = json.dumps({
"model": model,
"messages": messages,
"temperature": temperature
}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()
def get(self, model: str, messages: list, temperature: float) -> str | None:
"""Cerca nella cache. Ritorna None se miss."""
key = self._make_key(model, messages, temperature)
if key in self.cache:
entry = self.cache[key]
if datetime.now() - entry["timestamp"] < self.ttl:
self.hits += 1
return entry["response"]
del self.cache[key]
self.misses += 1
return None
def set(self, model: str, messages: list, temperature: float, response: str):
"""Salva nella cache."""
key = self._make_key(model, messages, temperature)
self.cache[key] = {
"response": response,
"timestamp": datetime.now()
}
def stats(self) -> dict:
total = self.hits + self.misses
return {
"hits": self.hits,
"misses": self.misses,
"hit_rate": f"{self.hits/total*100:.1f}%" if total > 0 else "N/A",
"cached_entries": len(self.cache)
}
# Utilizzo
cache = LLMCache(ttl_hours=24)
def cached_llm_call(messages: list, model: str = "gpt-4o") -> str:
cached = cache.get(model, messages, 0.7)
if cached:
return cached
response = call_openai_with_retry(messages, model)
cache.set(model, messages, 0.7, response)
return response
Fallback Multi-Provider
In produzione, affidarsi a un singolo provider e rischioso. Un sistema di fallback multi-provider garantisce la disponibilità anche quando un provider ha problemi.
# Router multi-provider con fallback automatico
from anthropic import Anthropic
from openai import OpenAI
class LLMRouter:
"""Router che prova provider multipli in ordine di priorità."""
def __init__(self):
self.providers = [
{"name": "anthropic", "client": Anthropic(), "model": "claude-3-5-sonnet-20241022"},
{"name": "openai", "client": OpenAI(), "model": "gpt-4o"},
{"name": "ollama", "client": OpenAI(base_url="http://localhost:11434/v1", api_key="ollama"), "model": "llama3.1:8b"},
]
def call(self, messages: list, max_tokens: int = 1000) -> dict:
"""Prova ogni provider in ordine. Ritorna al primo successo."""
errors = []
for provider in self.providers:
try:
if provider["name"] == "anthropic":
response = provider["client"].messages.create(
model=provider["model"],
max_tokens=max_tokens,
messages=messages
)
return {
"content": response.content[0].text,
"provider": provider["name"],
"model": provider["model"]
}
else:
response = provider["client"].chat.completions.create(
model=provider["model"],
messages=messages,
max_tokens=max_tokens
)
return {
"content": response.choices[0].message.content,
"provider": provider["name"],
"model": provider["model"]
}
except Exception as e:
errors.append(f"{provider['name']}: {str(e)}")
continue
raise Exception(f"Tutti i provider falliti: {errors}")
# Utilizzo
router = LLMRouter()
result = router.call([{"role": "user", "content": "Ciao!"}])
print(f"Risposta da {result['provider']}: {result['content']}")
Monitoring e Gestione dei Costi
Senza monitoring, i costi delle API LLM possono sfuggire di mano rapidamente. Un sistema di tracking e essenziale per mantenere il controllo.
Confronto Costi per Provider (per 1M token)
| Modello | Input | Output | Note |
|---|---|---|---|
| GPT-4o | $2.50 | $10.00 | Best all-round |
| GPT-4o-mini | $0.15 | $0.60 | Ottimo rapporto qualità/prezzo |
| Claude 3.5 Sonnet | $3.00 | $15.00 | 200K context, sicurezza |
| Claude 3.5 Haiku | $0.25 | $1.25 | Veloce ed economico |
| Llama 3.1 8B (Ollama) | $0 | $0 | Costo hardware fisso |
# Sistema di monitoring costi
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class UsageTracker:
"""Traccia utilizzo e costi delle API LLM."""
daily_budget_usd: float = 50.0
records: list = field(default_factory=list)
# Prezzi per 1M token (input, output)
PRICING = {
"gpt-4o": (2.50, 10.00),
"gpt-4o-mini": (0.15, 0.60),
"claude-3-5-sonnet-20241022": (3.00, 15.00),
"claude-3-5-haiku-20241022": (0.25, 1.25),
}
def log_usage(self, model: str, input_tokens: int, output_tokens: int):
pricing = self.PRICING.get(model, (0, 0))
cost = (input_tokens * pricing[0] + output_tokens * pricing[1]) / 1_000_000
self.records.append({
"timestamp": datetime.now(),
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost_usd": cost
})
# Alert se vicini al budget
daily_total = self.get_daily_cost()
if daily_total > self.daily_budget_usd * 0.8:
print(f"ALERT: {daily_total:.2f}/{self.daily_budget_usd} USD budget giornaliero!")
def get_daily_cost(self) -> float:
today = datetime.now().date()
return sum(
r["cost_usd"] for r in self.records
if r["timestamp"].date() == today
)
def report(self) -> dict:
return {
"total_requests": len(self.records),
"total_cost": f"