部署多步驟代理工作流程 (例如大規模文件合成和長時間研究) 時，在即時模型上執行背景代理可能會造成不必要的基礎架構壓力，並觸發資源耗盡 (429) 錯誤。
Gemini Enterprise Agent Platform 提供延遲層，這是專為容許延遲的工作負載設計的排程器，可最佳化輸送量。排程器不會像處理即時通訊查詢一樣，立即處理長時間執行的自主任務，而是將複雜的多步驟代理工作流程排入離峰時段，以提高成功率和整體處理量。
使用延後層級提交要求時，API 會非同步接受工作，並立即傳回互動 ID。延後層級提供下列功能：
折扣費率：與標準要求相比，您可享有模型推論價格 50% 的折扣，因此可以控管生產環境中的代理程式成本。詳情請參閱「定價」一文。
提高處理量：延遲層級會將大量非同步工作負載移至離峰時段，藉此減輕 429 錯誤 (模型容量限制) 和速率限制，為即時生產需求釋出標準層級配額。
完成逾時：延遲層級的目標是在 24 小時內完成 95% 的工作。如果工作未在此時間範圍內完成，就會過期並轉換為
failed狀態。實際排隊時間取決於當前區域叢集的容量和需求。
用途
延遲層級適合可容許數小時處理時間的用途，例如：
金融：每日或每週的股票和市場研究。
法律和法規遵循：多份文件的法規和併購盡職調查。
策略：持續收集競爭情報並綜合分析趨勢。
安全性：掃描及修正程式碼集安全漏洞。
支援的代理程式
您可以為Deep Research Agent設定自主代理排程。
建立延後工作
以下範例說明如何使用 client.interactions.create()，透過延遲層級啟動 Deep Research 工作：
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}")
這個方法會立即傳回 status="in_progress" 和 service_tier="deferred"。
監控工作進度
等待離峰容量和主動執行時，互動 status 會維持 in_progress 狀態。代理程式執行規劃、搜尋和分析步驟時，新項目會附加至 steps 清單。
您可以透過程式輔助方式追蹤工作狀態，方法是定期輪詢互動或串流更新。
意見調查
定期輪詢互動 (例如每 15 到 30 秒)，直到互動達到下列其中一個終端狀態：completed、failed 或 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}.")
串流
只要設定 stream=True 以及 background=True 和 store=True，即可在互動進入 in_progress 狀態後，即時串流更新。串流會推送事件，例如中繼想法、文字差異和狀態更新。
如果工作仍在 in_progress 時連線中斷，您可以使用 client.interactions.get() 搭配 stream=True 重新連線至串流，並將上次收到的事件 ID 傳遞至 last_event_id。如果省略 last_event_id，API 會從頭開始重播每個事件。
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}")
取消工作
工作狀態為 queued、in_progress 或 requires_action 時，您可以取消工作。取消工作後，工作狀態會轉換為 cancelled。
如要取消工作，請使用 client.interactions.cancel()：
client.interactions.cancel(INTERACTION_ID)
擷取最終輸出內容和權杖用量
當互動達到 completed 狀態時，完整轉錄稿會顯示在 steps 清單中。最終答案是最後一個產生輸出的步驟的文字內容。
由於互動會儲存 (store=True)，因此您隨時都能使用任何工作階段的互動 ID 擷取結果：
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}"
)