Inference-per-dollar: padroneggiare i costi degli agenti AI con caching e interruttori di circuito

Il caching del contesto, che io chiamo la "regola del 90%" per il suo potenziale di significativi risparmi sui costi, è una tecnica in cui i provider memorizzano nella cache una parte sostanziale di un prompt grande e frequentemente ripetuto. Questo si riferisce tipicamente al "contesto" o alle "istruzioni di sistema" che rimangono statiche attraverso più turni di interazione di un agente. Quando le chiamate successive vengono effettuate con lo stesso contesto memorizzato nella cache e solo il nuovo input dell'utente, il provider può recuperare il contesto pre-elaborato. Ciò riduce drasticamente il consumo di token e la latenza per l'input di grandi dimensioni. Ho trovato questo particolarmente efficace per gli agenti che mantengono una persona di sistema coerente o una vasta base di conoscenza recuperata su più turni.

Per qualificarsi efficacemente per il caching e gli sconti associati (che possono essere sostanziali, spesso circa il 90% di riduzione per la parte di input in cache), i provider hanno tipicamente diversi requisiti. Presto molta attenzione a questi quando progetto i miei prompt:

• TTL minimo (time to live): le politiche di conservazione della cache variano significativamente in base al provider. Il caching esplicito (come quello di Vertex AI di Google) di solito addebita un costo per ora di archiviazione, consentendo di impostare TTL lunghi per basi di conoscenza stabili. Al contrario, il caching automatico (come quello di OpenAI o Anthropic) si basa su TTL effimeri di 5-10 minuti che si aggiornano all'uso. È necessario progettare la cadenza di interazione del proprio agente in modo che corrisponda a questi meccanismi TTL per massimizzare gli hit della cache.
• Conteggio minimo di token: il blocco memorizzato nella cache deve raggiungere una soglia specifica del provider per attivare il motore di caching. Questo varia da un minimo di 1.024 a 2.048 token per il caching automatico fino a 32.768 token per la creazione esplicita della cache. Questa soglia garantisce che l'overhead computazionale del caching sia giustificato dai risparmi derivanti dal riutilizzo di un contesto di grandi dimensioni.
• Corrispondenza esatta: questo è il requisito più cruciale e spesso trascurato. La parte memorizzata nella cache del prompt deve essere esattamente la stessa, bit per bit, di quella precedente. Anche un singolo spazio extra, un carattere di nuova riga o un riordino minore può rompere la cache e forzare una rielaborazione completa dell'intero prompt, costandovi il prezzo intero. Sono stato morso da questo in passato, quindi sono meticoloso sulla coerenza del prompt.

Quando si effettua la distribuzione in regioni europee, di solito scelgo europe-west1 o europe-west4 su gcp, eu-west-1 su aws, o westeurope o northeurope su azure. queste regioni offrono spesso un'eccellente latenza di rete all'interno dell'europa e sono frequentemente tra le prime a supportare nuove funzionalità di ai e distribuzioni di modelli. per le capacità specifiche della regione, consultare località google cloud, regioni aws per amazon bedrock e geografie e regioni azure.

La scelta del modello e del provider cloud giusti è un componente critico per l'ottimizzazione dei costi. La tariffazione per i modelli linguistici di grandi dimensioni (LLM) può variare significativamente, influenzando direttamente la metrica "Inference-per-Dollar". Di seguito sono riportate le mie considerazioni di benchmarking per i modelli leader, utilizzando un tasso di conversione approssimativo di 1 $ = 0,92 € per queste cifre:

• Gemini 2.5 Pro su GCP: 1,27 €/M (1,38 $/M) token di input.
• GPT-5.2 su Azure: 1,78 €/M (1,93 $/M) token di input.
• Opus 4.5 su AWS Bedrock: 2,30 €/M (2,50 $/M) token di input (illustrativo). Confermare sempre i prezzi di listino attuali nella propria regione su amazon bedrock pricing e id dei modelli nella documentazione di bedrock.

Questi dati evidenziano come anche piccole differenze di prezzo per milione di token possano portare a variazioni di costo sostanziali su larga scala. La mia scelta dipende dai compromessi specifici tra prestazioni, set di funzionalità e costi per il carico di lavoro del mio agente.

Un interruttore di circuito agentico è un meccanismo che implemento per impedire a un agente AI autonomo di entrare in cicli di ragionamento incontrollati che consumano token eccessivi e, di conseguenza, il budget. È una difesa pragmatica contro lo scenario della "carta di credito senza limiti". L'idea centrale è quella di imporre limiti di token rigidi e monitorare l'utilizzo cumulativo dei token all'interno di una sessione o di un'attività di un agente. Se l'agente si avvicina a una soglia predefinita, l'interruttore di circuito si attiva, terminando la linea di ragionamento attuale, passando a un modello più economico e più piccolo o scalando a una revisione umana. Non si tratta di soffocare l'intelligenza, ma di garantire un funzionamento responsabile e attento ai costi.

Ecco come visualizzo l'architettura per un agente AI che sfrutta il caching e un interruttore di circuito:

Governance e sicurezza del modello

Quando si implementano agenti AI, la sicurezza è fondamentale. Gli attacchi alla supply chain sugli strumenti e le dipendenze AI sono un rischio crescente per le pipeline CI/CD e gli ambienti di sviluppo. Implementare sempre le seguenti pratiche:

• Principio del minimo privilegio: assicurarsi che gli account di servizio dell'agente abbiano solo le autorizzazioni minime necessarie per accedere alle API LLM e ad altre risorse.
• Rotazione delle credenziali: ruotare regolarmente le chiavi API e le credenziali cloud. Implementare la rotazione automatica ove possibile.
• Isolamento dell'ambiente: distribuire gli agenti in ambienti isolati (ad esempio, spazi dei nomi Kubernetes, VM dedicate) per limitare il raggio d'azione dell'esplosione.
• Registrazione degli audit: abilitare la registrazione completa degli audit per tutte le chiamate API LLM e le azioni degli agenti. Integrare questi log con il SIEM per il rilevamento delle anomalie.
• Sicurezza della supply chain: verificare tutte le librerie e le dipendenze di terze parti. Utilizzare scanner come Trivy o Snyk in CI e seguire framework come NIST SSDF o SLSA dove appropriato.

Il conteggio locale dei token fornisce stime per i controlli preliminari e gli interruttori di circuito. Tuttavia, per la fatturazione, affidarsi sempre ai conteggi ufficiali dei token restituiti dal provider cloud nella risposta API. Questi sono i numeri che compaiono sulla fattura.

Azure e GCP forniscono funzionalità di piattaforma integrate per la limitazione della frequenza e le quote che sono il modo più robusto per imporre limiti rigidi.

• Azure: utilizzare la policy azure-openai-token-limit in API Management per applicare tariffe token per minuto o quote fisse.
• GCP: utilizzare la policy LLMTokenQuota di Apigee per gestire i costi e applicare limiti di consumo di token per periodo di tempo.

Il seguente codice python illustra un interruttore di circuito lato client, che è un modello flessibile che è possibile implementare direttamente nella logica dell'applicazione dell'agente.

Questo esempio dimostra un agente concettuale che interagisce con un modello GCP Vertex AI. Include un contatore di token lato client per l'interruttore di circuito e mostra come strutturare le chiamate per sfruttare la funzione di caching del contesto di Vertex AI.

# File: agent_sdk_runner.py
import os
from typing import Dict, Any, List

# --- SDK Imports --- #
# Use the official SDKs for production
import vertexai
from vertexai.generative_models import GenerativeModel, Part
from vertexai.preview.generative_models import caching

# A simple, local token counter for estimates.
# For production, consider a library like tiktoken, but for billing,
# ALWAYS use the usage_metadata from the API response.
def estimate_tokens(text: str) -> int:
    """Provides a rough estimate of token count. Not for billing."""
    return len(text) // 4

class AgentCircuitBreaker(Exception):
    """Custom exception for when the circuit breaker trips."""
    pass

class VertexAIAgent:
    def __init__(self, project_id: str, location: str, model_name: str, max_session_tokens: int):
        self.project_id = project_id
        self.location = location
        self.model_name = model_name
        self.max_session_tokens = max_session_tokens
        self.cumulative_tokens = 0

        vertexai.init(project=project_id, location=location)
        self.model = GenerativeModel(model_name)
        print(f"Agent initialized for model '{model_name}' with max session tokens: {max_session_tokens}")

    def _check_circuit_breaker(self, estimated_next_call_tokens: int):
        projected_total = self.cumulative_tokens + estimated_next_call_tokens
        if projected_total > self.max_session_tokens:
            raise AgentCircuitBreaker(
                f"Circuit breaker tripped! Projected tokens ({projected_total}) exceed session limit ({self.max_session_tokens})."
            )
        print(f"Circuit breaker check OK. Cumulative tokens: {self.cumulative_tokens}, Estimated next call: {estimated_next_call_tokens}")

    def run_interaction(self, system_prompt: str, user_queries: List[str]):
        responses = []
        # --- Caching logic --- #
        # Create a cache. In a real app, you would reuse this cache across sessions.
        # The content of the cache is automatically managed by the SDK.
        # IMPORTANT: Caching in the SDK requires an exact match of the cached prefix.
        # Here, 'system_prompt' is our prefix.
        cached_content = caching.CachedContent.create(
            model=self.model.model_name,
            contents=[Part.from_text(system_prompt)]
        )
        print(f"Vertex AI Context Cache created. TTL: {cached_content.expire_time}")

        # The system prompt is now cached. We only pay the full price once.
        # We can get the token count from the created cache for our circuit breaker.
        # Note: This is a conceptual example. The actual token count for cache creation isn't directly exposed this way.
        # We'll use the response metadata for accurate accounting.

        try:
            for i, query in enumerate(user_queries):
                full_prompt_for_estimation = query if i > 0 else system_prompt + query
                self._check_circuit_breaker(estimate_tokens(full_prompt_for_estimation))

                # For subsequent calls, we use the cache and only send the new user query.
                # The SDK handles combining the cached content with the new content.
                contents_for_call = [Part.from_text(query)]

                # Use the cache in the generation request
                response = self.model.generate_content(contents_for_call, cached_content=cached_content)

                # --- Use official token count from response for circuit breaker --- #
                input_tokens = response.usage_metadata.prompt_token_count
                output_tokens = response.usage_metadata.candidates_token_count
                is_cached = response.usage_metadata.total_token_count < (input_tokens + output_tokens)

                self.cumulative_tokens += input_tokens
                responses.append(response.text)

                print(f"  Query {i+1} processed. Input tokens (billed): {input_tokens}, Output tokens: {output_tokens}")
                print(f"  Cumulative billed input tokens: {self.cumulative_tokens}. Cache hit: {is_cached}")
                print(f"  Response: {response.text.splitlines()[0]}...")

        except AgentCircuitBreaker as e:
            print(f"\nSESSION TERMINATED: {e}")
            responses.append(f"Agent terminated early: {e}")
        finally:
            # Clean up the cache
            cached_content.delete()
            print("Vertex AI Context Cache deleted.")

        return responses

# --- Main Execution --- #
if __name__ == "__main__":
    # This is a conceptual example. You would need to set up authentication.
    # export GOOGLE_APPLICATION_CREDENTIALS="/path/to/your/key.json"
    GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "your-gcp-project-id")
    GCP_LOCATION = "europe-west4"

    # Illustrative model name for 2026
    MODEL_NAME = "gemini-1.5-pro-preview-0409"

    # Long system prompt to demonstrate caching benefits
    system_prompt = "You are a helpful financial assistant providing concise market analysis... " * 500
    # In a real scenario, this would need to be large enough to meet the provider's min token count for caching.

    user_queries = [
        "What is the current outlook for wind energy investments in EU-West1?",
        "How have carbon credit prices reacted to recent energy crises?",
        "Provide an overview of regulatory changes affecting green bonds in 2025."
    ]

    print("\n=== Testing Agent with Circuit Breaker (GCP) ===")
    # Set a tight token budget to demonstrate the circuit breaker
    agent = VertexAIAgent(GCP_PROJECT_ID, GCP_LOCATION, MODEL_NAME, max_session_tokens=1000)
    responses = agent.run_interaction(system_prompt, user_queries)

Output atteso (concettuale): l'output mostrerà l'agente che effettua le chiamate. I usage_metadata dall'SDK di Vertex AI forniscono l'esatto prompt_token_count. Per un hit della cache, questo conteggio sarà molto inferiore alla chiamata iniziale, riflettendo solo i token nella nuova query dell'utente. Se i token cumulativi superano max_session_tokens, verrà sollevata l'eccezione AgentCircuitBreaker, terminando la sessione.

Per verificare l'implementazione, eseguire lo script dell'agente con le credenziali cloud configurate. Monitorare l'output del log per:

• Hit della cache: dopo la prima chiamata, le chiamate successive dovrebbero mostrare un prompt_token_count significativamente inferiore nei usage_metadata restituiti dall'SDK. Ciò conferma che la cache viene utilizzata.
• Attivazione dell'interruttore di circuito: impostare un limite max_session_tokens basso. L'agente dovrebbe interrompere l'elaborazione delle query e sollevare l'eccezione AgentCircuitBreaker una volta che il conteggio cumulativo dei token supera questa soglia. Ciò conferma che il meccanismo di controllo dei costi funziona.

1. Errore: AgentCircuitBreaker: Agent circuit breaker tripped! Soluzione: questo è il comportamento previsto dell'interruttore di circuito. Per consentire interazioni più lunghe, aumentare il max_session_tokens durante l'inizializzazione dell'agente.

2. Errore: Mancata corrispondenza della cache nelle chiamate successive, anche con prompt di sistema identico. Soluzione: il caching richiede che la parte memorizzata nella cache del prompt sia esattamente identica. Assicurarsi che non vengano aggiunti spazi extra o caratteri. Verificare inoltre di soddisfare i requisiti minimi del provider per il caching, come il conteggio minimo di token nel contesto memorizzato nella cache. Per le API LLM reali, verificare i parametri di caching specifici del provider tramite la loro documentazione API, ad esempio i documenti di caching di google o i documenti di caching di openai.

• Il caching del contesto è cruciale per le interazioni multi-turno degli agenti; prestare sempre molta attenzione al requisito di corrispondenza esatta e al numero minimo di token per massimizzare i risparmi.
• Il benchmarking di modelli come gemini, gpt e opus per 1m token di input è essenziale per una selezione informata del provider e del modello basata sull'efficienza dei costi.
• Gli interruttori di circuito agentici sono una salvaguardia non negoziabile che previene cicli di ragionamento costosi e fornisce un controllo granulare sulla spesa degli agenti AI.
• Le best practice di sicurezza relative alle credenziali, all'isolamento dell'ambiente e alla supply chain sono fondamentali quando si distribuiscono agenti autonomi.

Il passo successivo più importante e attuabile per qualsiasi project manager o architetto AI è integrare questi controlli dei costi fin dal primo giorno. Un po' di disciplina architettonica preventiva consente di risparmiare molti costi operativi.

Risorse esterne:

• Caching e costi (fornitori): caching dei prompt openai, caching delle api gemini, panoramica della cache di contesto vertex ai, prezzi amazon bedrock
• SDK Python ufficiali:
  • client python di google cloud vertex ai
  • azure openai python (azure-ai-openai)
  • boto3 bedrock-runtime

Esempi di codice aggiuntivi:

Per modelli di agenti più avanzati e tecniche di ottimizzazione dei costi, consultare il repository github di langchain o gli esempi di hugging face transformers, adattando i loro esempi per incorporare il conteggio esplicito dei token e i controlli di budget.