跳至主要内容

Gemini Live API 入门

学习如何使用 Gemini Live API 和 Google ADK 构建实时语音助手。掌握 WebSocket、工具调用与原生音频流。
· 11分钟 阅读

使用 AI 探索

如果您曾尝试构建语音助手,就会知道传统流程有多笨拙:先录制用户语音,用语音转文本(STT)模型转写,再把文本发给大语言模型(LLM),等待响应,然后交给文本转语音(TTS)引擎播报。 

等用户听到答案时,通常已经过去了好几秒的尴尬沉默。这更像是缓慢的请求-响应循环,而不是自然的对话。 

Gemini Live API 通过允许您在有状态的 WebSocket 连接上持续流式传输音频和视频,改变了这一流程。由于模型能原生处理并输出音频,它可以即时响应,适应自然的语音细节,甚至能在句中处理打断(barge-in)。 

在本教程中,我将向您展示如何使用代码优先的 Google Agent Development Kit(ADK)构建一个实时 AI 旅行礼宾。我们将为代理接入 Google 搜索以实时查询目的地天气,连接到一个模拟的航班数据库,并启动本地开发者界面,测试其处理实时对话与工具调用的能力。

如果您是该生态的新手,建议先学习 Building AI Agents with Google ADK 课程 以打好基础。

您将达成的目标: 

  • 了解 Gemini Live API 的工作原理、能力范围与定价。
  • 用 Google Agent Development Kit 配置 Python 开发环境。 
  • 让语音助手同时调用内置工具(Google 搜索)与自定义业务逻辑(航班状态数据库查询)。 
  • 启动本地 Web 界面,与旅行代理进行实时对话。

前置条件: 

  • 具备 Python 基础知识。 
  • 一个 Google Cloud 项目(用于访问 Vertex AI)或 Google AI Studio 的 API 密钥。

什么是 Gemini Live API? 

从本质上说,Gemini Live API 是 Google 提供的开发者工具,让您可以用 Gemini 模型构建实时的语音与视觉交互。它不再局限于回合制的文本回复,而是实现用户与 AI 代理之间持续、自然的对话。 

从技术角度看,Gemini Live API 面向双向、实时流式传输。不同于使用无状态请求-响应的标准 HTTP REST API,Live API 依赖有状态、持久的 WebSocket(WSS)连接。通过这条开放通道,客户端可以流式发送原始音频、视频(以约 1 FPS 的图像帧传输)和文本,而模型会以原生音频与文本即时回传。

Gemini Live 的关键能力

  • 打断(可中断性,Barge-in): 在自然的人类对话中,如果需要澄清问题,我们不会等对方完整说完一大段。Gemini Live 开箱即用地支持打断;用户可以在模型说话的半句中插话,服务器会立即停止音频流,并根据新输入进行调整。
  • 情感对话与原生音频: 与其先生成文本再交给 TTS,引擎如 gemini-live-2.5-flash-native-audio 和 gemini-3.1-flash-live-preview 等模型可原生处理并输出音频。这能保留情感细微变化、笑声、语调适配与语流节奏,让互动更具人性化。 
  • 多模态输入与工具使用: Live API 不局限于音频。它可以通过视频帧“看见”您的环境,并实时触发外部函数或工具(如 Google 搜索或自定义 API),再将结果反馈到语音流中。

Gemini Live API 定价

由于实时流式传输涉及持续的全双工连接,其计费模型不同于标准的静态文本 API。Google 在 Google AI Studio 提供了慷慨的免费层用于原型开发,并在 Vertex AI 上提供面向生产的付费层。 

模型 gemini-3.1-flash-live-preview 以百万 token 为单位计费,或将其转换 

为音视频媒体的基于时长的度量:

指标

免费层(Google AI Studio)

付费层(Vertex AI)

输入价格(文本)

免费

每 100 万 tokens 收费 $0.75

输入价格(音频)

免费

每 100 万 tokens 收费 $3.00(或 $0.005/分钟)

输入价格(图像/视频)

免费

每 100 万 tokens 收费 $1.00(或 $0.002/分钟)

输出价格(文本)

免费

每 100 万 tokens 收费 $4.50

输出价格(音频)

免费

每 100 万 tokens 收费 $12.00(或 $0.018/分钟)

基于 Google 搜索的 Grounding

免费(限速)

每月前 5,000 次提示免费,之后每 1,000 次查询 $14

如果您在做原型,我强烈建议先在 Google AI Studio 免费测试,其在限速前提下提供了充足的可用额度。 

当您准备扩展时,可将项目迁移至 Vertex AI,采用按需付费(Pay-as-you-go)的生产级配置,获得有保障的吞吐与企业级 SLA。

如何访问 Gemini Live API

作为开发者,在实现 Live API 时有若干架构选择。由于它依赖 WebSocket,您如何路由连接会影响时延与安全性。

客户端直连 vs 服务器中转

  • 客户端直连:您的前端(如 Web 或移动应用)直接连接 Google 的 WebSocket 端点。这样可将时延降至最低,因为音频流不会多绕一跳。 
    • 但您不能在前端暴露原始的 Google Cloud API 密钥;必须实现能生成临时一次性令牌的认证流程,以便客户端安全访问。
  • 服务器中转:您的前端将音频流给自有后端服务器,再由后端与 Gemini Live API 建立 WebSocket。这种方式更安全,且可注入后端逻辑(如用户认证或数据库查询),但会给音频流带来轻微的网络延迟。

SDK 与合作伙伴集成

您可以编写原生 WebSocket 代码,但最简单的入门方式是使用 Google 的 GenAI SDK 或 Agent Development Kit(ADK)。 

在企业级生产环境中,生态正快速演进以支持 WebRTC 集成。LiveKit、Daily 的 Pipecat、Twilio 与 Voximplant 等平台提供稳健的基础设施,以处理网络抖动、回声消除与实时音频路由,并可直接接入 Gemini Live API。

用 Gemini Live API 构建实时语音助手

让我们动手实践,构建一个 AI 旅行礼宾。 

该代理将通过 Gemini Live API 与用户进行语音对话,更重要的是,它可以在对话中调用自定义 Python 函数,实时查询航班状态。

我们将使用 Google Agent Development Kit(ADK)以及强大的 gemini-3.1-flash-live-preview 模型,它同时针对标准文本轮次与实时 WebSocket 流式传输进行了优化。

1. 环境搭建

首先,您需要创建一个 Python 虚拟环境并安装所需的 ADK 包。在终端运行以下命令:

# Create and activate a virtual environment
python3 -m venv .venv
source .venv/bin/activate
# Install the Agent Development Kit
pip install google-adk

接下来,配置环境变量,以便 ADK 能安全地与您的 Google Cloud 项目进行身份验证:

export GOOGLE_CLOUD_PROJECT="your-project-id" 
export GOOGLE_CLOUD_LOCATION="us-central1"

如果您使用的是来自 Google AI Studio 的 Gemini API Key,请在项目根目录创建一个 .env 文件:

GEMINI_API_KEY=your_google_ai_studio_api_key

2. 初始化 Gemini Live 代理 

在项目目录创建一个名为 agent.py 的文件。ADK 会自动检测并运行 root_agent 变量。在这段代码中,我们将定义一个自定义 Python 函数来检查航班状态,并把它直接传递到代理的工具带中。初始化代理目录: 

mkdir voice_assistant 
touch voice_assistant/__init__.py 
touch voice_assistant/agent.py

在 voice_assistant/__init__.py 中暴露代理包:

from . import agent

3. 定义代理并暴露自定义工具 

打开 voice_assistant/agent.py,编写旅行代理的配置。以下代码中,我们定义了一个标准的 Python 函数 check_flight_status。 

在 Google ADK 中,任何带有规范类型注解与文档字符串(docstring)的 Python 函数都会自动转换为代理工具。LLM 会根据文档字符串的描述与参数来判断何时、如何调用该函数。 

模型配置 

利用 Gemini 3.1 Flash Live Preview 为了同时支持 ADK 开发服务器中的两种交互模式——标准文本聊天(使用标准 REST generateContent 端点)与实时的双向 WebSocket 麦克风接口——我们将模型显式设置为 gemini-3.1-flash-live-preview。

不同于仅支持流式的旧模型(通过标准 HTTP REST 请求触发会返回 404 Not Found),gemini-3.1-flash-live-preview 原生支持标准 REST 内容生成与有状态的 WebSocket 流式传输。

from google.adk.agents.llm_agent import Agent
from google.adk.tools import google_search
def check_flight_status(flight_no: str) -> str:
   """Check the real-time flight status (e.g. delays, gate info, cancellations).
   Args:
       flight_no: The flight number to look up (e.g., AA120, DL450, UA990).
   """
   flight_no = flight_no.upper().strip()
   if "AA120" in flight_no:
       return "Flight AA120 to Paris is currently delayed by 45 minutes due to air traffic control. New departure time: 3:15 PM. Gate status: B22."
   elif "DL450" in flight_no:
       return "Flight DL450 to London is on time. Gate status: Boarding now at Gate C10."
   elif "UA990" in flight_no:
       return "Flight UA990 to Tokyo is cancelled due to adverse weather conditions at the destination. Passengers are advised to rebook at the counter."
   else:
       return f"Flight {flight_no} is currently scheduled on time. No delay reports found in the flight database."
# Define the root agent with search grounding and custom database flight checks
root_agent = Agent(
   model='gemini-3.1-flash-live-preview',
   name='root_agent',
   description='A real-time AI Travel Concierge with flight tracking and search grounding.',
   instruction=(
       'You are an expert AI Travel Concierge. You assist users with real-time flight status checkups, '
       'destination recommendations, and general travel planning. '
       'Always answer professionally and conversationally. '
       'Use the check_flight_status tool when a user asks about a specific flight status. '
       'Use Google Search to look up weather, transit conditions, or travel advisories at destinations.'
   ),
   tools=[google_search, check_flight_status],
)

4. 运行并测试交互式开发 UI

ADK 内置了开发服务器与 Web UI,便于您在本地测试代理而无需编写前端代码。从工作区启动 Web 服务器:

.venv/bin/adk web .

您应能看到服务器已启动的输出信息:

adk web command in terminal

  1. 在浏览器中打开 http://127.0.0.1:8000。
  2. 从应用列表中选择您的代理。
  3. 点击麦克风按钮,开始实时会话。

adk web interface

从应用列表中选择 root_agent 以打开聊天界面。

4. 对着麦克风说:您能帮我查询一下航班 AA120 的状态吗? 代理会解析您的请求,识别需要数据库信息,触发自定义 check_flight_status Python 函数,并在向您播报答案前显示 JSON 工具调用的结果。

chatting with voice_assistant

5. 测试 Google 搜索 Grounding:接着,问旅行礼宾一个目的地天气问题:伦敦现在的天气如何? 代理会激活内置的 google_search 工具,检索网络,并将实时状况综合进其音频输出中。

agent response

6. 测试打断功能:先问一个较宽泛的问题:T给我推荐罗马最值得参观的历史景点。当代理开始逐条列举时,请大声打断:等等!改查一下 UA990 航班!。代理会立刻停止语音流,打断罗马相关内容,运行自定义航班跟踪工具,并告知 UA990 航班已取消。这展示了 Live API 连接的低时延、全双工特性。

Follow up question for agent

Gemini Live API 生产实践最佳做法

构建实时语音应用需要在提示架构、网络状态管理、音频摄取与资源扩缩等方面做特殊设计。以下要点源自官方开发者指南。

1. 设计清晰的系统指令

要发挥 Gemini Live 的最佳表现,请以严格的层级来组织系统指令(SI):人设(Persona)、对话规则与防护栏(Guardrails)。

  • 分离人设:让每个代理的角色只对应一条独立的系统指令,不要让单个代理承载多个 personas。
  • 划分对话流程:明确区分一次性要素(如“仅询问一次用户姓名与票号”)与开放式对话循环(如“持续回答旅行推荐问题”)。
  • 用句子描述工具调用:按步骤描述工具的使用。例如:“首先,请用户提供票号。然后用该号码调用 look_up_ticket。”
  • 强制语言一致性:原生音频模型在语境变化时可能发生语言漂移。若您的代理仅支持英语,请在指令中强制要求:RESPOND IN English. YOU MUST RESPOND UNMISTAKABLY IN English.

2. 精准的工具定义与调用条件

在向 Live API 暴露工具时:

  • 务必包含类型注解与详细的参数说明。
  • 在工具的元数据描述中显式记录“调用条件”(如“仅在用户确认了出发日期与目的地后调用此工具”)。这可防止 LLM 过早触发函数。

3. 管理原生音频上下文(压缩)

原生音频 token 累积很快,大约每秒对话 25 个 token。对于长时会话,这会迅速占满上下文窗口并增加时延。

  • 在连接设置中启用 ContextWindowCompressionConfig,当达到阈值后丢弃较早的历史:
from google.genai import types
live_config = types.LiveConnectConfig(
   context_window_compression=types.ContextWindowCompressionConfig(
       trigger_tokens=100_000,
       sliding_window=types.SlidingWindow(target_tokens=4_000),
   ),
)

注意:压缩上下文会导致早期对话历史丢失。

4. 透明的会话恢复

实时移动连接容易发生丢包与短暂断连。请实现透明的会话恢复,而不是从头开始全新会话状态:

  • 在流式连接中配置 SessionResumptionConfig(transparent=True)。
  • 监听来自服务器的 session_resumption_update 事件并保存 new_handle。
  • 维护本地的待发消息缓冲区。连接断开时,捕获 APIError 或 GoAway 信号,使用保存的 handle 重连,并仅重放未被确认的客户端消息(利用 server-reported last_consumed_client_message_index 修剪缓冲)。

5. 音频流优化

  • 客户端缓冲:不要在客户端缓冲较长音频帧再发送。请持续以小块(20–40 ms 间)的音频片段进行流式传输,以最小化端到端时延。
  • 麦克风重采样:确保客户端音频采集将输入(通常为 44.1 kHz 或 48 kHz)在发送至 WebSocket 前重采样至 16 kHz PCM。
  • 显式 VAD 信号:让模型处理语音活动检测(VAD)。如需在前端立即显示 UI 指示(如“AI 正在聆听”与“AI 正在思考”),可在配置中启用 explicit_vad_signal=True,以直接在服务器消息流中接收 VAD 启停事件。

6. 安全性(令牌管理)

  • 切勿硬编码主密钥:避免在客户端代码(浏览器或移动应用)中暴露 AI Studio API Key 或 Google Cloud 服务账号凭据。
  • 部署一次性令牌交换:在安全后端构建身份验证端点。当客户端希望发起语音会话时,后端生成短期、受限的 OAuth 或 API 访问令牌(例如通过 Vertex AI IAM),并将其传给客户端。客户端使用该临时凭证直接打开 WebSocket 连接。

结语:对话式 AI 的地平线

从回合制、仅文本的 LLM,迈向低时延的多模态流式交互,标志着我们构建人机界面的范式转变。通过在有状态的 WebSocket 连接中原生处理音视频,Gemini Live API 让我们能打造更像人与人交流、而非静态软件的代理式体验。

在本指南中,我们探讨了 Gemini Live 的内部架构、审视了定价模型,并使用 Google Agent Development Kit(ADK)构建了一个完整可用的 AI 旅行礼宾。我们见证了 ADK 如何将标准 Python 函数映射为可调用的代理工具,使语音助手既能进行广泛的基于网络的搜索 Grounding,也能访问私有的内部数据库查询。

接下来可以构建什么? 

现在您已掌握实时语音代理的基础,不妨扩展您的 AI 旅行礼宾: 

  1. 接入实时视频流:支持摄像头图像传输,让礼宾能实时“看到”地图、登机牌或纸质旅行凭证。 
  2. 集成具交易能力的自定义工具:实现 book_flight 或 reserve_hotel 等会改变实际状态的工具,并集成认证范围,以确保用户确认的安全性。 
  3. 生产级中间件路由:从本地 ADK 测试迁移到基于 WebRTC 的框架,使用 LiveKit 或 Pipecat 将生产语音通话直接从 Web 与移动应用流式传输。

准备好构建更高级的对话式工作流了吗?欢迎查看 DataCamp 的AI Agent Fundamentals技能路径,深入学习代理架构、工具使用与生成式 AI 部署。

Aryan Irani

作者

Aryan Irani

我在互联网上写作与创作。Google Workspace 领域的 Google 开发者专家,毕业于 NMIMS 计算机科学专业,热衷于自动化与生成式 AI 的实践与构建。

主题