Le funzionalità di IA devono spesso soddisfare due esigenze contemporaneamente: gestire le richieste in modo responsabile ed evitare sprechi di risorse di calcolo. Questi obiettivi possono entrare in conflitto. Più controlli possono aumentare la latenza; limiti di input rigorosi possono rifiutare attività utili; inviare ogni richiesta a un modello di grandi dimensioni può essere costoso. Un piccolo gateway tra un’applicazione e il relativo modello può rendere queste decisioni esplicite e verificabili.
Questo tutorial crea un gateway Python che rifiuta le richieste troppo grandi, blocca una semplice categoria di dati sensibili, oscura gli indirizzi email, applica un limite di token per richiesta e scrive eventi di audit strutturati. Usa un modello simulato, così l’esempio funziona senza credenziali, accesso alla rete o SDK di un fornitore. I controlli sono misure ingegneristiche, non una dichiarazione di conformità legale: i requisiti effettivi dipendono dal sistema, dai dati, dalla giurisdizione e dall’organizzazione.
Il problema: i controlli devono far parte del percorso della richiesta
Le policy descritte nella documentazione non limitano automaticamente le chiamate ai modelli. Se ogni team di prodotto implementa i propri controlli, la stessa richiesta può essere gestita in modo diverso nelle varie parti di un prodotto. Un gateway centralizza alcune decisioni operative: quali input accettare, quali trasformazioni applicare, quanto lavoro consentire e quali informazioni registrare.
In questo esempio non esiste un contatore universale dei token. La tokenizzazione dipende dal modello e dal relativo tokenizer, quindi contare le parole separate da spazi è solo una stima approssimativa. Il gateway usa questa stima per illustrare il controllo del budget; un’integrazione di produzione dovrebbe usare il tokenizer del provider o i dati di utilizzo, quando opportuno. Allo stesso modo, un’espressione regolare per gli indirizzi email non è un rilevatore completo di informazioni personali o riservate.
Configurazione
Usa Python 3 e crea una directory per l’esempio. L’implementazione usa solo la libreria standard. Salva il programma principale riportato di seguito come gateway.py. In un servizio reale, conserva i valori delle policy in una configurazione sottoposta a revisione, evita di inserire segreti nel repository e decidi quali campi possono essere conservati nei log.
from dataclasses import dataclass
@dataclass(frozen=True)
class Policy:
max_input_chars: int = 4000
max_estimated_tokens: int = 900
max_output_tokens: int = 300
def estimate_tokens(text: str) -> int:
"""A deliberately rough estimate for this dependency-free demo."""
return len(text.split())
if __name__ == "__main__":
policy = Policy()
prompt = "Summarise the deployment checklist for the team."
estimate = estimate_tokens(prompt)
print(f"Input estimate: {estimate}")
print(f"Within budget: {estimate <= policy.max_estimated_tokens}")Eseguilo con python gateway.py. L’esempio definisce limiti come impostazioni immutabili, riducendo il rischio che il codice che gestisce le richieste modifichi accidentalmente la policy durante una chiamata. L’output è un primo controllo utile, ma il programma non rifiuta ancora le richieste né invoca un modello.
Passaggio 1: definire cosa accetta il gateway
La convalida dovrebbe avvenire prima di qualsiasi operazione costosa. Per prima cosa, rifiuta gli input vuoti o troppo lunghi. Poi applica eventuali regole specifiche del dominio. L’esempio seguente blocca il marcatore volutamente limitato [CONFIDENTIAL] per mostrare il rifiuto in base a una policy. Da solo non identifica i materiali riservati: un sistema reale richiede controlli adeguati ai propri dati e al proprio modello di minaccia.
Aggiungi al programma le funzioni seguenti. L’eccezione permette di distinguere una richiesta rifiutata da un errore del modello o di programmazione, così il chiamante può gestirla in modo intenzionale.
class PolicyError(ValueError):
pass
def validate_input(text: str, max_chars: int) -> None:
if not isinstance(text, str) or not text.strip():
raise PolicyError("Input must be a non-empty string")
if len(text) > max_chars:
raise PolicyError("Input exceeds the character limit")
if "[CONFIDENTIAL]" in text.upper():
raise PolicyError("Input contains a blocked confidentiality marker")
if __name__ == "__main__":
try:
validate_input("Summarise a public release note.", 4000)
print("Accepted")
except PolicyError as exc:
print(f"Rejected: {exc}")Questo controllo autonomo può essere eseguito così com’è. Nel programma finale, la stessa convalida viene eseguita all’interno del gateway, subito prima di qualsiasi chiamata al modello. Non affidarti all’interfaccia utente per applicare i limiti: le richieste possono arrivare da altri client, processi in background o futuri percorsi dell’applicazione.
Passaggio 2: trasformare i dati e controllare il budget
Il blocco non è l’unica opzione. Alcune applicazioni possono ridurre l’esposizione sostituendo valori specifici prima di inviare il testo a un modello. La funzione successiva oscura le stringhe che sembrano indirizzi email. È una dimostrazione circoscritta, non un sistema generico di prevenzione della perdita di dati. Evita di inserire il prompt originale nei record di audit, a meno che non vi sia una necessità documentata e una policy di conservazione approvata.
Il controllo del budget considera la stima dell’input e il limite configurato per l’output. Un provider reale potrebbe conteggiare anche le istruzioni di sistema, gli schemi degli strumenti, la cronologia della conversazione e altri campi come input. Quando integri un’API reale, includili nella stima.
import re
EMAIL_PATTERN = re.compile(r"\b[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}\b")
def redact_emails(text: str) -> str:
return EMAIL_PATTERN.sub("[REDACTED_EMAIL]", text)
def check_budget(text: str, max_total_tokens: int, output_tokens: int) -> int:
estimate = len(text.split())
if estimate + output_tokens > max_total_tokens:
raise ValueError("Estimated request exceeds token budget")
return estimate
if __name__ == "__main__":
prompt = "Send the summary to [email protected] after review."
safe_prompt = redact_emails(prompt)
print(safe_prompt)
print("Estimated input tokens:", check_budget(safe_prompt, 100, 20))Esegui questo blocco autonomamente per vedere la trasformazione. L’oscuramento può alterare il significato e le espressioni regolari possono non rilevare indirizzi insoliti o identificatori diversi dagli indirizzi email. In produzione, definisci quali categorie di dati possono essere inviate, quali devono essere rimosse e cosa fare quando il rilevatore non è sicuro. Non considerare una corrispondenza riuscita con un’espressione regolare come prova che il testo rimanente sia sicuro.
Passaggio 3: collegare i controlli a un adattatore del modello
Il gateway seguente riunisce tutti i componenti. MockModel funge da client del modello e restituisce una risposta prevedibile. Sostituiscilo con un adattatore per il provider scelto solo dopo aver deciso come gestire autenticazione, timeout, tentativi, trattamento dei dati e registrazione lato provider.
Il gateway scrive un oggetto JSON per ogni richiesta in un file di audit JSON Lines. Registra un identificatore della richiesta, l’esito, la dimensione stimata dell’input e il nome del modello, ma non il prompt né la risposta. In alcuni ambienti anche i metadati possono essere sensibili, quindi scegli con attenzione i campi e i tempi di conservazione. Il codice registra sia le richieste accettate sia quelle rifiutate, evitando però dettagli delle eccezioni che potrebbero contenere input dell’utente.
import json
import re
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Protocol
@dataclass(frozen=True)
class Policy:
max_input_chars: int = 4000
max_total_tokens: int = 1200
max_output_tokens: int = 300
class PolicyError(ValueError):
pass
class Model(Protocol):
name: str
def generate(self, prompt: str, max_output_tokens: int) -> str:
...
class MockModel:
name = "mock-model"
def generate(self, prompt: str, max_output_tokens: int) -> str:
words = prompt.split()
return "Processed request with " + str(len(words)) + " input words."
EMAIL_PATTERN = re.compile(r"\b[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}\b")
def estimate_tokens(text: str) -> int:
return len(text.split())
def redact_emails(text: str) -> str:
return EMAIL_PATTERN.sub("[REDACTED_EMAIL]", text)
def validate_input(text: str, policy: Policy) -> None:
if not isinstance(text, str) or not text.strip():
raise PolicyError("Input must be a non-empty string")
if len(text) > policy.max_input_chars:
raise PolicyError("Input exceeds the character limit")
if "[CONFIDENTIAL]" in text.upper():
raise PolicyError("Input contains a blocked confidentiality marker")
def write_audit(path: Path, event: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as stream:
stream.write(json.dumps(event, sort_keys=True) + "\n")
class AIGateway:
def __init__(self, model: Model, policy: Policy, audit_path: Path):
self.model = model
self.policy = policy
self.audit_path = audit_path
def handle(self, text: str) -> str:
request_id = str(uuid.uuid4())
estimated = 0
outcome = "rejected"
try:
validate_input(text, self.policy)
safe_text = redact_emails(text)
estimated = estimate_tokens(safe_text)
if estimated + self.policy.max_output_tokens > self.policy.max_total_tokens:
raise PolicyError("Estimated request exceeds token budget")
result = self.model.generate(safe_text, self.policy.max_output_tokens)
outcome = "accepted"
return result
finally:
event = {
"request_id": request_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"outcome": outcome,
"estimated_input_tokens": estimated,
"model": self.model.name,
}
write_audit(self.audit_path, event)
if __name__ == "__main__":
gateway = AIGateway(
model=MockModel(),
policy=Policy(),
audit_path=Path("logs/ai-gateway.jsonl"),
)
try:
answer = gateway.handle("Summarise this release note for [email protected].")
print(answer)
except PolicyError as exc:
print(f"Request rejected: {exc}")Salva il blocco completo come gateway.py ed esegui python gateway.py. Il programma crea logs/ai-gateway.jsonl nella directory corrente. Esamina il file per verificare che contenga metadati e non il prompt. In un servizio distribuito, usa una pipeline di logging adeguata, con controllo degli accessi, limiti di conservazione e protezione dai problemi dovuti alle scritture concorrenti.
Passaggio 4: verificare i percorsi delle policy
I test dovrebbero coprire sia le richieste accettate sia quelle rifiutate. Questo modulo di test della libreria standard importa il gateway e verifica che l’adattatore simulato venga eseguito per un input valido, mentre un marcatore bloccato genera un errore di policy. Verifica inoltre che una richiesta rifiutata venga registrata nell’audit. Salvalo come test_gateway.py accanto a gateway.py, quindi esegui python -m unittest -v.
import json
import tempfile
import unittest
from pathlib import Path
from gateway import AIGateway, MockModel, Policy, PolicyError
class GatewayTests(unittest.TestCase):
def setUp(self):
self.temp_dir = tempfile.TemporaryDirectory()
self.audit_path = Path(self.temp_dir.name) / "audit.jsonl"
self.gateway = AIGateway(MockModel(), Policy(), self.audit_path)
def tearDown(self):
self.temp_dir.cleanup()
def test_accepts_valid_request_and_writes_event(self):
result = self.gateway.handle("Summarise the public update.")
self.assertIn("Processed request", result)
event = json.loads(self.audit_path.read_text(encoding="utf-8"))
self.assertEqual(event["outcome"], "accepted")
self.assertNotIn("prompt", event)
def test_rejects_confidential_marker_and_audits(self):
with self.assertRaises(PolicyError):
self.gateway.handle("Review [CONFIDENTIAL] draft")
event = json.loads(self.audit_path.read_text(encoding="utf-8"))
self.assertEqual(event["outcome"], "rejected")
if __name__ == "__main__":
unittest.main()Il test dell’audit controlla intenzionalmente solo la struttura di base dell’evento. Aggiungi test per input vuoto, input troppo lungo, rifiuto per superamento del budget di token e oscuramento. Se il sistema usa strumenti o recupero di informazioni, aggiungi test anche per questi percorsi: una policy applicata solo al prompt iniziale potrebbe non governare il testo introdotto in seguito.
Errori comuni
- Definire una stima come conteggio esatto dei token. Il conteggio delle parole è una semplificazione didattica. Quando disponibile, usa il conteggio specifico del modello e lascia margine per le istruzioni di sistema e il contesto della conversazione.
- Supporre che l’oscuramento renda sicuri i dati. L’esempio gestisce solo stringhe che sembrano indirizzi email. Nomi, identificativi di account, segreti e combinazioni di dettagli potrebbero comunque consentire l’identificazione.
- Registrare troppe informazioni. I prompt e le risposte possono contenere informazioni sensibili. Definisci lo scopo di ogni campo di audit, limita l’accesso e stabilisci i tempi di conservazione prima di attivare la registrazione in produzione.
- Confondere un errore di policy con un’interruzione del servizio. Una richiesta rifiutata è una decisione intenzionale; un timeout o un errore del provider è un problema operativo. In un adattatore reale, gestiscili e monitorali separatamente.
- Aggiungere tentativi senza limiti. I tentativi possono aumentare la latenza e consumare più risorse. Limitali, usa timeout e valuta se ripetere una richiesta sia sicuro per l’operazione in questione.
- Trattare un modello simulato come un’integrazione di produzione. Il modello simulato non usa rete, autenticazione, trattamento dei dati lato provider né conteggio reale dei token. Questi aspetti richiedono un’implementazione e una revisione esplicite.
Conclusione
Un gateway offre a un’applicazione un unico punto in cui applicare limiti alle richieste, eseguire trasformazioni, controllare un budget di calcolo e registrare un evento di audit minimo. L’esempio è volutamente semplice: i suoi rilevatori e la stima dei token non sono sufficienti per un utilizzo che richieda garanzie elevate. Mantieni visibili i controlli, verifica i percorsi di rifiuto e adattali al modello, ai flussi di dati e ai requisiti operativi effettivi prima di collegare un servizio di produzione.
