部署多步骤智能体工作流（例如大规模文档合成和长时间运行的研究）时，在实时模型上运行后台智能体可能会造成不必要的基础设施压力，并触发资源耗尽 (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 状态时连接断开，您可以使用 stream=True 通过 client.interactions.get() 重新连接到数据流，并将上次收到的事件 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)
检索最终输出和 token 用量
当互动达到 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}"
)