第 1 部分：ADK Gemini Live API Toolkit 简介¶

Google 的 Agent Development Kit (ADK) 提供了一个生产就绪的框架，用于构建与 Gemini 模型的双向流式应用。本指南介绍 ADK 的流式架构，该架构通过多模态渠道（文本、音频、视频）实现用户与 AI 智能体之间的实时双向通信。

你将学到什么：本部分涵盖双向流式的基础知识、底层的 Live API 技术（Gemini Live API 和 Gemini Live API (Agent Platform)）、ADK 的架构组件（LiveRequestQueue、Runner、Agent），以及一个完整的 FastAPI 实现示例。

ADK Gemini Live API Toolkit 演示¶

为了帮助你理解本指南中的概念，我们提供了一个可运行的演示应用，展示 ADK 双向流式通信的实际效果。这个基于 FastAPI 的演示实现了完整的流式生命周期，具有实用的、贴近真实场景的架构。

演示仓库：adk-samples/python/agents/bidi-demo

演示功能包括：

• WebSocket 通信：实时双向流式通信，支持并发的上行/下行任务
• 多模态请求：文本、音频和图像/视频输入，支持自动转录
• 灵活的响应：基于模型能力的文本或音频输出
• 交互式 UI：带有事件控制台的 Web 界面，用于监控 Live API 事件
• Google 搜索集成：配备工具调用能力的智能体

我们强烈建议在深入阅读本指南之前，先安装并运行此演示。动手实验将帮助你更深入地理解概念，演示代码在本指南的所有部分中都可作为实用参考。

有关安装说明和使用详情，请参阅演示 README。

1.1 什么是双向流式？¶

双向流式通信（Bidi-streaming）代表了传统 AI 交互方式的根本性转变。它摒弃了僵化的"提问并等待"模式，实现了实时双向通信，让人和 AI 能够同时说话、倾听和回应。这创造了自然的、类似人类的对话体验，具有即时响应能力，以及革命性地打断正在进行的交互的能力。

试想一下发送电子邮件和打电话之间的区别。传统的 AI 交互就像发电子邮件——你发送一条完整的消息，等待收到完整的回复，然后再发送另一条完整的消息。双向流式通信就像打电话——流畅、自然，可以随时打断、澄清和实时回应。

关键特性¶

这些特性将双向流式通信与传统 AI 交互区分开来，使其在创建引人入胜的用户体验方面具有独特优势：

• 双向通信：无需等待完整响应即可进行持续的数据交换。用户可以在 AI 响应过程中用新输入打断，创造自然的对话流程。AI 在检测到用户说完后（通过自动语音活动检测或显式活动信号）开始响应。

• 响应式打断：这可能是实现自然用户体验最重要的功能——用户可以在 AI 响应过程中用新输入打断，就像人类对话一样。如果 AI 正在解释量子物理学，而你突然问"等等，什么是电子？"，AI 会立即停下来回答你的问题。

• 最适合多模态：双向流式通信擅长多模态交互，因为它可以通过单个连接同时处理不同类型的输入。用户可以在展示文档的同时说话，在语音通话中输入后续问题，或在通信模式之间无缝切换，而不会丢失上下文。这种统一方法消除了为每种模态管理单独通道的复杂性。

sequenceDiagram
    participant Client as 用户
    participant Agent

    Client->>Agent: "你好！"
    Client->>Agent: "解释一下日本的历史"
    Agent->>Client: "你好！"
    Agent->>Client: "当然！日本的历史是……"（部分内容）
    Client->>Agent: "啊，等等。"

    Agent->>Client: "好的，有什么可以帮你的？" [interrupted: true]

与其他流式类型的差异¶

理解双向流式通信与其他方法的差异，对于认识其独特价值至关重要。流式技术领域包含几种不同的模式，每种服务于不同的用例：

实际应用场景¶

双向流式通信通过使智能体能够以类似人类的响应速度和智能水平运作，彻底改变了智能体 AI 应用。这些应用展示了流式技术如何将静态的 AI 交互转变为动态的、智能体驱动的体验，使其真正具有智能性和主动性。

在Shopper's Concierge 演示的视频中，多模态双向流式通信功能通过实现更快、更直观的购物体验，显著改善了电商用户体验。对话理解与快速并行搜索的结合，最终实现了虚拟试穿等高级功能，提升了买家信心并减少了在线购物的摩擦。

此外，双向流式通信还有许多可能的实际应用场景：

客服与联络中心¶

这是最直接的应用场景。该技术可以创建超越传统聊天机器人的高级虚拟客服。

• 用例：客户致电零售公司的客服热线，反馈产品缺陷问题。
• 多模态（视频）：客户可以说"我的咖啡机底部漏水了，我给你看一下。"然后用手机摄像头实时直播问题。AI 智能体可以利用视觉能力识别型号和具体故障点。
• 实时交互与打断：如果客服说"好的，我正在为您处理 Model X 咖啡机的退货"，客户可以打断说"不，等等，是 Model Y Pro"，智能体可以立即纠正方向，无需重新开始对话。

电商与个性化购物¶

智能体可以充当实时、互动的个人购物顾问，提升在线零售体验。

• 用例：用户正在浏览时尚网站，想要搭配建议。
• 多模态（语音与图像）：用户可以将一件衣服举到摄像头前并问"你能帮我找一条和这条裤子搭配的鞋子吗？"智能体会分析裤子的颜色和款式。
• 实时交互：对话可以流畅地来回进行："给我看更休闲一些的"……"好的，这双运动鞋怎么样？"……"太好了，把蓝色的 10 号加到我的购物车里。"

现场服务与技术支持¶

在现场工作的技术人员可以使用免提、语音激活的助手获得实时帮助。

• 用例：一名暖通空调技术人员正在现场诊断一台复杂的商用空调机组。
• 多模态（视频与语音）：技术人员佩戴智能眼镜或使用手机，可以将其第一视角实时传输给 AI 智能体。他们可以问"我听到这个压缩机有奇怪的声音。你能识别一下并调出这个型号的诊断流程图吗？"
• 实时交互：智能体可以逐步引导技术人员，技术人员可以在任何时刻提出澄清问题或打断，而无需放下手中的工具。

医疗保健与远程医疗¶

智能体可以作为患者就诊、分诊和基本咨询的第一接触点。

• 用例：患者使用医疗机构的应用程序进行关于皮肤状况的初步咨询。
• 多模态（视频/图像）：患者可以安全地分享皮疹的实时视频或高分辨率图像。AI 可以进行初步分析并提出澄清问题。

金融服务与财富管理¶

智能体可以为客户提供安全、互动且数据丰富的方式来管理财务。

• 用例：客户想要查看投资组合并讨论市场趋势。
• 多模态（屏幕共享）：智能体可以共享屏幕来显示图表和投资组合表现数据。客户也可以共享屏幕，指向某篇新闻文章并问"这个事件对我的科技股可能有什么影响？"
• 实时交互：通过访问客户账户数据分析当前投资组合配置，模拟潜在交易对投资组合风险状况的影响。

1.2 Gemini Live API¶

ADK Gemini Live API Toolkit 的能力由 Live API 技术提供支持，可通过两个平台获得：Gemini Live API（通过 Google AI Studio）和 Gemini Live API (Agent Platform)（通过 Google Cloud）。两者都提供与 Gemini 模型的实时、低延迟流式对话，但服务于不同的开发和部署需求。

在本指南中，我们使用 "Live API" 来统称这两个平台，仅在讨论平台特定功能或差异时才指明"Gemini Live API"或"Gemini Live API (Agent Platform)"。

什么是 Live API？¶

Live API 是 Google 的实时对话式 AI 技术，支持与 Gemini 模型进行低延迟双向流式通信。与传统的请求-响应 API 不同，Live API 建立持久的 WebSocket 连接，支持：

核心能力：

• 多模态流式通信：实时处理音频、视频和文本的连续流
• 语音活动检测（VAD）：自动检测用户何时说完，无需显式信号即可实现自然的轮替。AI 知道何时开始响应以及何时等待更多输入
• 即时响应：以最小延迟提供类似人类的语音或文本响应
• 智能打断：允许用户在 AI 响应过程中打断，就像人类对话一样
• 音频转录：实时转录用户输入和模型输出，无需单独的转录服务即可实现无障碍功能和对话日志记录
• 会话管理：长对话可以通过会话恢复跨越多个连接，API 在重新连接时保留完整的对话历史和上下文
• 工具集成：函数调用在流式模式下无缝工作，工具在后台执行的同时对话继续进行

原生音频模型特性：

• 主动音频：模型可以基于上下文感知主动发起响应，创造更自然的交互体验，AI 主动提供帮助或澄清（仅限原生音频模型）
• 情感对话：高级模型理解语调和情感上下文，根据对话氛围和用户情绪调整响应（仅限原生音频模型）

技术规格：

• 音频输入：16 位 PCM，16kHz（单声道）
• 音频输出：16 位 PCM，24kHz（原生音频模型）
• 视频输入：每秒 1 帧，建议 768x768 分辨率
• 上下文窗口：因模型而异（Live API 模型通常为 32k-128k token）。具体限制请参阅 Gemini 模型。
• 语言支持：支持 24+ 种语言，自动检测

Gemini Live API 与 Gemini Live API (Agent Platform) 对比¶

两个 API 提供相同的核心 Live API 技术，但在部署平台、身份验证和企业功能方面有所不同：

1.3 ADK Gemini Live API Toolkit¶

从零开始构建实时智能体应用面临重大工程挑战。虽然 Live API 提供了底层流式技术，但将其集成到生产应用中需要解决复杂的问题：管理 WebSocket 连接和重连逻辑、编排工具执行和响应处理、跨会话持久化对话状态、协调多模态输入的并发数据流，以及处理开发和生产环境之间的平台差异。

ADK 将这些挑战转化为简单的声明式 API。开发者无需花费数月时间构建会话管理、工具编排和状态持久化的基础设施，而是可以专注于定义智能体行为和创建用户体验。本节探讨 ADK 自动处理的内容，以及为什么它是构建生产级流式应用的推荐路径。

原始 Live API 与 ADK Gemini Live API Toolkit 对比：

平台灵活性¶

ADK 最强大的特性之一是对 Gemini Live API 和 Gemini Live API (Agent Platform) 的透明支持。这种平台灵活性实现了无缝的开发到生产工作流程：使用免费 API 密钥通过 Gemini API 在本地开发，然后使用企业级 Google Cloud 基础设施通过 Agent Platform 部署到生产环境——无需更改应用代码，只需更改环境配置。

平台选择的工作原理¶

ADK 使用 GOOGLE_GENAI_USE_ENTERPRISE 环境变量来确定使用哪个 Live API 平台：

• GOOGLE_GENAI_USE_ENTERPRISE=FALSE（或未设置）：通过 Google AI Studio 使用 Gemini Live API
• GOOGLE_GENAI_USE_ENTERPRISE=TRUE：通过 Google Cloud 使用 Gemini Live API (Agent Platform)

此环境变量由底层 google-genai SDK 在 ADK 创建 LLM 连接时读取。切换平台时无需更改代码——只需更改环境配置。

开发阶段：Gemini Live API (Google AI Studio)¶

# .env.development
GOOGLE_GENAI_USE_ENTERPRISE=FALSE
GOOGLE_API_KEY=your_api_key_here

优势：

• 使用 Google AI Studio 的免费 API 密钥进行快速原型开发
• 无需 Google Cloud 设置
• 即时实验流式功能
• 开发期间零基础设施成本

生产阶段：Gemini Live API (Agent Platform)¶

# .env.production
GOOGLE_GENAI_USE_ENTERPRISE=TRUE
GOOGLE_CLOUD_PROJECT=your_project_id
GOOGLE_CLOUD_LOCATION=us-central1

优势：

• 通过 Google Cloud 获得企业级基础设施
• 高级监控、日志记录和成本控制
• 与现有 Google Cloud 服务集成
• 生产级 SLA 和支持
• 无需更改代码——只需环境配置

通过处理会话管理、工具编排、状态持久化和平台差异的复杂性，ADK 让你可以专注于构建智能体体验，而不是与流式基础设施作斗争。相同的代码在开发和生产环境中无缝运行，让你无需承担实现负担即可获得双向流式通信的全部能力。

1.4 ADK Gemini Live API Toolkit 架构概览¶

既然你已了解 Live API 技术以及 ADK 为何增加价值，接下来让我们探索 ADK 的实际工作原理。本节映射了从应用经过 ADK 的流水线到 Live API 再返回的完整数据流，展示哪些组件负责哪些职责。

你将看到 LiveRequestQueue、Runner 和 Agent 等关键组件如何编排流式对话，而无需你管理 WebSocket 连接、协调异步流或处理平台特定的 API 差异。

高层架构¶

graph TB
    subgraph "应用"
        subgraph "客户端"
            C1["Web / 移动端"]
        end

        subgraph "传输层"
            T1["WebSocket / SSE（如 FastAPI）"]
        end
    end

    subgraph "ADK"
        subgraph "ADK Gemini Live API Toolkit"
            L1[LiveRequestQueue]
            L2[Runner]
            L3[Agent]
            L4[LLM Flow]
        end

        subgraph "LLM 集成"
            G1[GeminiLlmConnection]
            G2[Gemini Live API / Gemini Live API on Agent Platform]
        end
    end

    C1 <--> T1
    T1 -->|"live_request_queue.send()"| L1
    L1 -->|"runner.run_live(queue)"| L2
    L2 -->|"agent.run_live()"| L3
    L3 -->|"_llm_flow.run_live()"| L4
    L4 -->|"llm.connect()"| G1
    G1 <--> G2
    G1 -->|"yield LlmResponse"| L4
    L4 -->|"yield Event"| L3
    L3 -->|"yield Event"| L2
    L2 -->|"yield Event"| T1

    classDef external fill:#e1f5fe,stroke:#01579b,stroke-width:2px
    classDef adk fill:#f3e5f5,stroke:#4a148c,stroke-width:2px

    class C1,T1 external
    class L1,L2,L3,L4,G1,G2 adk

这种架构展示了 ADK 清晰的关注点分离：你的应用处理用户交互和传输协议，ADK 管理流式编排和状态，Live API 提供 AI 智能。通过抽象 LLM 端流式连接管理、事件循环和协议转换的复杂性，ADK 使你能够专注于构建智能体行为和用户体验，而不是流式基础设施。

1.5 ADK Gemini Live API Toolkit 应用生命周期¶

ADK Gemini Live API Toolkit 将 Live API 会话集成到 ADK 框架的应用生命周期中。这种集成创建了一个四阶段的生命周期，将 ADK 的智能体管理与 Live API 的实时流式能力相结合：

• 阶段 1：应用初始化（启动时执行一次）

• ADK 应用初始化

  • 创建 Agent：用于与用户交互、使用外部工具以及与其他智能体协调。
  • 创建 SessionService：用于获取或创建 ADK Session
  • 创建 Runner：为智能体提供运行时

• 阶段 2：会话初始化（每个用户会话执行一次）

• ADK Session 初始化：
  • 使用 SessionService 获取或创建 ADK Session

• ADK Gemini Live API Toolkit 初始化：

  • 创建 RunConfig 用于配置 ADK Gemini Live API Toolkit
  • 创建 LiveRequestQueue 用于向 Agent 发送用户消息
  • 启动 run_live() 事件循环

• 阶段 3：使用 run_live() 事件循环进行双向流式通信（每个用户会话执行一次或多次）

• 上行：用户通过 LiveRequestQueue 向智能体发送消息

• 下行：智能体通过 Event 向用户响应

• 阶段 4：终止 Live API 会话（每个用户会话执行一次或多次）

• LiveRequestQueue.close()

生命周期流程概览：

graph TD
    A[阶段 1：应用初始化<br/>启动时执行一次] --> B[阶段 2：会话初始化<br/>每个用户连接]
    B --> C[阶段 3：双向流式通信<br/>活跃通信]
    C --> D[阶段 4：终止<br/>关闭会话]
    D -.新连接.-> B

    style A fill:#e3f2fd
    style B fill:#e8f5e9
    style C fill:#fff3e0
    style D fill:#ffebee

此流程图展示了高层生命周期阶段及其连接方式。下面的详细时序图说明了每个阶段内的具体组件和交互。

sequenceDiagram
    participant Client
    participant App as 应用服务器
    participant Queue as LiveRequestQueue
    participant Runner
    participant Agent
    participant API as Live API

    rect rgb(230, 240, 255)
        Note over App: 阶段 1：应用初始化（启动时执行一次）
        App->>Agent: 1. 创建 Agent(model, tools, instruction)
        App->>App: 2. 创建 SessionService()
        App->>Runner: 3. 创建 Runner(app_name, agent, session_service)
    end

    rect rgb(240, 255, 240)
        Note over Client,API: 阶段 2：会话初始化（每次用户连接时）
        Client->>App: 1. WebSocket 连接(user_id, session_id)
        App->>App: 2. get_or_create_session(app_name, user_id, session_id)
        App->>App: 3. 创建 RunConfig(streaming_mode, modalities)
        App->>Queue: 4. 创建 LiveRequestQueue()
        App->>Runner: 5. 启动 run_live(user_id, session_id, queue, config)
        Runner->>API: 连接到 Live API 会话
    end

    rect rgb(255, 250, 240)
        Note over Client,API: 阶段 3：使用 run_live() 事件循环进行双向流式通信

        par 上行：用户通过 LiveRequestQueue 发送消息
            Client->>App: 用户消息（文本/音频/视频）
            App->>Queue: send_content() / send_realtime()
            Queue->>Runner: 缓冲的请求
            Runner->>Agent: 处理请求
            Agent->>API: 流式传输到 Live API
        and 下行：智能体通过 Event 响应
            API->>Agent: 流式响应
            Agent->>Runner: 处理响应
            Runner->>App: yield Event（文本/音频/工具/轮次）
            App->>Client: 通过 WebSocket 转发 Event
        end

        Note over Client,API: （事件循环持续运行直到收到关闭信号）
    end

    rect rgb(255, 240, 240)
        Note over Client,API: 阶段 4：终止 Live API 会话
        Client->>App: WebSocket 断开连接
        App->>Queue: close()
        Queue->>Runner: 关闭信号
        Runner->>API: 从 Live API 断开连接
        Runner->>App: run_live() 退出
    end

在以下各节中，你将看到每个阶段的详细介绍，展示何时创建每个组件以及它们如何协同工作。理解此生命周期模式对于构建能够高效处理多个并发会话的健壮流式应用至关重要。

阶段 1：应用初始化¶

这些组件在应用启动时创建一次，并在所有流式会话间共享。它们定义了智能体的能力、管理对话历史，并编排流式执行。

定义你的智能体¶

Agent 是流式应用的核心——它定义了你的 AI 能做什么、应该如何行为，以及由哪个 AI 模型驱动。你可以使用特定的模型、智能体可以使用的工具（如 Google Search 或自定义 API）以及塑造其个性和行为的指令来配置智能体。

"""ADK Gemini Live API Toolkit 演示的 Google Search 智能体定义。"""

import os
from google.adk.agents import Agent
from google.adk.tools import google_search

# 支持原生音频的 Live API 默认模型：
# - Gemini Live API：gemini-2.5-flash-native-audio-preview-12-2025
# - Gemini Live API (Agent Platform)：gemini-live-2.5-flash-native-audio
agent = Agent(
    name="google_search_agent",
    model=os.getenv("DEMO_AGENT_MODEL", "gemini-2.5-flash-native-audio-preview-12-2025"),
    tools=[google_search],
    instruction="You are a helpful assistant that can search the web."
)

智能体实例是无状态且可重用的——你创建一次，即可在所有流式会话中使用。智能体配置在 ADK 智能体文档中有详细说明。

定义你的 SessionService¶

ADK Session 跨流式会话管理对话状态和历史。它存储和检索会话数据，支持对话恢复和上下文持久化等功能。

要创建 Session 或获取指定 session_id 的现有会话，每个 ADK 应用都需要一个 SessionService。对于开发目的，ADK 提供了一个简单的 InMemorySessionService，当应用关闭时会丢失 Session 状态。

from google.adk.sessions import InMemorySessionService

# 定义你的会话服务
session_service = InMemorySessionService()

对于生产应用，请根据基础设施选择持久化会话服务：

使用 DatabaseSessionService 的情况：

• 你需要 SQLite、PostgreSQL 或 MySQL 的持久化存储
• 你正在构建单服务器应用（SQLite）或多服务器部署（PostgreSQL/MySQL）
• 你需要完全控制数据存储和备份
• 示例：
  • SQLite：DatabaseSessionService(db_url="sqlite:///./sessions.db")
  • PostgreSQL：DatabaseSessionService(db_url="postgresql://user:pass@host/db")

使用 VertexAiSessionService 的情况：

• 你已经在使用 Google Cloud Platform
• 你需要具有内置可扩展性的托管存储
• 你需要与 Agent Platform 功能紧密集成
• 示例：VertexAiSessionService(project="my-project")

两者都提供会话持久化能力——根据基础设施和规模需求进行选择。使用持久化会话服务时，Session 的状态将在应用关闭后得到保留。有关更多详情，请参阅 ADK 会话管理文档。

定义你的 Runner¶

Runner 为 Agent 提供运行时。它管理对话流程、协调工具执行、处理事件，并与会话存储集成。你在应用启动时创建一个 Runner 实例，并在所有流式会话中重用它。

from google.adk.runners import Runner

APP_NAME = "bidi-demo"

# 定义你的 runner
runner = Runner(
    app_name=APP_NAME,
    agent=agent,
    session_service=session_service
)

app_name 参数是必需的，用于在会话存储中标识你的应用。你应用的所有会话都组织在此名称下。

阶段 2：会话初始化¶

获取或创建会话¶

ADK Session 为 ADK Gemini Live API Toolkit 应用提供"对话线程"。就像你不会每次发短信都从头开始一样，智能体也需要关于当前交互的上下文。Session 是 ADK 专门设计用于跟踪和管理这些独立对话线程的对象。

ADK Session 与 Live API 会话¶

ADK Session（由 SessionService 管理）在多个双向流式会话之间提供持久化对话存储（可跨越数小时、数天甚至数月），而 Live API 会话（由 Live API 后端管理）是临时的流式上下文，仅存在于单个双向流式事件循环期间（通常跨越几分钟或几小时），我们稍后将讨论。当循环启动时，ADK 使用来自 ADK Session 的历史记录初始化 Live API 会话，然后在发生新事件时更新 ADK Session。

会话标识符由应用定义¶

会话由三个参数标识：app_name、user_id 和 session_id。这种三层结构支持多租户应用，每个用户可以拥有多个并发会话。

user_id 和 session_id 都是任意字符串标识符，你可以根据应用需求自行定义。ADK 对 session_id 不做格式验证（除了 .strip()）——你可以使用任何对应用有意义的字符串值：

• user_id 示例：用户 UUID（"550e8400-e29b-41d4-a716-446655440000"）、电子邮件地址（"alice@example.com"）、数据库 ID（"user_12345"）或简单标识符（"demo-user"）
• session_id 示例：自定义会话令牌、UUID、基于时间戳的 ID（"session_2025-01-27_143022"）或简单标识符（"demo-session"）

自动生成：如果你将 session_id=None 或空字符串传递给 create_session()，ADK 会自动为你生成一个 UUID（例如 "550e8400-e29b-41d4-a716-446655440000"）。

组织层次结构：这些标识符按三级结构组织会话：

app_name → user_id → session_id → Session

这种设计支持以下场景：

• 多租户应用，不同用户拥有隔离的对话空间
• 单个用户拥有多个并发聊天线程（例如不同话题）
• 按设备或按浏览器的会话隔离

推荐模式：获取或创建¶

推荐的生产模式是先检查会话是否存在，如果需要则创建它。这种方法可以安全地处理新会话和对话恢复：

# 获取或创建会话（处理新会话和重连两种情况）
session = await session_service.get_session(
    app_name=APP_NAME,
    user_id=user_id,
    session_id=session_id
)
if not session:
    await session_service.create_session(
        app_name=APP_NAME,
        user_id=user_id,
        session_id=session_id
    )

此模式在所有场景下都能正确工作：

• 新对话：如果会话不存在，会自动创建
• 恢复对话：如果会话已存在（例如网络中断后重连），将复用现有会话并保留完整的对话历史
• 幂等性：可以安全地多次调用而不会出错

重要提示：在使用相同标识符调用 runner.run_live() 之前，会话必须已存在。如果会话不存在，run_live() 将抛出 ValueError: Session not found。

创建 RunConfig¶

RunConfig 定义此特定会话的流式行为——使用哪些模态（文本或音频）、是否启用转录、语音活动检测、主动性以及其他高级功能。

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

# 原生音频模型需要 AUDIO 响应模态和音频转录
response_modalities = ["AUDIO"]
run_config = RunConfig(
    streaming_mode=StreamingMode.BIDI,
    response_modalities=response_modalities,
    input_audio_transcription=types.AudioTranscriptionConfig(),
    output_audio_transcription=types.AudioTranscriptionConfig(),
    session_resumption=types.SessionResumptionConfig()
)

RunConfig 是会话特定的——每个流式会话可以有不同的配置。例如，一个用户可能偏好纯文本响应，而另一个用户使用语音模式。有关完整的配置选项，请参阅第 4 部分：理解 RunConfig。

创建 LiveRequestQueue¶

LiveRequestQueue 是在流式通信期间向智能体发送消息的通信通道。它是一个线程安全的异步队列，缓冲用户消息（文本内容、音频 blob、活动信号），供有序处理。

from google.adk.agents.live_request_queue import LiveRequestQueue

live_request_queue = LiveRequestQueue()

LiveRequestQueue 是会话特定且有状态的——你为每个流式会话创建一个新队列，并在会话结束时关闭它。与 Agent 和 Runner 不同，队列不能在会话间复用。

阶段 3：使用 run_live() 事件循环进行双向流式通信¶

一旦流式循环运行，你就可以并发地向智能体发送消息并接收响应——这就是双向流式通信的实际运作。智能体可以在你发送新输入的同时生成响应，实现自然的基于打断的对话。

向智能体发送消息¶

在流式会话期间，使用 LiveRequestQueue 方法向智能体发送不同类型的消息：

from google.genai import types

# 发送文本内容
content = types.Content(parts=[types.Part(text=json_message["text"])])
live_request_queue.send_content(content)

# 发送音频 blob
audio_blob = types.Blob(
    mime_type="audio/pcm;rate=16000",
    data=audio_data
)
live_request_queue.send_realtime(audio_blob)

这些方法是非阻塞的——它们会立即将消息添加到队列中，无需等待处理。即使在 AI 处理繁重任务时，也能实现流畅、响应灵敏的用户体验。

有关详细的 API 文档，请参阅第 2 部分：使用 LiveRequestQueue 发送消息。

接收和处理事件¶

run_live() 异步生成器在智能体处理输入和生成响应时持续产出 Event 对象。每个事件代表一个离散的发生——部分文本生成、音频块、工具执行、转录、打断或轮次完成。

async for event in runner.run_live(
    user_id=user_id,
    session_id=session_id,
    live_request_queue=live_request_queue,
    run_config=run_config
):
    event_json = event.model_dump_json(exclude_none=True, by_alias=True)
    await websocket.send_text(event_json)

事件专为流式交付而设计——你在响应生成过程中就收到部分响应，而不是只收到完整的消息。这使实时 UI 更新和响应灵敏的用户体验成为可能。

有关全面的事件处理模式，请参阅第 3 部分：使用 run_live() 处理事件。

阶段 4：终止 Live API 会话¶

当流式会话应结束时（用户断开连接、对话完成、超时发生），优雅地关闭队列以向 Live API 会话发出终止信号。

关闭队列¶

通过队列发送关闭信号以终止流式循环：

live_request_queue.close()

这会通知 run_live() 停止产出事件并退出异步生成器循环。智能体会完成正在进行的任何处理，流式会话干净地结束。

FastAPI 应用示例¶

以下是一个完整的 FastAPI WebSocket 应用，展示了所有四个阶段与正确的双向流式通信的集成。关键模式是上行/下行任务：上行任务从 WebSocket 接收消息并发送到 LiveRequestQueue，下行任务从 run_live() 接收 Event 对象并发送到 WebSocket。

完整实现：

import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from google.adk.runners import Runner
from google.adk.agents.run_config import RunConfig, StreamingMode
from google.adk.agents.live_request_queue import LiveRequestQueue
from google.adk.sessions import InMemorySessionService
from google.genai import types
from google_search_agent.agent import agent

# ========================================
# 阶段 1：应用初始化（启动时执行一次）
# ========================================

APP_NAME = "bidi-demo"

app = FastAPI()

# 定义你的会话服务
session_service = InMemorySessionService()

# 定义你的 runner
runner = Runner(
    app_name=APP_NAME,
    agent=agent,
    session_service=session_service
)

# ========================================
# WebSocket 端点
# ========================================

@app.websocket("/ws/{user_id}/{session_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: str, session_id: str) -> None:
    await websocket.accept()

    # ========================================
    # 阶段 2：会话初始化（每个流式会话执行一次）
    # ========================================

    # 创建 RunConfig
    response_modalities = ["AUDIO"]
    run_config = RunConfig(
        streaming_mode=StreamingMode.BIDI,
        response_modalities=response_modalities,
        input_audio_transcription=types.AudioTranscriptionConfig(),
        output_audio_transcription=types.AudioTranscriptionConfig(),
        session_resumption=types.SessionResumptionConfig()
    )

    # 获取或创建会话
    session = await session_service.get_session(
        app_name=APP_NAME,
        user_id=user_id,
        session_id=session_id
    )
    if not session:
        await session_service.create_session(
            app_name=APP_NAME,
            user_id=user_id,
            session_id=session_id
        )

    # 创建 LiveRequestQueue
    live_request_queue = LiveRequestQueue()

    # ========================================
    # 阶段 3：活跃会话（并发双向通信）
    # ========================================

    async def upstream_task() -> None:
        """从 WebSocket 接收消息并发送到 LiveRequestQueue。"""
        try:
            while True:
                # 从 WebSocket 接收文本消息
                data: str = await websocket.receive_text()

                # 发送到 LiveRequestQueue
                content = types.Content(parts=[types.Part(text=data)])
                live_request_queue.send_content(content)
        except WebSocketDisconnect:
            # 客户端断开连接 - 通知队列关闭
            pass

    async def downstream_task() -> None:
        """从 run_live() 接收 Event 并发送到 WebSocket。"""
        async for event in runner.run_live(
            user_id=user_id,
            session_id=session_id,
            live_request_queue=live_request_queue,
            run_config=run_config
        ):
            # 将事件以 JSON 格式发送到 WebSocket
            await websocket.send_text(
                event.model_dump_json(exclude_none=True, by_alias=True)
            )

    # 并发运行两个任务
    try:
        await asyncio.gather(
            upstream_task(),
            downstream_task(),
            return_exceptions=True
        )
    finally:
        # ========================================
        # 阶段 4：会话终止
        # ========================================

        # 始终关闭队列，即使发生异常
        live_request_queue.close()

关键概念¶

上行任务（WebSocket → LiveRequestQueue）

上行任务持续从 WebSocket 客户端接收消息，并将其转发到 LiveRequestQueue。这使用户可以随时向智能体发送消息，即使智能体正在生成响应。

async def upstream_task() -> None:
    """从 WebSocket 接收消息并发送到 LiveRequestQueue。"""
    try:
        while True:
            data: str = await websocket.receive_text()
            content = types.Content(parts=[types.Part(text=data)])
            live_request_queue.send_content(content)
    except WebSocketDisconnect:
        pass  # 客户端断开连接

下行任务（run_live() → WebSocket）

下行任务持续从 run_live() 接收 Event 对象，并将其发送到 WebSocket 客户端。这会将智能体的响应、工具执行、转录和其他事件实时流式传输给用户。

async def downstream_task() -> None:
    """从 run_live() 接收 Event 并发送到 WebSocket。"""
    async for event in runner.run_live(
        user_id=user_id,
        session_id=session_id,
        live_request_queue=live_request_queue,
        run_config=run_config
    ):
        await websocket.send_text(
            event.model_dump_json(exclude_none=True, by_alias=True)
        )

带清理的并发执行

两个任务使用 asyncio.gather() 并发运行，实现真正的双向流式通信。try/finally 块确保即使发生异常也会调用 LiveRequestQueue.close()，最大限度地减少会话资源使用。

try:
    await asyncio.gather(
        upstream_task(),
        downstream_task(),
        return_exceptions=True
    )
finally:
    live_request_queue.close()  # 始终清理

这种模式——带有保证清理的并发上行/下行任务——是生产级流式应用的基础。生命周期模式（初始化一次，流式传输多次）实现了高效的资源使用和清晰的关注点分离，应用组件保持无状态和可重用，而会话特定的状态隔离在 LiveRequestQueue、RunConfig 和会话记录中。

生产注意事项¶

此示例展示了核心模式。对于生产应用，请考虑：

• 错误处理（ADK）：为 ADK 流式事件添加适当的错误处理。有关错误事件处理的详情，请参阅第 3 部分：错误事件。
  • 通过在关闭期间捕获 asyncio.CancelledError 来优雅地处理任务取消
  • 使用 return_exceptions=True 检查 asyncio.gather() 的异常——异常不会自动传播
• 错误处理（Web）：在上行/下行任务中处理 Web 应用特定的错误。例如，使用 FastAPI 时你需要：
  • 捕获 WebSocketDisconnect（客户端断开连接）、ConnectionClosedError（连接丢失）和 RuntimeError（向已关闭的连接发送数据）
  • 在发送前使用 websocket.client_state 验证 WebSocket 连接状态，以防止连接关闭时出错
• 身份验证和授权：为端点实现身份验证和授权
• 速率限制和配额：添加速率限制和超时控制。有关并发会话和配额管理的指导，请参阅第 4 部分：并发 Live API 会话和配额管理。
• 结构化日志：使用结构化日志进行调试。
• 持久化会话服务：考虑使用持久化会话服务（DatabaseSessionService 或 VertexAiSessionService）。有关更多详情，请参阅 ADK 会话服务文档。

1.6 我们将学到什么¶

本指南将带你逐步了解 ADK Gemini Live API Toolkit 的架构，遵循流式应用的自然流程：消息如何从用户上行传输到智能体，事件如何从智能体下行传输到用户，如何配置会话行为，以及如何实现多模态功能。每个部分都聚焦于流式架构的特定组件，提供你可以立即应用的实用模式：

• 第 2 部分：使用 LiveRequestQueue 发送消息 - 了解 ADK 的 LiveRequestQueue 如何提供统一接口来处理文本、音频和控制消息。你将理解 LiveRequest 消息模型、如何发送不同类型的内容、管理用户活动信号，以及通过单一优雅的 API 优雅地终止会话。

• 第 3 部分：使用 run_live() 处理事件 - 掌握 ADK 流式架构中的事件处理。学习如何处理不同的事件类型（文本、音频、转录、工具调用）、使用打断和轮次完成信号管理对话流程、为网络传输序列化事件，以及利用 ADK 的自动工具执行。理解事件处理对于构建响应灵敏的流式应用至关重要。

• 第 4 部分：理解 RunConfig - 配置复杂的流式行为，包括多模态交互、智能主动性、会话恢复和成本控制。了解不同模型上可用的功能，以及如何通过 RunConfig 声明式地控制流式会话。

• 第 5 部分：如何使用音频、图像和视频 - 使用 ADK 的多模态能力实现语音和视频功能。了解音频规格、流式架构、语音活动检测、音频转录，以及构建自然的语音 AI 体验的最佳实践。

前置条件和学习资源¶

为了在生产环境中构建 ADK Gemini Live API Toolkit 应用，我们建议具备以下技术的基础知识：

ADK（Agent Development Kit）

Google 的生产就绪框架，用于构建具有流式能力的 AI 智能体。ADK 为会话管理、工具编排和状态持久化提供了高级抽象，消除了从头实现底层流式基础设施的需要。

Live API（Gemini Live API 和 Gemini Live API (Agent Platform)）

Google 的实时对话式 AI 技术，支持与 Gemini 模型进行低延迟双向流式通信。Live API 提供了支持 ADK 流式能力的底层 WebSocket 协议，处理多模态输入/输出和自然对话流程。

Python 异步编程

Python 内置的异步编程支持，使用 async/await 语法和 asyncio 库。ADK 流式通信建立在异步生成器和协程之上，需要熟悉异步函数、等待任务和使用 asyncio.gather() 的并发执行等概念。

Pydantic

一个使用 Python 类型注解进行数据验证和设置管理的 Python 库。ADK 大量使用 Pydantic 模型处理结构化数据（如 Event、RunConfig 和 Content），提供类型安全、自动验证和通过 .model_dump_json() 进行 JSON 序列化。

FastAPI

一个现代、高性能的 Python Web 框架，用于构建带有自动 OpenAPI 文档的 API。FastAPI 对 WebSocket 和异步请求处理的原生支持使其成为构建 ADK 流式端点的理想选择。FastAPI 包含在 adk-python 包中，并被 ADK 的 adk web 工具用于快速原型开发。也可以使用其他支持 WebSocket 的框架（如 Flask-SocketIO 或 Starlette）。

WebSocket

一种提供全双工（双向）通信通道的协议，在单个 TCP 连接上运行。WebSocket 实现了客户端和服务器之间的实时双向数据流，使其成为流式应用的标准传输方式。与 HTTP 请求-响应不同，WebSocket 连接是持久的，允许双方随时发送消息。

SSE（Server-Sent Events）

一种服务器通过 HTTP 向 Web 客户端推送数据的标准。与 WebSocket 不同，SSE 是单向的（仅服务端到客户端），因此更简单但灵活性较低。当你不需要客户端到服务端的流式传输时（例如用户输入通过单独的 HTTP POST 请求发送），SSE 适用于流式传输智能体响应。

虽然本指南全面介绍了 ADK 特定的概念，但熟悉这些底层技术将帮助你构建更健壮的生产应用。

总结¶

在本简介中，你了解了 ADK 如何将复杂的实时流式基础设施转化为开发者友好的框架。我们介绍了 Live API 双向流式通信能力的基础知识，研究了 ADK 如何通过 LiveRequestQueue、Runner 和 run_live() 等抽象简化流式通信的复杂性，并探索了从初始化到会话终止的完整应用生命周期。你现在了解了 ADK 如何承担繁重的工作——LLM 端流式连接管理、状态持久化、平台差异和事件协调——从而你可以专注于构建智能体体验。有了这个基础，你已准备好在后续部分深入了解发送消息、处理事件、配置会话和实现多模态功能的具体内容。

下一章：第 2 部分：使用 LiveRequestQueue 发送消息 →