Skip to content

第 4 部分:了解 RunConfig

在第 3 部分中,你学习了如何处理来自 run_live() 的事件以处理模型响应、工具调用和流式更新。本部分向你展示如何通过 RunConfig 配置这些流式会话——控制响应格式、管理会话生命周期和执行生产约束。

你将学到什么:本部分涵盖响应模态及其约束,探讨 BIDI 和 SSE 流式模式之间的差异,检查 ADK 会话与 Live API 会话之间的关系,并展示如何通过会话恢复和上下文窗口压缩来管理会话持续时间。你将了解如何处理并发会话配额,实施配额管理的架构模式,通过 max_llm_calls 和音频持久性选项配置成本控制,并实时跟踪令牌使用情况以进行生产监控(v1.18.0 中的新功能)。掌握了 RunConfig,你可以构建平衡功能丰富性与操作约束的生产就绪流式应用程序。

了解更多

有关音频/视频相关 RunConfig 配置的详细信息,请参阅 第 5 部分:Live API 中的音频、图像和视频

RunConfig 参数快速参考

此表提供了本部分涵盖的所有 RunConfig 参数的快速参考:

参数 类型 用途 平台支持 参考
response_modalities list[str] 控制输出格式(TEXT 或 AUDIO) 两者 详情
streaming_mode StreamingMode 选择 BIDI 或 SSE 模式 两者 详情
session_resumption SessionResumptionConfig 启用自动重新连接 两者 详情
context_window_compression ContextWindowCompressionConfig 无限会话持续时间 两者 详情
max_llm_calls int 限制每个会话的总 LLM 调用 两者 详情
save_live_blob bool 持久化音频/视频流 两者 详情
custom_metadata dict[str, Any] 将元数据附加到调用事件 两者 详情
support_cfc bool 启用组合函数调用 Gemini(仅限 2.x 模型) 详情
speech_config SpeechConfig 语音和语言配置 两者 第 5 部分:语音配置
input_audio_transcription AudioTranscriptionConfig 转录用户语音 两者 第 5 部分:音频转录
output_audio_transcription AudioTranscriptionConfig 转录模型语音 两者 第 5 部分:音频转录
realtime_input_config RealtimeInputConfig VAD 配置 两者 第 5 部分:语音活动检测
proactivity ProactivityConfig 启用主动音频 Gemini(仅限原生音频) 第 5 部分:主动性和情感对话
enable_affective_dialog bool 情感适应 Gemini(仅限原生音频) 第 5 部分:主动性和情感对话

源引用

run_config.py

平台支持图例:

  • 两者:在 Gemini Live API 和 Vertex AI Live API 上均受支持
  • Gemini:仅在 Gemini Live API 上受支持
  • 特定于模型:需要特定的模型架构(例如,原生音频)

导入路径:

上表中引用的所有配置类型类均从 google.genai.types 导入:

from google.genai import types
from google.adk.agents.run_config import RunConfig, StreamingMode

# 配置类型通过 types 模块访问
run_config = RunConfig(
    session_resumption=types.SessionResumptionConfig(),
    context_window_compression=types.ContextWindowCompressionConfig(...),
    speech_config=types.SpeechConfig(...),
    # 等等
)

RunConfig 类本身和 StreamingMode 枚举从 google.adk.agents.run_config 导入。

响应模态

响应模态控制模型如何生成输出——作为文本或音频。Gemini Live API 和 Vertex AI Live API 都有相同的限制:每个会话只有一个响应模态。

配置:

# 第 2 阶段:会话初始化 - RunConfig 决定流式行为

# 默认行为:ADK 在未指定时自动将 response_modalities 设置为 ["AUDIO"]
# (原生音频模型需要)
run_config = RunConfig(
    streaming_mode=StreamingMode.BIDI  # 双向 WebSocket 通信
)

# 上面等同于:
run_config = RunConfig(
    response_modalities=["AUDIO"],  # 由 ADK 在 run_live() 中自动设置
    streaming_mode=StreamingMode.BIDI  # 双向 WebSocket 通信
)

# ✅ 正确:仅文本响应
run_config = RunConfig(
    response_modalities=["TEXT"],  # 模型仅以文本响应
    streaming_mode=StreamingMode.BIDI  # 仍然使用双向流式处理
)

# ✅ 正确:仅音频响应(显式)
run_config = RunConfig(
    response_modalities=["AUDIO"],  # 模型仅以音频响应
    streaming_mode=StreamingMode.BIDI  # 双向 WebSocket 通信
)

Gemini Live API 和 Vertex AI Live API 都将会话限制为单个响应模态。尝试同时使用两者将导致 API 错误:

# ❌ 不正确:不支持同时使用两种模态
run_config = RunConfig(
    response_modalities=["TEXT", "AUDIO"],  # 错误:不能同时使用
    streaming_mode=StreamingMode.BIDI
)
# 来自 Live API 的错误:"Only one response modality is supported per session"

默认行为:

当未指定 response_modalities 时,ADK 的 run_live() 方法会自动将其设置为 ["AUDIO"],因为原生音频模型需要显式的响应模态。如果需要,你可以通过显式设置 response_modalities=["TEXT"] 来覆盖此设置。

关键约束:

  • 你必须在会话开始时选择 TEXTAUDIO不能在会话中途切换模态
  • 对于 原生音频模型,你必须选择 AUDIO。如果你想从原生音频模型接收音频和文本响应,请使用音频转录功能,该功能提供音频输出的文本转录。有关详细信息,请参阅 音频转录
  • 响应模态仅影响模型输出——你始终可以发送文本、语音或视频输入(如果模型支持这些输入模态),无论选择的响应模态如何

StreamingMode:BIDI 或 SSE

ADK 支持两种不同的流式模式,它们使用不同的 API 端点和协议:

  • StreamingMode.BIDI:ADK 使用 WebSocket 连接到 Live API(通过 live.connect() 的双向流式端点)
  • StreamingMode.SSE:ADK 使用 HTTP 流式传输连接到 标准 Gemini API(通过 generate_content_async() 的一元/流式端点)

“Live API”专门指双向 WebSocket 端点 (live.connect()),而“Gemini API”或“标准 Gemini API”指传统的基于 HTTP 的端点 (generate_content() / generate_content_async())。两者都是更广泛的 Gemini API 平台的一部分,但使用不同的协议和功能。

注意: 这些模式指的是 ADK 到 Gemini API 的通信协议,而不是你的应用程序的面向客户端的架构。你可以使用任一模式为你的客户端构建 WebSocket 服务器、REST API、SSE 端点或任何其他架构。

本指南重点介绍 StreamingMode.BIDI,这是实时音频/视频交互和 Live API 功能所必需的。但是,了解 BIDI 和 SSE 模式之间的差异以选择适合你用例的方法是值得的。

配置:

from google.adk.agents.run_config import RunConfig, StreamingMode

# 用于实时音频/视频的 BIDI 流式处理
run_config = RunConfig(
    streaming_mode=StreamingMode.BIDI,
    response_modalities=["AUDIO"]  # 支持音频/视频模态
)

# 用于基于文本的交互的 SSE 流式处理
run_config = RunConfig(
    streaming_mode=StreamingMode.SSE,
    response_modalities=["TEXT"]  # 仅文本模态
)

协议和实现差异

这两种流式模式在通信模式和功能上存在根本差异。BIDI 模式启用真正的双向通信,你可以在接收模型响应的同时发送新输入,而 SSE 模式遵循传统的请求-响应模式,你发送完整的请求并流式传回响应。

StreamingMode.BIDI - 双向 WebSocket 通信:

BIDI 模式建立持久的 WebSocket 连接,允许同时发送和接收。这启用了实时功能,如中断、实时音频流和立即轮流:

sequenceDiagram
    participant App as 你的应用程序
    participant ADK as ADK
    participant Queue as LiveRequestQueue
    participant Gemini as Gemini Live API

    Note over ADK,Gemini: "协议:WebSocket"

    App->>ADK: runner.run_live(run_config)
    ADK->>Gemini: live.connect() - WebSocket
    activate Gemini

    Note over ADK,Queue: "可以在接收时发送"

    App->>Queue: send_content(text)
    Queue->>Gemini: "→ 内容(通过 WebSocket)"
    App->>Queue: send_realtime(audio)
    Queue->>Gemini: "→ 音频 blob(通过 WebSocket)"

    Gemini-->>ADK: "← 部分响应 (partial=True)"
    ADK-->>App: "← 事件:部分文本/音频"
    Gemini-->>ADK: "← 部分响应 (partial=True)"
    ADK-->>App: "← 事件:部分文本/音频"

    App->>Queue: send_content(interrupt)
    Queue->>Gemini: "→ 新内容"

    Gemini-->>ADK: ← turn_complete=True
    ADK-->>App: "← 事件:回合完成"

    deactivate Gemini

    Note over ADK,Gemini: "回合检测:turn_complete 标志"

StreamingMode.SSE - 单向 HTTP 流式传输:

SSE(服务器发送事件)模式使用 HTTP 流式传输,你预先发送完整的请求,然后作为块流接收响应。这是一种更简单、更传统的模式,适用于基于文本的聊天应用程序:

sequenceDiagram
    participant App as 你的应用程序
    participant ADK as ADK
    participant Gemini as Gemini API

    Note over ADK,Gemini: "协议:HTTP"

    App->>ADK: runner.run(run_config)
    ADK->>Gemini: generate_content_stream() - HTTP
    activate Gemini

    Note over ADK,Gemini: "请求发送完毕,然后流式传输响应"

    Gemini-->>ADK: "← 部分块 (partial=True)"
    ADK-->>App: "← 事件:部分文本"
    Gemini-->>ADK: "← 部分块 (partial=True)"
    ADK-->>App: "← 事件:部分文本"
    Gemini-->>ADK: "← 部分块 (partial=True)"
    ADK-->>App: "← 事件:部分文本"

    Gemini-->>ADK: "← 最终块 (finish_reason=STOP)"
    ADK-->>App: "← 事件:完整响应"

    deactivate Gemini

    Note over ADK,Gemini: "回合检测:finish_reason"

何时使用每种模式

你在 BIDI 和 SSE 之间的选择取决于你的应用程序要求和你需要支持的交互模式。这是一份实用指南,可帮助你进行选择:

在以下情况下使用 BIDI:

  • 构建具有实时交互的语音/视频应用程序
  • 需要双向通信(在接收时发送)
  • 需要 Live API 功能(音频转录、VAD、主动性、情感对话)
  • 支持中断和自然轮流(见 第 3 部分:处理中断标志
  • 实施实时流式工具或实时数据源
  • 可以规划并发会话配额(取决于平台/层级,50-1,000 个会话)

在以下情况下使用 SSE:

  • 构建基于文本的聊天应用程序
  • 标准请求/响应交互模式
  • 使用不支持 Live API 的模型(例如,Gemini 1.5 Pro, Gemini 1.5 Flash)
  • 更简单的部署,无需 WebSocket 要求
  • 需要更大的上下文窗口(Gemini 1.5 支持高达 2M 令牌)
  • 相比并发会话配额,更喜欢标准 API 速率限制 (RPM/TPM)

流式模式和模型兼容性

SSE 模式通过 HTTP 流式传输使用标准 Gemini API (generate_content_async),而 BIDI 模式通过 WebSocket 使用 Live API (live.connect())。Gemini 1.5 模型 (Pro, Flash) 不支持 Live API 协议,因此必须与 SSE 模式一起使用。Gemini 2.0/2.5 Live 模型支持这两种协议,但通常与 BIDI 模式一起使用以访问实时音频/视频功能。

通过 SSE 访问的标准 Gemini 模型(1.5 系列)

虽然本指南重点介绍使用 Gemini 2.0 Live 模型的双向流式处理,但 ADK 也通过 SSE 流式传输支持 Gemini 1.5 模型系列。这些模型提供了不同的权衡——更大的上下文窗口和经过验证的稳定性,但没有实时音频/视频功能。以下是 1.5 系列通过 SSE 访问时支持的功能:

模型:

  • gemini-1.5-pro
  • gemini-1.5-flash

支持:

  • ✅ 文本输入/输出 (response_modalities=["TEXT"])
  • ✅ SSE 流式传输 (StreamingMode.SSE)
  • ✅ 具有自动执行功能的函数调用
  • ✅ 大上下文窗口(1.5-pro 高达 2M 令牌)

不支持:

  • ❌ 实时音频功能(音频 I/O、转录、VAD)
  • ❌ 通过 run_live() 的双向流式处理
  • ❌ 主动性和情感对话
  • ❌ 视频输入

了解 Live API 连接和会话

构建 ADK 双向流式应用程序时,了解 ADK 如何管理自身与 Live API 后端之间的通信层至关重要。本节探讨 连接(ADK 建立到 Live API 的 WebSocket 传输链路)和 会话(Live API 维护的逻辑对话上下文)之间的根本区别。与传统的请求-响应 API 不同,双向流式架构引入了独特的约束:连接超时、因模态而异的会话持续时间限制(仅音频 vs 音频+视频)、有限的上下文窗口,以及 Gemini Live API 和 Vertex AI Live API 之间不同的并发会话配额。

ADK Session 与 Live API 会话

理解 ADK SessionLive API 会话 之间的区别对于使用 ADK 双向流式处理构建可靠的流式应用程序至关重要。

ADK Session(由 SessionService 管理): - 用于对话历史记录、事件和状态的持久对话存储,通过 SessionService.create_session() 创建 - 存储选项:内存、数据库 (PostgreSQL/MySQL/SQLite) 或 Vertex AI - 跨多个 run_live() 调用和应用程序重启存活(使用持久 SessionService

Live API 会话(由 Live API 后端管理): - 在 run_live() 事件循环运行时由 Live API 维护,并在通过调用 LiveRequestQueue.close() 结束流式传输时销毁 - 受平台持续时间限制,并且可以使用会话恢复句柄跨多个连接恢复(见下文 ADK 如何管理会话恢复

它们如何协同工作:

  1. 当调用 run_live() 时:
  2. SessionService 检索 ADK Session
  3. 使用来自 session.events 的对话历史记录初始化 Live API 会话
  4. 与 Live API 后端双向流式传输事件
  5. 随着新事件的发生更新 ADK Session
  6. run_live() 结束时
  7. Live API 会话终止
  8. ADK Session 持久存在
  9. 当再次调用 run_live()应用程序重新启动时
    • ADK 从 ADK Session 加载历史记录
    • 使用该上下文创建一个新的 Live API 会话

简而言之,ADK Session 提供持久的长期对话存储,而 Live API 会话是短暂的流式上下文。这种分离使生产应用程序能够跨网络中断、应用程序重启和多个流式会话保持对话连续性。

下图说明了 ADK 会话持久性和短暂 Live API 会话上下文之间的关系,显示了如何跨多个 run_live() 调用维护对话历史记录:

sequenceDiagram
    participant App as 你的应用程序
    participant SS as SessionService
    participant ADK_Session as ADK 会话<br/>(持久存储)
    participant ADK as ADK (run_live)
    participant LiveSession as Live API 会话<br/>(短暂)

    Note over App,LiveSession: "第一次 run_live() 调用"

    App->>SS: get_session(user_id, session_id)
    SS->>ADK_Session: "加载会话数据"
    ADK_Session-->>SS: "包含事件历史的会话"
    SS-->>App: "会话对象"

    App->>ADK: runner.run_live(...)
    ADK->>LiveSession: "使用来自 ADK 会话的历史记录初始化"
    activate LiveSession

    Note over ADK,LiveSession: "双向流式传输..."

    ADK->>ADK_Session: "使用新事件更新"

    App->>ADK: queue.close()
    ADK->>LiveSession: "终止"
    deactivate LiveSession
    Note over LiveSession: "Live API 会话已销毁"
    Note over ADK_Session: "ADK 会话持久存在"

    Note over App,LiveSession: "第二次 run_live() 调用(或重启后)"

    App->>SS: get_session(user_id, session_id)
    SS->>ADK_Session: "加载会话数据"
    ADK_Session-->>SS: "包含事件历史的会话"
    SS-->>App: "会话对象(包含以前的历史记录)"

    App->>ADK: runner.run_live(...)
    ADK->>LiveSession: "使用完整历史记录初始化新会话"
    activate LiveSession

    Note over ADK,LiveSession: "双向流式传输继续..."

关键见解:

  • ADK 会话跨多个 run_live() 调用和应用程序重启存活
  • Live API 会话是短暂的 - 每个流式会话创建和销毁
  • 通过 ADK 会话的持久存储维护对话连续性
  • SessionService 管理持久层(内存、数据库或 Vertex AI)

既然我们了解了 ADK Session 对象和 Live API 会话之间的区别,让我们关注 Live API 连接和会话——支持实时双向流式处理的后端基础设施。

Live API 连接和会话

在 Live API 层面理解 连接会话 之间的区别对于构建可靠的 ADK 双向流式应用程序至关重要。

连接:ADK 和 Live API 服务器之间的物理 WebSocket 链路。这是承载双向流式数据的网络传输层。

会话:Live API 维护的逻辑对话上下文,包括对话历史记录、工具调用状态和模型上下文。一个会话可以跨越多个连接。

方面 连接 会话
是什么? WebSocket 网络连接 逻辑对话上下文
范围 传输层 应用层
可以跨越? 单个网络链路 通过恢复跨越多个连接
故障影响 网络错误或超时 丢失对话历史记录

各平台的 Live API 连接和会话限制

了解每个平台的约束对于生产规划至关重要。Gemini Live API 和 Vertex AI Live API 有不同的限制,影响对话可以运行多长时间以及多少用户可以同时连接。最重要的区别是 连接持续时间(单个 WebSocket 连接保持打开的时间)和 会话持续时间(逻辑对话可以持续的时间)。

约束类型 Gemini Live API
(Google AI Studio)
Vertex AI Live API
(Google Cloud)
备注
连接持续时间 ~10 分钟 未单独记录 每个 Gemini WebSocket 连接自动终止;ADK 通过会话恢复透明地重新连接
会话持续时间(仅音频) 15 分钟 10 分钟 无上下文窗口压缩的最大会话持续时间。两个平台:启用上下文窗口压缩后无限制
会话持续时间(音频 + 视频) 2 分钟 10 分钟 Gemini 对视频有更短的限制;Vertex 平等对待所有会话。两个平台:启用上下文窗口压缩后无限制
并发会话 50 (Tier 1)
1,000 (Tier 2+)
高达 1,000 Gemini 限制因 API 层级而异;Vertex 限制是每个 Google Cloud 项目

Live API 会话恢复

默认情况下,Live API 将连接持续时间限制为大约 10 分钟——每个 WebSocket 连接在此持续时间后自动关闭。为了克服此限制并启用更长时间的对话,Live API 提供了 会话恢复,这是一项透明地跨多个连接迁移会话的功能。启用后,Live API 生成恢复句柄,允许重新连接到相同的会话上下文,保留完整的对话历史记录和状态。

ADK 完全自动化了这一点:当你在 RunConfig 中启用会话恢复时,ADK 会自动处理所有重新连接逻辑——检测连接关闭、缓存恢复句柄并在后台无缝重新连接。你不需要编写任何重新连接代码。会话无缝地持续超过 10 分钟的连接限制,自动处理连接超时、网络中断和计划的重新连接。

ADK 重新连接管理的范围

ADK 管理 ADK 到 Live API 的连接(ADK 和 Gemini/Vertex Live API 后端之间的 WebSocket)。这对你的应用程序代码是透明的。

你的应用程序仍然负责

  • 管理到你的应用程序的客户端连接(例如,用户的 WebSocket 到你的 FastAPI 服务器)
  • 如果需要,实施客户端重新连接逻辑
  • 处理客户端和你的应用程序之间的网络故障

当 ADK 重新连接到 Live API 时,你的应用程序的事件循环继续正常运行——你继续从 run_live() 接收事件而不会中断。从你的应用程序的角度来看,Live API 会话无缝继续。

配置:

from google.genai import types

run_config = RunConfig(
    session_resumption=types.SessionResumptionConfig()
)

何时不启用会话恢复:

虽然建议对大多数生产应用程序使用会话恢复,但在以下情况下你可能不需要它:

  • 短会话(<10 分钟):如果你的会话通常在 ~10 分钟连接超时内完成,恢复会增加不必要的开销
  • 无状态交互:每个回合都独立的请求-响应式交互不会从会话连续性中受益
  • 开发/测试:当每个会话重新开始而不带过状态时,调试更简单
  • 成本敏感型部署:会话恢复可能会产生额外的平台成本或资源使用(请与你的平台核实)

最佳实践:默认情况下为生产启用会话恢复,仅当你有特定理由不使用它时才禁用。

ADK 如何管理会话恢复

虽然 Gemini Live API 和 Vertex AI Live API 都支持会话恢复,但直接使用它需要管理恢复句柄、检测连接关闭并实施重新连接逻辑。ADK 承担了这一复杂性的全部责任,在幕后自动利用会话恢复,因此开发人员无需编写任何重新连接代码。你只需在 RunConfig 中启用它,ADK 就会透明地处理一切。

ADK 的自动管理:

  1. 初始连接:ADK 建立到 Live API 的 WebSocket 连接
  2. 句柄更新:在整个会话期间,Live API 发送包含更新句柄的 session_resumption_update 消息。ADK 自动将最新句柄缓存在 InvocationContext.live_session_resumption_handle
  3. 优雅连接关闭:当达到 ~10 分钟连接限制时,WebSocket 优雅关闭(无异常)
  4. 自动重新连接:ADK 的内部循环检测到关闭并使用最近缓存的句柄自动重新连接
  5. 会话继续:同一会话无缝继续,保留完整上下文

实现细节

在重新连接期间,ADK 从 InvocationContext.live_session_resumption_handle 检索缓存的句柄,并将其包含在 live.connect() 调用的新 LiveConnectConfig 中。这完全由 ADK 的内部重新连接循环处理——开发人员无需直接访问或管理这些句柄。

序列图:自动重新连接

下面的序列图说明了当达到 ~10 分钟连接超时时,ADK 如何自动管理 Live API 会话恢复。ADK 检测到优雅关闭,检索缓存的恢复句柄,并在无需更改应用程序代码的情况下透明地重新连接:

sequenceDiagram
    participant App as 你的应用程序
    participant ADK as ADK (run_live)
    participant WS as WebSocket 连接
    participant API as Live API (Gemini/Vertex AI)
    participant LiveSession as Live 会话上下文

    Note over App,LiveSession: "初始连接(启用会话恢复)"

    App->>ADK: runner.run_live(run_config=RunConfig(session_resumption=...))
    ADK->>API: WebSocket connect()
    activate WS
    API->>LiveSession: "创建新会话"
    activate LiveSession

    Note over ADK,API: "双向流式传输(0-10 分钟)"

    App->>ADK: send_content(text) / send_realtime(audio)
    ADK->>API: "→ 内容(通过 WebSocket)"
    API->>LiveSession: "更新对话历史记录"
    API-->>ADK: "← 流式响应"
    ADK-->>App: "← yield event"

    Note over API,LiveSession: "Live API 发送恢复句柄更新"
    API-->>ADK: session_resumption_update { new_handle: "abc123" }
    ADK->>ADK: "在 InvocationContext 中缓存句柄"

    Note over WS,API: "~10 分钟已过 - 连接超时"

    API->>WS: "关闭 WebSocket(优雅关闭)"
    deactivate WS
    Note over LiveSession: "会话上下文保留"

    Note over ADK: "检测到优雅关闭 - 未引发异常"
    ADK->>ADK: "while True 循环继续"

    Note over ADK,API: "自动重新连接"

    ADK->>API: WebSocket connect(session_resumption.handle="abc123")
    activate WS
    API->>LiveSession: "附加到现有会话"
    API-->>ADK: "会话恢复,具有完整上下文"

    Note over ADK,API: "双向流式传输继续"

    App->>ADK: send_content(text) / send_realtime(audio)
    ADK->>API: "→ 内容(通过 WebSocket)"
    API->>LiveSession: "更新对话历史记录"
    API-->>ADK: "← 流式响应"
    ADK-->>App: "← yield event"

    Note over App,LiveSession: "会话继续直到持续时间限制或显式关闭"

    deactivate WS
    deactivate LiveSession

事件和会话持久性

有关哪些事件保存到 ADK Session 与哪些仅在流式传输期间产生的详细信息,请参阅 第 3 部分:保存到 ADK 会话的事件

Live API 上下文窗口压缩

问题: Live API 会话面临两个限制对话持续时间的关键约束。首先,会话持续时间限制 施加了硬性时间上限:如果不进行压缩,Gemini Live API 将仅音频会话限制为 15 分钟,音频+视频会话仅为 2 分钟,而 Vertex AI 将所有会话限制为 10 分钟。其次,上下文窗口限制 限制了对话长度:模型具有有限的令牌容量(gemini-2.5-flash-native-audio-preview-09-2025 为 128k 令牌,Vertex AI 模型为 32k-128k)。长时间的对话——尤其是扩展的客户支持会话、辅导互动或多小时的语音对话——将达到时间限制或令牌限制,导致会话终止或丢失关键的对话历史记录。

解决方案: 上下文窗口压缩 同时解决了这两个约束。它使用滑动窗口方法,在令牌计数达到配置的阈值时自动压缩或总结早期的对话历史记录。Live API 保留最近上下文的全部细节,同时压缩较旧的部分。关键是,启用上下文窗口压缩将会话持续时间延长至无限时间,移除了会话持续时间限制(Gemini Live API 上的 15 分钟仅音频/2 分钟音频+视频;Vertex AI 上所有会话的 10 分钟),同时也防止了令牌限制耗尽。但是,这有一个权衡:由于该功能总结了早期的对话历史记录而不是保留全部,过去上下文的细节将随着时间的推移逐渐丢失。模型将能够访问较旧交换的压缩摘要,而不是完整的逐字历史记录。

平台行为和官方限制

会话持续时间管理和上下文窗口压缩是 Live API 平台功能。ADK 通过 RunConfig 配置这些功能并将配置传递给 Live API,但实际的执行和实现由 Gemini/Vertex AI Live API 后端处理。

重要:本指南中提到的持续时间限制和“无限”会话行为基于当前的 Live API 行为。这些限制可能会由 Google 更改。请始终在官方文档中验证当前的会话持续时间限制和压缩行为:

ADK 提供了一种通过 RunConfig 配置上下文窗口压缩的简单方法。但是,开发人员负责根据其特定要求——模型上下文窗口大小、预期的对话模式和质量需求——适当地配置压缩参数(trigger_tokenstarget_tokens):

from google.genai import types
from google.adk.agents.run_config import RunConfig

# 对于 gemini-2.5-flash-native-audio-preview-09-2025(128k 上下文窗口)
run_config = RunConfig(
    context_window_compression=types.ContextWindowCompressionConfig(
        trigger_tokens=100000,  # 在 ~78% 的 128k 上下文处开始压缩
        sliding_window=types.SlidingWindow(
            target_tokens=80000  # 压缩至 ~62% 的上下文,保留最近的回合
        )
    )
)

# 对于 gemini-live-2.5-flash(Vertex AI 上的 32k 上下文窗口)
run_config = RunConfig(
    context_window_compression=types.ContextWindowCompressionConfig(
        trigger_tokens=25000,  # 在 ~78% 的 32k 上下文处开始压缩
        sliding_window=types.SlidingWindow(
            target_tokens=20000  # 压缩至 ~62% 的上下文
        )
    )
)

它是如何工作的:

当启用上下文窗口压缩时:

  1. Live API 监控对话上下文的总令牌计数
  2. 当上下文达到 trigger_tokens 阈值时,压缩激活
  3. 早期的对话历史记录使用滑动窗口方法进行压缩或总结
  4. 最近的上下文(最后 target_tokens 值)保留全部细节
  5. 两个关键效应同时发生:
  6. 会话持续时间限制被移除(Gemini Live API 上不再有 15 分钟/2 分钟上限,Vertex AI 上不再有 10 分钟上限)
  7. 令牌限制得到管理(无论对话长度如何,会话都可以无限期继续)

选择合适的阈值:

  • trigger_tokens 设置为模型上下文窗口的 70-80% 以留出余量
  • target_tokens 设置为 60-70% 以提供足够的压缩
  • 使用实际的对话模式进行测试以优化这些值

参数选择策略:

上面的示例使用 78% 作为 trigger_tokens,62% 作为 target_tokens。理由如下:

  1. trigger_tokens 为 78%:在达到硬性限制之前提供缓冲
  2. 为当前回合完成留出空间
  3. 防止响应中途的压缩中断
  4. 典型的对话可以继续进行几个回合

  5. target_tokens 为 62%:压缩后留出大量空间

  6. 每次压缩释放 16 个百分点 (78% - 62%)
  7. 允许在下一次压缩之前进行多个回合
  8. 平衡上下文保留与压缩频率

  9. 根据你的用例进行调整

  10. 长回合(详细的技术讨论):增加缓冲 → 70% 触发,50% 目标
  11. 短回合(快速问答):更紧的边距 → 85% 触发,70% 目标
  12. 上下文关键(需要历史细节):更高的目标 → 80% 触发,70% 目标
  13. 性能敏感(最小化压缩开销):更低的触发 → 70% 触发,50% 目标

始终使用实际的对话模式进行测试以找到最佳值。

何时不使用上下文窗口压缩

虽然压缩启用了无限的会话持续时间,但请考虑以下权衡:

上下文窗口压缩的权衡:

方面 有压缩 无压缩 最适合
会话持续时间 无限 15 分钟(音频)
2 分钟(视频)Gemini
10 分钟 Vertex
压缩:长会话
无压缩:短会话
上下文质量 较旧的上下文被总结 完整的逐字历史记录 压缩:一般对话
无压缩:精度关键
延迟 压缩开销 无开销 压缩:异步场景
无压缩:实时
内存使用 有界 随会话增长 压缩:长会话
无压缩:短会话
实现 配置阈值 无配置 压缩:生产
无压缩:原型

常见用例:

在以下情况下启用压缩:

  • 会话需要超过平台持续时间限制(15/2/10 分钟)
  • 扩展对话可能会达到令牌限制(2.5-flash 为 128k)
  • 可能持续数小时的客户支持会话
  • 具有长时间互动的教育辅导

在以下情况下禁用压缩:

  • 所有会话都在持续时间限制内完成
  • 早期对话的精确回忆至关重要
  • 开发/测试阶段(完整的历史记录有助于调试)
  • 总结导致的质量下降是不可接受的

最佳实践:仅当你需要比平台持续时间限制更长的会话或当对话可能超过上下文窗口令牌限制时才启用压缩。

Live API 连接和会话管理的最佳实践

基本:启用会话恢复

  • 始终为生产应用程序启用会话恢复 在 RunConfig 中
  • ✅ 这使 ADK 能够透明地自动处理 Gemini 的 ~10 分钟连接超时
  • ✅ 会话跨多个 WebSocket 连接无缝继续,无需用户中断
  • ✅ 会话恢复句柄缓存和管理
from google.genai import types

run_config = RunConfig(
    response_modalities=["AUDIO"],
    session_resumption=types.SessionResumptionConfig()
)
  • 启用上下文窗口压缩 如果你需要超过 15 分钟(仅音频)或 2 分钟(音频+视频)的会话
  • ✅ 一旦启用,会话持续时间变为无限——无需监控基于时间的限制
  • ✅ 根据你的模型的上下文窗口配置 trigger_tokenstarget_tokens
  • ✅ 使用现实的对话模式测试压缩设置
  • ⚠️ 明智地使用:压缩在总结期间会增加延迟,并可能丢失对话细微差别——仅当扩展会话对你的用例确实必要时才启用
from google.genai import types
from google.adk.agents.run_config import RunConfig

run_config = RunConfig(
    response_modalities=["AUDIO"],
    session_resumption=types.SessionResumptionConfig(),
    context_window_compression=types.ContextWindowCompressionConfig(
        trigger_tokens=100000,
        sliding_window=types.SlidingWindow(target_tokens=80000)
    )
)

可选:监控会话持续时间

仅适用于未使用上下文窗口压缩的情况:

  • ✅ 关注 会话持续时间限制,而不是连接超时(ADK 自动处理这些)
  • Gemini Live API:监控 15 分钟限制(仅音频)或 2 分钟限制(音频+视频)
  • Vertex AI Live API:监控 10 分钟会话限制
  • ✅ 在会话持续时间限制前 1-2 分钟警告用户
  • ✅ 为超过会话限制的对话实施优雅的会话转换

并发 Live API 会话和配额管理

问题: 生产语音应用程序通常同时为多个用户提供服务,每个用户都需要自己的 Live API 会话。但是,Gemini Live API 和 Vertex AI Live API 都施加了严格的并发会话限制,这些限制因平台和定价层级而异。如果没有适当的配额规划和会话管理,应用程序可能会很快达到这些限制,导致新用户的连接失败或高峰使用期间的服务质量下降。

解决方案: 了解特定于平台的配额,设计你的架构以保持在并发会话限制内,在需要时实施会话池或排队策略,并主动监控配额使用情况。ADK 自动处理单个会话生命周期,但开发人员必须构建其应用程序以在配额约束内管理多个并发用户。

了解并发 Live API 会话配额

这两个平台都限制了可以同时运行多少个 Live API 会话,但限制和机制有很大不同:

Gemini Live API (Google AI Studio) - 基于层级的配额:

层级 并发会话 TPM(每分钟令牌数) 访问权限
免费层 有限* 1,000,000 免费 API 密钥
第 1 层 50 4,000,000 即用即付
第 2 层 1,000 10,000,000 更高的使用层级
第 3 层 1,000 10,000,000 更高的使用层级

*免费层并发会话限制未明确记录,但明显低于付费层。

Vertex AI Live API (Google Cloud) - 基于项目的配额:

资源类型 限制 范围
并发实时双向连接 每分钟 10 个 每个项目,每个区域
最大并发会话 高达 1,000 每个项目
会话创建/删除/更新 每分钟 100 个 每个项目,每个区域

请求增加配额:

要请求增加 Live API 并发会话,请导航到 Google Cloud Console 中的 配额页面。过滤名为 "Bidi generate content concurrent requests" 的配额,以查找每个项目、区域和基础模型的配额值,并提交配额增加请求。你需要配额管理员角色 (roles/servicemanagement.quotaAdmin) 才能提出请求。有关详细说明,请参阅 查看和管理配额

Cloud Console 上的配额值

主要区别:

  1. Gemini Live API:并发会话限制随 API 层级急剧扩展(50 → 1,000 个会话)。最适合具有不可预测或快速扩展的用户群且愿意支付更高层级费用的应用程序。

  2. Vertex AI Live API:受连接建立速率(10/分钟)限制,但支持高达 1,000 个总并发会话。最适合具有逐步扩展模式和现有 Google Cloud 基础设施的企业应用程序。此外,你可以请求增加配额,以便为具有更高并发要求的生产部署做准备。

管理配额的架构模式

一旦你了解了并发会话配额,下一个挑战就是构建你的应用程序以在这些限制内有效运行。正确的方法取决于你的预期用户并发性、扩展要求和对排队的容忍度。本节介绍了两种架构模式——从用于低并发应用程序的简单直接映射到用于可能在高峰使用期间超过配额限制的应用程序的带排队的会话池。选择符合你当前规模的模式,并将其设计为随着用户群的增长而演变。

选择正确的架构:

                开始:设计配额管理
                              |
                              v
                   预期并发用户?
                     /                    \
            < 配额限制           > 配额限制或不可预测
                   |                              |
                   v                              v
          模式 1:直接映射    模式 2:会话池
          - 简单的 1:1 映射         - 排队等待用户
          - 无配额逻辑             - 优雅降级
          - 快速开发               - 高峰处理
                   |                              |
                   v                              v
              适合:                      适合:
              - 原型                       - 大规模生产
              - 小团队                     - 不可预测的负载
              - 受控用户                   - 公共应用程序

快速决策指南:

因素 直接映射 会话池
预期用户 始终 < 配额 可能超过配额
用户体验 始终即时 高峰期可能等待
实现复杂性
运营开销 监控队列深度
最适合 原型,内部工具 生产,公共应用

模式 1:直接映射(简单应用程序)

对于并发用户永远不会超过配额限制的小规模应用程序,为每个连接的用户创建一个专用的 Live API 会话,并进行简单的 1:1 映射:

  1. 当用户连接时: 立即为他们启动一个 run_live() 会话
  2. 当他们断开连接时: 会话结束
  3. 无配额管理逻辑: 假设你的总并发用户将始终低于你的配额限制

这是最简单的架构,非常适合原型、开发环境和具有可预测用户负载的小规模应用程序。

模式 2:带有排队的会话池

对于在高峰使用期间可能超过并发会话限制的应用程序,跟踪活动 Live API 会话的数量并在应用程序级别强制执行你的配额限制:

  1. 当新用户连接时: 检查是否有可用的会话槽
  2. 如果有槽可用: 立即启动会话
  3. 如果你已达到配额限制:
  4. 将用户放入等待队列
  5. 通知他们正在等待可用槽
  6. 随着会话结束: 自动处理队列以为等待用户启动会话

这提供了优雅降级——用户在高峰时段短暂等待,而不是经历硬连接故障。

其他控制

support_cfc(实验性)

此参数启用组合函数调用 (CFC),允许模型以复杂的模式编排多个工具——并行调用工具,将输出链接为其他工具的输入,或根据中间结果有条件地执行工具。

⚠️ 实验性功能: CFC 支持是实验性的,可能会发生变化。

关键行为:support_cfc=True 时,ADK 始终在内部使用 Live API (WebSocket),无论 streaming_mode 设置如何。这是因为只有 Live API 后端支持 CFC 功能。

# 即使使用 SSE 模式,当启用 CFC 时,ADK 也会通过 Live API 进行路由
run_config = RunConfig(
    support_cfc=True,
    streaming_mode=StreamingMode.SSE  # ADK 在内部使用 Live API
)

模型要求:

ADK 在会话初始化时验证 CFC 兼容性,如果模型不受支持,将引发错误:

  • 支持gemini-2.x 模型(例如,gemini-2.5-flash-native-audio-preview-09-2025, gemini-2.0-flash-live-001
  • 不支持gemini-1.5-x 模型
  • 验证:当 support_cfc=True 时,ADK 检查模型名称是否以 gemini-2 开头 (runners.py:1200-1203)
  • 代码执行器:当启用 CFC 时,ADK 自动注入 BuiltInCodeExecutor 以进行安全的并行工具执行

CFC 功能:

  • 并行执行:同时调用多个独立工具(例如,一次获取多个城市的天气)
  • 函数链:将一个工具的输出用作另一个工具的输入(例如,get_location()get_weather(location)
  • 条件执行:根据先前工具调用的中间结果执行工具

用例:

CFC 专为受益于智能工具编排的复杂多步工作流而设计:

  • 同时从多个 API 聚合数据
  • 工具相互馈送的多步分析流水线
  • 需要条件探索的复杂研究任务
  • 任何需要超越顺序执行的复杂工具协调的场景

对于双向流式应用程序: 虽然 CFC 适用于 BIDI 模式,但它主要针对基于文本的工具编排进行了优化。对于实时音频/视频交互(本指南的重点),标准函数调用通常提供更好的性能和更简单的实现。

了解更多:

max_llm_calls

此参数限制每个调用上下文允许的总 LLM 调用次数,提供针对失控成本和无限智能体循环的保护。

BIDI 流式处理的限制:

max_llm_calls 限制不适用于具有 StreamingMode.BIDIrun_live() 此参数仅保护 SSE 流式模式和 run_async() 流程。如果你正在构建双向流式应用程序(本指南的重点),你将不会从此参数获得自动成本保护。

对于实时流式会话,实施你自己的保障措施:

  • 会话持续时间限制
  • 回合计数跟踪
  • 通过跟踪模型回合事件中的令牌使用情况进行自定义成本监控(见 第 3 部分:事件类型和处理
  • 应用程序级断路器

save_live_blob

此参数控制是否将音频和视频流持久化到 ADK 的会话和工件服务,以用于调试、合规性和质量保证目的。

目前,ADK 的实现仅持久化音频。启用后,ADK 将音频流持久化到:

用例:

  • 调试:语音交互问题,助手行为分析
  • 合规性:受监管行业(医疗保健、金融服务)的审计跟踪
  • 质量保证:监控对话质量,识别问题
  • 训练数据:收集数据以进行模型改进
  • 开发/测试:测试环境和成本敏感型部署

存储注意事项:

启用 save_live_blob=True 具有显著的存储影响:

  • 音频文件大小:在 16kHz PCM 下,音频输入每分钟产生 ~1.92 MB
  • 会话存储:音频存储在会话服务和工件服务中
  • 保留策略:检查你的工件服务配置以了解保留期
  • 成本影响:对于大容量语音应用程序,存储成本可能会迅速累积

最佳实践:

  • 仅在需要时启用(调试、合规性、训练)
  • 实施保留策略以自动删除旧的音频工件
  • 考虑采样(例如,保存 10% 的会话以进行质量监控)
  • 如果你的工件服务支持,请使用压缩

custom_metadata

此参数允许你将任意键值元数据附加到当前调用期间生成的事件。元数据存储在 Event.custom_metadata 字段中并持久化到会话存储,使你能够使用特定于应用程序的上下文标记事件,以进行分析、调试、路由或合规性跟踪。

配置:

from google.adk.agents.run_config import RunConfig

# 将元数据附加到此调用中的所有事件
run_config = RunConfig(
    custom_metadata={
        "user_tier": "premium",
        "session_type": "customer_support",
        "campaign_id": "promo_2025",
        "ab_test_variant": "variant_b"
    }
)

它是如何工作的:

当你在 RunConfig 中提供 custom_metadata 时:

  1. 元数据附加:字典附加到调用期间生成的每个 Event
  2. 会话持久性:带有元数据的事件存储在会话服务(数据库、Vertex AI 或内存)中
  3. 事件访问:通过 event.custom_metadata 从任何事件检索元数据
  4. A2A 集成:对于智能体到智能体 (A2A) 通信,ADK 自动将 A2A 请求元数据传播到此字段

类型规范:

custom_metadata: Optional[dict[str, Any]] = None

元数据是一个灵活的字典,接受任何 JSON 可序列化的值(字符串、数字、布尔值、嵌套对象、数组)。

用例:

  • 用户细分:使用用户层级、订阅级别或群组信息标记事件
  • 会话分类:按类型(支持、销售、入职)标记会话以进行分析
  • 活动跟踪:将事件与营销活动或实验相关联
  • A/B 测试:跟踪应用程序的哪个变体生成了事件
  • 合规性:附加管辖区、同意标志或数据保留策略
  • 调试:添加跟踪 ID、功能标志或环境标识符
  • 分析:存储自定义维度以进行下游分析

示例 - 从事件中检索元数据:

async for event in runner.run_live(
    session=session,
    live_request_queue=queue,
    run_config=RunConfig(
        custom_metadata={"user_id": "user_123", "experiment": "new_ui"}
    )
):
    if event.custom_metadata:
        print(f"User: {event.custom_metadata.get('user_id')}")
        print(f"Experiment: {event.custom_metadata.get('experiment')}")

智能体到智能体 (A2A) 集成:

当使用 RemoteA2AAgent 时,ADK 自动从 A2A 请求中提取元数据并填充 custom_metadata

# A2A 请求元数据自动映射到 custom_metadata
# 来源:a2a/converters/request_converter.py
custom_metadata = {
    "a2a_metadata": {
        # 原始 A2A 请求元数据出现在这里
    }
}

模型要求:

ADK 在会话初始化时验证 CFC 兼容性,如果模型不受支持,将引发错误:

  • 支持gemini-2.x 模型(例如,gemini-2.5-flash-native-audio-preview-09-2025, gemini-2.0-flash-live-001
  • 不支持gemini-1.5-x 模型
  • 验证:当 support_cfc=True 时,ADK 检查模型名称是否以 gemini-2 开头 (runners.py:1200-1203)
  • 代码执行器:当启用 CFC 时,ADK 自动注入 BuiltInCodeExecutor 以进行安全的并行工具执行

CFC 功能:

  • 并行执行:同时调用多个独立工具(例如,一次获取多个城市的天气)
  • 函数链:将一个工具的输出用作另一个工具的输入(例如,get_location()get_weather(location)
  • 条件执行:根据先前工具调用的中间结果执行工具

用例:

CFC 专为受益于智能工具编排的复杂多步工作流而设计:

  • 同时从多个 API 聚合数据
  • 工具相互馈送的多步分析流水线
  • 需要条件探索的复杂研究任务
  • 任何需要超越顺序执行的复杂工具协调的场景

对于双向流式应用程序: 虽然 CFC 适用于 BIDI 模式,但它主要针对基于文本的工具编排进行了优化。对于实时音频/视频交互(本指南的重点),标准函数调用通常提供更好的性能和更简单的实现。

了解更多:

总结

在本部分中,你学习了 RunConfig 如何通过声明性配置对 ADK 双向流式会话进行复杂的控制。我们涵盖了响应模态及其约束,探讨了 BIDI 和 SSE 流式模式之间的差异,检查了 ADK 会话与 Live API 会话之间的关系,并学习了如何通过会话恢复和上下文窗口压缩来管理会话持续时间。你现在了解了如何处理并发会话配额,实施配额管理的架构模式,通过 max_llm_calls 和音频持久性选项配置成本控制。掌握了 RunConfig,你可以构建平衡功能丰富性与操作约束的生产就绪流式应用程序——启用扩展对话,管理平台限制,有效控制成本,并监控资源消耗。


上一篇:第 3 部分 - 使用 run_live() 处理事件 | 下一篇:第 5 部分 - 如何使用音频、图像和视频