Lorsque vous déployez des workflows d'agent en plusieurs étapes (tels que la synthèse de documents à grande échelle et la recherche de longue durée), l'exécution d'agents en arrière-plan sur des modèles en temps réel peut créer une pression inutile sur l'infrastructure et déclencher des erreurs d'épuisement des ressources (429).
Gemini Enterprise Agent Platform propose un niveau différé, un planificateur optimisé pour le débit conçu spécifiquement pour les charges de travail tolérant la latence Au lieu de traiter les tâches autonomes de longue durée avec la même urgence immédiate qu'une requête de chat en direct, le planificateur met en file d'attente vos workflows d'agent complexes en plusieurs étapes pendant les heures creuses afin de cibler des taux de réussite élevés et un débit global.
Lorsque vous envoyez une requête à l'aide du niveau différé, l'API accepte la tâche de manière asynchrone et renvoie immédiatement un ID d'interaction. Le niveau différé présente les caractéristiques suivantes :
Tarif réduit : vous bénéficiez d'une remise de 50% sur le prix de l'inférence de modèle par rapport à une requête standard. Vous pouvez ainsi gérer le coût de votre agent en production. Pour en savoir plus, consultez la section Tarifs.
Débit plus élevé : le niveau différé atténue les erreurs 429 (contraintes de capacité du modèle ) et les limites de débit en déplaçant vos charges de travail asynchrones lourdes en dehors des heures de pointe, ce qui libère le quota du niveau standard pour vos besoins de production en temps réel.
Délai d'achèvement : le niveau différé vise à effectuer 95% des tâches dans les 24 heures. Si une tâche n'est pas terminée dans ce délai, elle expire et passe à l'état
failed. Le temps réel passé dans la file d'attente dépend de la capacité et de la demande actuelles du cluster régional.
Cas d'utilisation
Le niveau différé est adapté aux cas d'utilisation qui peuvent tolérer des délais de traitement de plusieurs heures, comme les exemples suivants :
Finance : études quotidiennes ou hebdomadaires sur les actions et le marché.
Mentions légales et conformité : diligence raisonnable réglementaire et en matière de fusions-acquisitions sur plusieurs documents.
Stratégie : veille concurrentielle continue et synthèse des tendances.
Sécurité : analyse et correction des failles de la base de code.
Agents compatibles
Vous pouvez configurer la planification autonome des agents pour l'agent de recherche approfondie.
Créer une tâche différée
L'exemple suivant montre comment démarrer une
tâche de recherche approfondie
à l'aide du niveau différé avec 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}")
La méthode renvoie immédiatement status="in_progress" et
service_tier="deferred".
Suivre la progression des tâches
En attendant la capacité hors pointe et pendant l'exécution active, l'interaction
status reste in_progress. À mesure que l'agent exécute les étapes de planification, de recherche et d'analyse, de nouveaux éléments sont ajoutés à la liste steps.
Vous pouvez suivre l'état de la tâche par programmation en interrogeant l'interaction périodiquement ou en diffusant des mises à jour.
Sondages
Interrogez l'interaction périodiquement (par exemple, toutes les 15 à 30 secondes) jusqu'à ce qu'elle atteigne l'un des états finaux : completed, failed ou 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
Vous pouvez diffuser des mises à jour en temps réel une fois que l'interaction passe à l'état in_progress en définissant stream=True avec background=True et store=True.
Le flux envoie des événements tels que des pensées intermédiaires, des deltas de texte et des mises à jour d'état au fur et à mesure qu'ils se produisent.
Si la connexion est interrompue alors que la tâche est toujours in_progress, vous pouvez vous reconnecter au flux à l'aide de client.interactions.get() avec stream=True et transmettre l'ID du dernier événement reçu à last_event_id. Si vous omettez last_event_id, l'API relit chaque événement depuis le début.
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}")
Annuler une tâche
Vous pouvez annuler une tâche lorsque son état est queued, in_progress ou requires_action. Lorsque vous annulez une tâche, son état passe à cancelled.
Pour annuler une tâche, utilisez client.interactions.cancel() :
client.interactions.cancel(INTERACTION_ID)
Récupérer la sortie finale et l'utilisation des jetons
Lorsque l'interaction atteint l'état completed, la transcription complète est disponible dans la liste steps. La réponse finale est le contenu textuel de la dernière étape qui a produit une sortie.
Comme l'interaction est stockée (store=True), vous pouvez récupérer le résultat à tout moment à l'aide de l'ID d'interaction de n'importe quelle session :
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}"
)