Bei der Bereitstellung mehrstufiger agentischer Workflows (z. B. groß angelegte Dokumentsynthese und langwierige Recherchen) kann die Ausführung von Hintergrund-KI-Agenten auf Echtzeitmodellen zu unnötigem Infrastrukturdruck führen und Fehler aufgrund von Ressourcenerschöpfung (429) auslösen.
Die Gemini Enterprise Agent Platform bietet eine verzögerte Ebene, einen durchsatzoptimierten Planer, der speziell für latenztolerante Arbeitslasten entwickelt wurde. Anstatt langwierige autonome Aufgaben mit derselben sofortigen Dringlichkeit wie eine Live-Chat-Abfrage zu behandeln, reiht der Planer Ihre komplexen mehrstufigen agentischen Workflows in die Warteschlange für Zeiten mit geringer Auslastung ein, um hohe Erfolgsraten und einen hohen Gesamtdurchsatz zu erzielen.
Wenn Sie eine Anfrage über die verzögerte Ebene senden, akzeptiert die API die Aufgabe asynchron und gibt sofort eine Interaktions-ID zurück. Die verzögerte Ebene bietet Folgendes:
Rabattierter Preis: Sie erhalten einen Rabatt von 50% auf die Preise für die Modellinferenz im Vergleich zu einer Standardanfrage. So können Sie Ihre Agentenkosten in der Produktion besser verwalten. Weitere Informationen finden Sie unter Preise.
Höherer Durchsatz: Die verzögerte Ebene reduziert 429-Fehler (Einschränkungen der Modellkapazität ) und Ratenlimits, indem sie Ihre großen, asynchronen Arbeitslasten in Zeiten mit geringer Auslastung verschiebt. So wird Kontingent der Standardebene für Ihre Echtzeit-Produktionsanforderungen freigegeben.
Zeitüberschreitung bei Abschluss: Die verzögerte Ebene zielt darauf ab, 95% der Aufgaben innerhalb von 24 Stunden abzuschließen. Wenn eine Aufgabe nicht innerhalb dieses Zeitraums abgeschlossen wird, läuft sie ab und wechselt in den Status
failed. Die tatsächliche Zeit in der Warteschlange hängt von der aktuellen regionalen Clusterkapazität und -nachfrage ab.
Anwendungsfälle
Die verzögerte Ebene eignet sich gut für Anwendungsfälle, bei denen eine Bearbeitungszeit von mehreren Stunden akzeptabel ist, z. B. für die folgenden Beispiele:
Finanzen: Tägliche oder wöchentliche Aktien- und Marktforschung.
Rechtliche Hinweise und Compliance: Due-Diligence-Prüfung in Bezug auf Vorschriften und Fusionen und Übernahmen für mehrere Dokumente.
Strategie: Kontinuierliche Wettbewerbsanalyse und Trendsynthese.
Sicherheit: Scannen der Codebasis auf Sicherheitslücken und Behebung von Sicherheitslücken.
Unterstützte KI-Agenten
Sie können die autonome Agentenplanung für den Deep Research Agent konfigurieren.
Verzögerte Aufgabe erstellen
Das folgende Beispiel zeigt, wie Sie eine Deep Research
Aufgabe
mit der verzögerten Ebene und client.interactions.create() starten:
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}")
Die Methode wird sofort mit status="in_progress" und
service_tier="deferred" zurückgegeben.
Aufgabenfortschritt beobachten
Während auf Kapazität außerhalb der Spitzenzeiten gewartet wird und die Aufgabe aktiv ausgeführt wird, bleibt der status der Interaktion in_progress. Wenn der Agent Planungs-, Such- und Analyseschritte ausführt, werden der Liste steps neue Elemente hinzugefügt.
Sie können den Aufgabenstatus programmatisch verfolgen, indem Sie die Interaktion regelmäßig abfragen oder Updates streamen.
Umfragen
Fragen Sie die Interaktion regelmäßig ab (z. B. alle 15 bis 30 Sekunden), bis sie einen der Endzustände erreicht: completed, failed oder 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
Sie können Updates in Echtzeit streamen, sobald die Interaktion den Status in_progress erreicht. Dazu legen Sie stream=True neben background=True und store=True fest.
Der Stream überträgt Ereignisse wie Zwischengedanken, Textänderungen und Statusaktualisierungen, sobald sie auftreten.
Wenn die Verbindung unterbrochen wird, während die Aufgabe noch den Status in_progress hat, können Sie die Verbindung zum Stream mit client.interactions.get() und stream=True wiederherstellen und die zuletzt empfangene Ereignis-ID an last_event_id übergeben. Wenn Sie last_event_id weglassen, spielt die API jedes Ereignis von Anfang an ab.
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}")
Aufgabe abbrechen
Sie können eine Aufgabe abbrechen, wenn ihr Status queued, in_progress oder requires_action ist. Wenn Sie eine Aufgabe abbrechen, wechselt ihr Status zu cancelled.
Verwenden Sie client.interactions.cancel(), um eine Aufgabe abzubrechen:
client.interactions.cancel(INTERACTION_ID)
Endgültige Ausgabe und Tokennutzung abrufen
Wenn die Interaktion den Status completed erreicht, ist das vollständige Transkript in der Liste steps verfügbar. Die endgültige Antwort ist der Textinhalt des letzten Schritts, der eine Ausgabe erzeugt hat.
Da die Interaktion gespeichert ist (store=True), können Sie das Ergebnis jederzeit mit der Interaktions-ID aus einer beliebigen Sitzung abrufen:
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}"
)