Quando esegui il deployment di workflow agentici in più passaggi (ad esempio la sintesi di documenti su larga scala e la ricerca a lunga esecuzione), l'esecuzione di agenti in background su modelli in tempo reale può creare una pressione non necessaria sull'infrastruttura e attivare errori di esaurimento delle risorse (429).
Gemini Enterprise Agent Platform offre un livello differito, uno scheduler ottimizzato per la velocità effettiva progettato specificamente per i carichi di lavoro con tolleranza alla latenza Anziché trattare le attività autonome a lunga esecuzione con la stessa urgenza immediata di una query di chat live, lo scheduler mette in coda i workflow agentici complessi in più passaggi nelle ore non di punta per raggiungere tassi di successo elevati e una velocità effettiva complessiva.
Quando invii una richiesta utilizzando il livello differito, l'API accetta l'attività in modo asincrono e restituisce immediatamente un ID interazione. Il livello differito presenta le seguenti caratteristiche:
Tariffa scontata: ricevi uno sconto del 50% sui prezzi dell'inferenza del modello rispetto a una richiesta standard, in modo da poter gestire il costo dell'agente in produzione. Per ulteriori informazioni, consulta la pagina Prezzi.
Velocità effettiva più elevata: il livello differito attenua gli errori 429 (vincoli di capacità del modello ) e i limiti di frequenza spostando i carichi di lavoro asincroni pesanti nelle ore non di punta , liberando la quota del livello standard per le esigenze di produzione in tempo reale.
Timeout di completamento: il livello differito ha come obiettivo il completamento del 95% delle attività entro 24 ore. Se un'attività non viene completata entro questo periodo, scade e passa allo stato
failed. Il tempo effettivo trascorso nella coda dipende dalla capacità e dalla domanda attuali del cluster regionale.
Casi d'uso
Il livello differito è adatto ai casi d'uso che possono tollerare ore di tempo di risposta, come i seguenti esempi:
Finanza: ricerca quotidiana o settimanale su azioni e mercati.
Informazioni legali e sulla conformità: due diligence normativa e di fusioni e acquisizioni su più documenti.
Strategia: intelligence competitiva continua e sintesi delle tendenze.
Sicurezza: scansione e correzione delle vulnerabilità del codebase.
Agenti supportati
Puoi configurare la pianificazione degli agenti autonomi per l'agente Deep Research.
Creare un'attività differita
L'esempio seguente mostra come avviare un'attività di Deep Research
task
utilizzando il livello differito con client.interactions.create():
import time
from google import genai
client = genai.Client(
enterprise=True,
project="PROJECT_ID",
location="global",
)
PROMPT = "Analyze the latest market trends in renewable energy storage."
DEEP_RESEARCH_AGENT = "deep-research-preview-04-2026"
interaction = client.interactions.create(
input=PROMPT,
agent=DEEP_RESEARCH_AGENT, # Agent identifier
service_tier="deferred", # Run on deferred tier for off-peak scheduling
background=True, # Return immediately instead of waiting for the answer
store=True, # Persist interaction state to poll or stream later
stream=False, # `stream` must be set to False during task creation
)
print(f"Interaction ID: {interaction.id}")
print(f"Status: {interaction.status}")
print(f"Service tier: {interaction.service_tier}")
Il metodo restituisce immediatamente status="in_progress" e
service_tier="deferred".
Monitorare l'avanzamento dell'attività
Durante l'attesa della capacità non di punta e l'esecuzione attiva, lo status dell'interazione rimane in_progress. Man mano che l'agente esegue i passaggi di pianificazione, ricerca e analisi, vengono aggiunti nuovi elementi all'elenco steps.
Puoi monitorare lo stato dell'attività a livello di programmazione eseguendo il polling periodico dell'interazione o lo streaming degli aggiornamenti.
Sondaggi
Esegui il polling periodico dell'interazione (ad esempio ogni 15-30 secondi) finché non raggiunge uno degli stati finali: completed, failed o cancelled.
TERMINAL_STATES = ("completed", "failed", "cancelled")
POLL_INTERVAL_SECONDS = 15
TIMEOUT_MINUTES = 60
started = time.time()
deadline = started + TIMEOUT_MINUTES * 60
while True:
current = client.interactions.get(interaction.id)
elapsed = int(time.time() - started)
steps = getattr(current, "steps", None) or []
print(f"[{elapsed:>4}s] status={current.status} steps={len(steps)}")
if current.status in TERMINAL_STATES:
break
if time.time() >= deadline:
raise TimeoutError(
f"Still {current.status} after {TIMEOUT_MINUTES} min. The interaction "
"continues running server-side; re-run the check to resume polling."
)
time.sleep(POLL_INTERVAL_SECONDS)
print(f"\nFinished in {int(time.time() - started)}s with status={current.status}.")
Streaming
Puoi eseguire lo streaming degli aggiornamenti in tempo reale una volta che l'interazione entra nello stato in_progress impostando stream=True insieme a background=True e store=True.
Lo stream invia eventi come pensieri intermedi, delta di testo e aggiornamenti di stato man mano che si verificano.
Se la connessione si interrompe mentre l'attività è ancora in_progress, puoi riconnetterti allo stream utilizzando client.interactions.get() con stream=True e passare l'ID dell'ultimo evento ricevuto a last_event_id. Se ometti last_event_id, l'API riproduce ogni evento dall'inizio.
INTERACTION_ID = interaction.id # from the create step
MAX_RECONNECTS = 5
STREAM_TIMEOUT = 300 # seconds
print(
f"streaming interaction: {INTERACTION_ID} (status={interaction.status})\n"
)
def render(event):
"""Prints one SSE event. Returns True once the interaction has finished."""
if event.event_type == "step.delta":
delta = event.delta
if delta.type == "text":
print(delta.text, end="", flush=True)
elif delta.type == "thought_summary":
summary = (getattr(delta.content, "text", "") or "").strip()
if summary:
print(f"\n[thinking] {summary[:200]}", flush=True)
elif delta.type.endswith("_call"):
queries = getattr(getattr(delta, "arguments", None), "queries", None)
print(
f"\n[{delta.type}] {', '.join(queries) if queries else ''}",
flush=True,
)
elif event.event_type == "interaction.status_update":
print(f"[status] {event.status}", flush=True)
elif event.event_type == "interaction.completed":
print(f"\n\n[status] {event.interaction.status}", flush=True)
return True
elif event.event_type == "error":
print(f"\n[error] {event.error.message}", flush=True)
return True
return False
last_event_id = None
finished = False
for attempt in range(MAX_RECONNECTS):
try:
# stream=True turns the GET into a live subscription. last_event_id=None on
# the first pass, so the server starts from the beginning of the run.
for event in client.interactions.get(
INTERACTION_ID,
stream=True,
last_event_id=last_event_id,
timeout=STREAM_TIMEOUT,
):
last_event_id = event.event_id or last_event_id
finished = render(event) or finished
except Exception as e: # pylint: disable=broad-except
# A dropped connection loses nothing: the run continues server-side and the
# next iteration reattaches from last_event_id.
print(f"\n[stream dropped: {type(e).__name__}] reattaching...", flush=True)
if finished:
break
# The server also closes the stream when the run ends, without an error.
if (
client.interactions.get(INTERACTION_ID, timeout=STREAM_TIMEOUT).status
!= "in_progress"
):
break
else:
print(f"\n[gave up after {MAX_RECONNECTS} reconnects]")
print(f"\n\nStreamed interaction: {INTERACTION_ID}")
Annullare un'attività
Puoi annullare un'attività mentre il suo stato è queued, in_progress o requires_action. Quando annulli un'attività, il suo stato passa a cancelled.
Per annullare un'attività, utilizza client.interactions.cancel():
client.interactions.cancel(INTERACTION_ID)
Recuperare l'output finale e l'utilizzo dei token
Quando l'interazione raggiunge lo stato completed, la trascrizione completa è disponibile nell'elenco steps. La risposta finale è il contenuto di testo dell'ultimo passaggio che ha prodotto l'output.
Poiché l'interazione è archiviata (store=True), puoi recuperare il risultato in qualsiasi momento utilizzando l'ID interazione da qualsiasi sessione:
def get_final_text(completed_interaction):
"""Returns the text of the last step that produced output."""
for step in reversed(getattr(completed_interaction, "steps", None) or []):
text = "".join(
part.text for part in (getattr(step, "content", None) or [])
if getattr(part, "text", None)
)
if text:
return text
return ""
final = client.interactions.get(interaction.id)
print(f"Status: {final.status}\n")
print(get_final_text(final) or "(No text output)")
if final.usage:
print(
f"\nToken usage:\n"
f" Input tokens: {final.usage.total_input_tokens}\n"
f" Output tokens: {final.usage.total_output_tokens}\n"
f" Total tokens: {final.usage.total_tokens}"
)