Saat men-deploy alur kerja berbasis agen multilangkah (seperti sintesis dokumen skala besar dan riset yang berjalan lama), menjalankan agen latar belakang pada model real-time dapat menimbulkan tekanan infrastruktur yang tidak perlu dan memicu error kehabisan resource (429).
Gemini Enterprise Agent Platform menawarkan tingkatan yang ditangguhkan, penjadwal yang dioptimalkan untuk throughput yang dirancang khusus untuk beban kerja yang toleran terhadap latensi. Scheduler mengantrekan alur kerja agent multi-langkah yang kompleks ke jam-jam di luar jam sibuk untuk menargetkan tingkat keberhasilan yang tinggi dan throughput secara keseluruhan, alih-alih memperlakukan tugas otonom yang berjalan lama dengan urgensi langsung yang sama seperti kueri live chat.
Saat Anda mengirimkan permintaan menggunakan tingkat yang ditangguhkan, API akan menerima tugas secara asinkron dan langsung menampilkan ID interaksi. Paket yang ditangguhkan menawarkan fitur berikut:
Tarif diskon: Anda menerima diskon 50% untuk harga inferensi model dibandingkan dengan permintaan standar, sehingga Anda dapat mengelola biaya agen dalam produksi. Untuk mengetahui informasi selengkapnya, lihat Harga.
Throughput yang lebih tinggi: Tingkat yang ditangguhkan mengurangi 429 (kendala kapasitas model) dan batas kecepatan dengan memindahkan workload asinkron yang berat di luar jam sibuk, sehingga membebaskan kuota Tingkat Standar untuk kebutuhan produksi real-time Anda.
Waktu tunggu penyelesaian: Tingkat yang ditangguhkan menargetkan penyelesaian 95% tugas dalam waktu 24 jam. Jika tugas tidak selesai dalam jangka waktu ini, tugas akan berakhir dan bertransisi ke status
failed. Waktu sebenarnya yang dihabiskan dalam antrean bergantung pada kapasitas dan permintaan cluster regional saat ini.
Kasus penggunaan
Tingkatan yang ditangguhkan cocok untuk kasus penggunaan yang dapat mentoleransi waktu penyelesaian selama beberapa jam, seperti contoh berikut:
Keuangan: Riset pasar dan ekuitas harian atau mingguan.
Hukum & kepatuhan: Uji tuntas peraturan dan merger serta akuisisi multi-dokumen.
Strategi: Kecerdasan kompetitif berkelanjutan dan sintesis tren.
Keamanan: Pemindaian dan perbaikan kerentanan codebase.
Agen yang didukung
Anda dapat mengonfigurasi penjadwalan agen otonom untuk Agen Riset Mendalam.
Membuat tugas yang ditangguhkan
Contoh berikut menunjukkan cara memulai tugas Deep Research
menggunakan tingkat yang ditangguhkan dengan 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}")
Metode ini akan segera ditampilkan dengan status="in_progress" dan
service_tier="deferred".
Memantau progres tugas
Saat menunggu kapasitas di luar jam sibuk dan berjalan secara aktif, interaksi
status tetap in_progress. Saat agen menjalankan langkah-langkah perencanaan, penelusuran, dan analisis, item baru ditambahkan ke daftar steps.
Anda dapat melacak status tugas secara terprogram dengan melakukan polling interaksi secara berkala atau melakukan streaming update.
Polling
Polling interaksi secara berkala (seperti setiap 15–30 detik) hingga
interaksi mencapai salah satu status akhir: completed, failed,
atau 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
Anda dapat melakukan streaming update secara real time setelah interaksi memasuki status in_progress
dengan menetapkan stream=True bersama background=True dan store=True.
Aliran ini mengirimkan peristiwa seperti pemikiran sementara, perbedaan teks, dan
pembaruan status saat terjadi.
Jika koneksi terputus saat tugas masih in_progress, Anda dapat
terhubung kembali ke stream menggunakan client.interactions.get() dengan stream=True dan
meneruskan ID peristiwa terakhir yang diterima ke last_event_id. Jika Anda menghilangkan
last_event_id, API akan memutar ulang setiap peristiwa dari awal.
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}")
Membatalkan tugas
Anda dapat membatalkan tugas saat statusnya queued, in_progress, atau
requires_action. Saat Anda membatalkan tugas, statusnya akan bertransisi ke
cancelled.
Untuk membatalkan tugas, gunakan client.interactions.cancel():
client.interactions.cancel(INTERACTION_ID)
Mengambil output akhir dan penggunaan token
Saat interaksi mencapai status completed, transkrip lengkap tersedia di daftar steps. Jawaban akhir adalah konten teks dari
langkah terakhir yang menghasilkan output.
Karena interaksi disimpan (store=True), Anda dapat mengambil hasilnya kapan saja menggunakan ID interaksi dari sesi mana pun:
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}"
)