使用 Gemini 模型进行延迟调度¶
智能体工作负载有不同的延迟需求。交互式助手必须立即给出答案，但摘要任务、批量评估运行或文档处理流水线则可以等待容量。延迟调度让 ADK 智能体将这些模型调用排队，改用非高峰时段的容量来运行，而不是争抢交互式容量。
你可以按每次运行来请求延迟调度，方法是使用 RunConfig 的 service_tier 设置。该设置属于运行配置，而不是模型或智能体，因此同一个智能体定义可以同时服务交互式请求和批处理工作负载。
预览：延迟容量需要允许列表访问权限
此 Google Cloud 功能属于预览（Preview）功能，要在延迟容量上运行请求，需要为你的 Google Cloud 项目开通允许列表。有关详情，请参阅自主智能体调度。
入门¶
延迟调度要求 Gemini 模型配置了 Google Cloud for Gemini Interactions API。在模型上设置 use_interactions_api=True，然后在运行智能体时传入 RunConfig(service_tier=ServiceTier.DEFERRED)，如下例所示：
import asyncio
from google.adk.agents import LlmAgent
from google.adk.agents import RunConfig
from google.adk.apps import App
from google.adk.models import ServiceTier
from google.adk.models.google_llm import Gemini
from google.adk.runners import InMemoryRunner
from google.genai import types
root_agent = LlmAgent(
name='batch_agent',
model=Gemini(
model='gemini-flash-latest',
use_interactions_api=True, # 延迟调度所必需
),
instruction='处理输入文档并生成摘要。',
)
app = App(name='batch_app', root_agent=root_agent)
runner = InMemoryRunner(app=app)
async def main() -> None:
session = await runner.session_service.create_session(
app_name=app.name,
user_id='user_123',
session_id='session_456',
)
# 为本次运行中的每一次模型调用请求非高峰时段容量。
run_config = RunConfig(service_tier=ServiceTier.DEFERRED)
async for event in runner.run_async(
user_id='user_123',
session_id=session.id,
new_message=types.Content(
role='user',
parts=[types.Part.from_text(
text='总结季度业绩指标。'
)],
),
run_config=run_config,
):
if event.content and event.content.parts:
for part in event.content.parts:
if part.text:
print(part.text)
asyncio.run(main())
运行器会提交模型调用，等待排队的任务完成，然后产出响应事件。你的代码消费事件的方式与标准运行完全相同。
等待期间运行看起来像是卡住了
请求仍在队列中排队时，run_async() 方法不会产出任何事件。这需要多长时间取决于后端负载，ADK 不设置任何上限。在 Web UI 或终端中，这种延迟看起来就像卡住了。请显示一个进度指示器，或设置客户端截止时间。
要确认该层级已生效，请在应用日志中查找以下消息：
| 日志消息 | 级别 | 含义 |
|---|---|---|
Using service_tier from run_config: deferred |
DEBUG |
ADK 已将该层级应用到请求。 |
Interaction <id> is queued; waiting for the result. |
INFO |
后端已将该任务收进队列。 |
Interaction <id> reached status completed. |
INFO |
结果已就绪。 |
run_config.service_tier=... has no effect for agent <name> |
WARNING |
ADK 丢弃了该层级。请参阅疑难排查。 |
延迟调度的工作原理¶
标准模型请求是同步的。ADK 发送请求，模型在同一连接上返回响应。当你通过设置 ServiceTier.DEFERRED 启用延迟调度时，ADK 会把请求标记为后台执行，后端会将其排队并立即返回一个交互 ID，而不是返回结果。
随后 ADK 会等待该结果，以指数退避方式查询排队的任务，并吸收瞬时读取失败，直到达到最终状态。它再把该结果转换为普通的响应事件并产出。这个循环是内部实现：它不会暴露交互 ID，你也不必编写任何检索代码。不过它会在排队等待之外再增加几秒的轮询延迟，因此延迟调度不应用于简短、对延迟敏感的调用。
等待过程的以下特性会影响你如何设计智能体：
- ADK 不设置客户端截止时间。 后端对该交互的完成超时是等待过程的唯一界限。若想更早停止，请参阅设置客户端截止时间。
- 每个模型轮次分别排队。 在调用工具的智能体中，每个轮次都会创建自己的交互，因此总延迟是每个轮次排队等待时间的总和，而不是对整次运行的单次等待。
配置选项¶
RunConfig 的 service_tier 设置会为一次运行中的每一次模型调用选择容量池：
| 选项 | 类型 | 默认值 | 描述 |
|---|---|---|---|
service_tier |
Optional[ServiceTier \| str] |
None |
本次运行中模型调用的服务层级。 |
ServiceTier 枚举定义了以下层级：
ServiceTier.DEFERRED：将调用排队，改用非高峰时段容量运行。容量紧张时，调用会等待空位，而不是失败。其他任何层级都不会改变 ADK 执行请求的方式，并且该层级不能与流式传输结合使用。ServiceTier.FLEX：以更低成本提供尽力而为的容量，不保证延迟。ServiceTier.STANDARD：默认层级。ServiceTier.PRIORITY：为延迟敏感的调用预留容量。
不设置 service_tier 时，请求中会完全省略该字段，这等同于 ServiceTier.STANDARD。ServiceTier 枚举是 str 的子类，因此你可以传入 'deferred' 这样的普通字符串来代替枚举成员。字符串形式还让你可以在 ADK 尚未定义对应常量时，就使用后端已支持的层级。
高级用法¶
以下各节介绍如何用客户端截止时间来限定延迟运行的等待时间，以及如何在通过 HTTP 提供智能体服务时请求延迟调度。
设置客户端截止时间¶
延迟请求会等待非高峰时段容量，其耗时取决于后端负载。要限定总耗时的上限，可把运行包裹在 asyncio.timeout() 中，这需要 Python 3.11 或更高版本：
import asyncio
import logging
from google.adk.agents import RunConfig
from google.adk.models import ServiceTier
from google.genai import types
logger = logging.getLogger(__name__)
# 承接「入门」示例，复用 runner 和 session。
message = types.Content(
role='user',
parts=[types.Part.from_text(text='生成一份季度总结。')],
)
run_config = RunConfig(service_tier=ServiceTier.DEFERRED)
try:
async with asyncio.timeout(300):
async for event in runner.run_async(
user_id='user_123',
session_id=session.id,
new_message=message,
run_config=run_config,
):
if event.content and event.content.parts:
for part in event.content.parts:
if part.text:
print(part.text)
except TimeoutError:
logger.error('延迟运行超过了 300 秒的客户端截止时间。')
客户端截止时间不会取消请求
超时只会让 ADK 停止轮询结果，并不会让后端停止。排队中的请求仍会运行完成并产生计费用量，而且之后你无法检索到它的输出。请把超时的轮次视为作废，不要用客户端截止时间来限制用量。
通过 HTTP 请求延迟调度¶
你用 adk api_server 提供服务的智能体，可以在 /run 和 /run_sse 端点的请求体中接受 service_tier：
{
"app_name": "batch_app",
"user_id": "user_123",
"session_id": "session_456",
"new_message": {
"role": "user",
"parts": [{"text": "总结批次结果。"}]
},
"service_tier": "deferred"
}
对于 /run_sse 请求，还必须设置 "streaming": false。把 "service_tier": "deferred" 与 "streaming": true 结合使用会返回 HTTP 422。HTTP 请求会在整个排队等待期间保持连接打开，这使得托管平台的请求超时成为延迟运行的实际限制：
- 调高代理和入口（ingress）超时。 负载均衡器和入口控制器默认会关闭长时间运行的后端连接。请检查你所用平台的限制，例如 Cloud Run 请求超时，并将其调高到能覆盖你预期的排队等待时间。
- 客户端断开连接只会取消检索，不会取消执行。 连接关闭时，服务器会取消它的轮询任务。排队中的请求仍然会运行，也仍然会计费，而你会丢失它的输出。
对于可能长时间等待的延迟工作负载，请改为部署到 Google Cloud Agent Platform 上的 Agent Runtime。Agent Runtime 在托管容器中运行该调用，不会为此保持 HTTP 连接打开。
限制¶
延迟调度有以下限制：
- 需要允许列表访问权限： 延迟容量要求为你的 Google Cloud 项目开通允许列表。
- 仅限 Gemini 和 Interactions API： 延迟调度只适用于设置了
use_interactions_api=True的Gemini模型。其他任何模型（包括自定义的BaseLlm子类）都会忽略该层级。 - 与流式传输不兼容： 构造
RunConfig(service_tier=ServiceTier.DEFERRED, streaming_mode=StreamingMode.SSE)会抛出pydantic.ValidationError。 ManagedAgent类忽略该层级： 它运行自己的交互循环，从不读取service_tier。- 不支持跨重启恢复： ADK 不会持久化进行中的交互 ID，也不提供把一次运行重新附加到已排队交互的方式。如果客户端进程停止，ADK 会放弃待处理的工作，下一次运行会创建新的交互。
疑难排查¶
以下各节介绍使用延迟调度时的一些常见问题及其解决方法。
ADK 忽略了该层级，而运行仍然成功¶
如果智能体的模型不是设置了 use_interactions_api=True 的 Gemini 实例，ADK 会丢弃该层级、每次运行记录一次警告，并在标准容量上执行该调用。运行会成功，因此日志是唯一的信号：
run_config.service_tier=... has no effect for agent <name>: its model does not
use the interactions API, which is the only path with a serving tier. Set
use_interactions_api=True on the model to apply the tier.
这个警告是故意为之。在多智能体运行中，可能只有部分智能体使用 Interactions API，因此对于不可用的层级，ADK 只会发出警告，而不是抛出异常。在日志中搜索 has no effect for agent，即可找到需要 use_interactions_api=True 的模型。
OpenAI service_tier 字段¶
OpenAIResponsesLlm 类也有一个 service_tier 字段。它是一个互不相关的设置：你在模型上设置它，而不是在运行上设置，并且 RunConfig.service_tier 不会传入其中。设置其中一个对另一个没有任何影响。