Live API - Websockets 快速入门
技术背景
Gemini Live API 提供了低延迟的双向流式交互能力，支持文本、音频、视频等多种模态。官方 SDK 封装了底层细节，但理解 WebSocket 协议对于调试、性能优化和构建非 Python 客户端至关重要。本教程通过直接连接 WebSocket，揭示了 API 的消息结构、握手流程和事件驱动模型，使开发者能够深入掌握实时 AI 系统的构建原理。
核心原理
Live API 基于 WebSocket 全双工通信，客户端发送 BidiGenerateContentSetup 消息进行初始化，随后通过 realtimeInput 发送用户输入，服务器以 serverContent 流式返回模型输出（如音频块）。消息采用 JSON 格式，音频数据以 base64 编码的 PCM 格式传输。模型通过 toolCall 消息请求调用外部函数，客户端必须回复 toolResponse 才能继续生成。整个交互是事件驱动的，客户端需监听并处理不同类型的服务器消息。
架构分析
本示例采用单文件 Python 脚本，核心是 AudioLoop 类，它封装了 WebSocket 连接、消息发送、接收和音频播放。run 方法管理连接生命周期，setup 发送初始化配置，send 发送文本输入，recv 循环接收音频块并写入 WAV 文件，handle_tool_call 处理函数调用。这种设计将协议细节与业务逻辑分离，便于扩展。生产环境可考虑将连接管理、消息解析和音频处理解耦为独立模块，并引入异步队列处理并发消息。
完整实战教程
*版权所有 2026 Google LLC.*
#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
Live API - Websockets 快速入门
本笔记本演示了 Gemini Live API 的简单用法。
本笔记本直接连接到 API 的 WebSocket，以展示不使用 SDK 进行构建的底层细节。
- 如果您对底层 WebSocket 细节不感兴趣，应阅读本笔记本的 SDK 版本。
本笔记本实现了一个简单的基于轮次的聊天，您以文本形式发送消息，模型以音频回复。API 的功能远不止于此。这里的目标是用简单的代码进行演示。
设置
安装与导入
%pip install -q websockets
import asyncio
import base64
import contextlib
import datetime
import os
import json
import wave
import itertools
from websockets.asyncio.client import connect
from IPython.display import display, Audio
常量
要运行以下单元格，您的 API 密钥必须存储在名为 GEMINI_API_KEY 的 Colab Secret 中。如果您还没有 API 密钥，或者不确定如何创建 Colab Secret，请参阅身份验证示例。
from google.colab import userdata
os.environ['GEMINI_API_KEY'] = userdata.get('GEMINI_API_KEY')
Live API 与 gemini-3.1-flash-live-preview 模型配合使用。你需要使用 v1beta API 版本。
MODEL = 'models/gemini-3.1-flash-live-preview'
HOST='generativelanguage.googleapis.com'
URI = f'wss://{HOST}/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key={os.environ["GEMINI_API_KEY"]}'
日志记录
取消注释 logger.setLevel 调用以显示日志消息。
import logging
logger = logging.getLogger('Bidi')
#logger.setLevel('DEBUG')
Wave 文件写入器
本节中的代码对于理解 API 并非必需，可以随意跳到下一节。
在 Colab 中播放音频的最简单方式是将其写入 .wav 文件。
@contextlib.contextmanager
def wave_file(filename, channels=1, rate=24000, sample_width=2):
with wave.open(filename, "wb") as wf:
wf.setnchannels(channels)
wf.setsampwidth(sample_width)
wf.setframerate(rate)
yield wf
主音频循环
下面的类实现了与 Live API 的交互。
class AudioLoop:
def __init__(self, turns, tools=None):
self.turns = turns
self.tools = tools or []
self.ws = None
self.index = 0
async def run(self):
logger.debug('connect')
async with connect(URI, additional_headers={'Content-Type': 'application/json'}) as ws:
self.ws = ws
await self.setup()
for text in self.turns:
await self.send(text)
await self.recv()
async def setup(self):
logger.debug("set_up")
await self.ws.send(json.dumps({
'setup' : {
"model": MODEL,
"generationConfig": {
"responseModalities": ["AUDIO"]
},
"tools": self.tools
}
}))
raw_response = await self.ws.recv(decode=False)
setup_response = json.loads(raw_response.decode('ascii'))
logger.debug(f'Connected: {setup_response}')
async def send(self, text):
logger.debug('send')
print(f"\n> {text}")
# Send text as a realtime input message.
msg = {
"realtimeInput": {
"text": text
}
}
await self.ws.send(json.dumps(msg))
logger.debug('sent')
async def recv(self):
# Start a new `.wav` file.
file_name = f"audio_{self.index}.wav"
with wave_file(file_name) as wav:
self.index += 1
logger.debug('receive')
# Read chunks from the socket.
async for raw_response in self.ws:
response = json.loads(raw_response.decode())
logger.debug(f'got chunk: {str(response)[:200]}')
# Handle tool calls — the model waits for a toolResponse before continuing.
tool_call = response.get('toolCall')
if tool_call is not None:
await self.handle_tool_call(tool_call)
continue
server_content = response.get('serverContent')
if server_content is None:
# Other server messages (sessionResumptionUpdate, etc.) — skip gracefully.
logger.debug(f'Non-content server message: {list(response.keys())}')
continue
# Write audio chunks to the `.wav` file.
model_turn = server_content.get('modelTurn')
if model_turn is not None:
for part in model_turn.get('parts', []):
if 'inlineData' in part:
pcm_data = base64.b64decode(part['inlineData']['data'])
print('.', end='')
logger.debug('Got pcm_data')
wav.writeframes(pcm_data)
# Break out of the loop once the model's turn is complete.
if server_content.get('turnComplete'):
logger.debug('turn_complete')
break
display(Audio(file_name, autoplay=True))
await asyncio.sleep(2)
async def handle_tool_call(self, tool_call):
"""Respond to a toolCall message from the server.
Each FunctionCall must be answered with a FunctionResponse carrying the
same id. Here we return a simple placeholder response so the model can
continue. A real implementation would execute the function and return its
actual output.
"""
function_responses = []
for fc in tool_call['functionCalls']:
print(f'\nTool call: {fc["name"]}({fc.get("args", {})})')
function_responses.append({
'id': fc['id'],
'name': fc['name'],
'response': {'output': 'done'},
})
await self.ws.send(json.dumps({
'toolResponse': {
'functionResponses': function_responses
}
}))
logger.debug(f'Sent tool response for: {[fc["name"] for fc in tool_call["functionCalls"]]}')
这里有5种方法值得介绍：
run - 主循环
此方法：
- 打开一个连接到 Live API 的
websocket - 调用初始的
setup方法 - 然后遍历硬编码的
turns，对每个调用send和recv。
setup - 初始设置
setup 方法发送 setup 消息，并等待响应。在收到模型的 setup_complete 响应之前，你不应尝试从模型发送或接收任何其他内容。
setup 消息（一个 BidiGenerateContentSetup 对象）是你可以设置 model、generation_config、system_instructions、tools 和 safety_settings 的地方。
send - 将输入文本发送到 API
send 方法接收一个硬编码的文本字符串，将其打印出来，并作为 realtimeInput 消息（BidiGenerateContentRealtimeInput）发送给模型。
recv - 从 API 收集音频并播放
recv 方法在一个循环中收集音频块，并将它们写入一个 .wav 文件。一旦模型发送 turnComplete 信号，它就跳出循环，然后播放音频。
它还会检测 toolCall 消息，并在继续之前将它们委托给 handle_tool_call —— 否则模型会停滞等待响应。
为了在 Colab 中保持简单，它会在播放之前收集所有音频。python 脚本 演示了如何在开始接收音频时立即播放（使用 PyAudio），以及如何中断模型。
handle_tool_call - 响应函数调用请求
当模型想要调用函数时，它会发送一条包含一个或多个 FunctionCall 对象的 toolCall 消息，每个对象都有唯一的 id、name 和可选的 args。
客户端必须回复一条 toolResponse，其中包含与每个调用匹配的 FunctionResponse（通过 id 匹配）。在响应发送之前，模型会被阻塞。
这里的实现返回一个简单的占位符 "done" 响应。实际应用会执行函数并返回真实结果。
运行
示例 1：与 Google 搜索的简单用法
turns = ["What is the capital of France?", "Who are you?"]
await AudioLoop(turns=turns, tools=[{'google_search': {}}]).run()
示例 2：函数调用
turns = ["Can you turn off the lights?"]
await AudioLoop(turns=turns, tools=[{'function_declarations': [{'name': 'turn_on_the_lights', 'description': 'Turns on the lights'}, {'name': 'turn_off_the_lights', 'description': 'Turns off the lights'}]}]).run()
# Note: code_execution is not supported in the Live API.
后续步骤
<a name="next_steps"></a>
本教程仅展示了使用 Python GenAI SDK 的 Live API 的基本用法。
- 如果您不需要代码，只想尝试多媒体流式传输，请使用 Google AI Studio 中的 Live API。
- 如果您想了解如何使用 Live API 和 SDK 设置可中断的流式音频和视频，请参阅 GitHub 中的 Live API 示例。
- 尝试 Live API 中的工具使用教程，了解 Gemini 在 Live API 中的函数调用功能。
- 其他 Gemini 示例也可以在 Cookbook 中找到。
常见问题
常见错误包括：1) API 密钥未正确设置，导致连接失败（401）；2) 未等待 setup_complete 就发送消息，导致消息丢失；3) 未处理 toolCall 消息，模型会停滞等待响应；4) 音频格式不匹配（采样率、位深），导致播放异常。排查时需启用日志（logger.setLevel('DEBUG')）观察消息流，并检查 WebSocket 连接状态。生产环境应实现重连机制、心跳检测和错误恢复。
企业应用扩展
生产环境优化建议：1) 使用异步音频流播放（如 PyAudio）实现低延迟；2) 实现中断机制，允许用户打断模型输出；3) 集成函数调用，执行真实业务逻辑；4) 使用连接池管理多个会话；5) 添加监控和日志聚合；6) 考虑使用官方 SDK 以简化开发，但保留底层控制能力。
总结
本教程深入 Gemini Live API 的底层 WebSocket 实现，演示如何不依赖 SDK 直接与模型进行双向实时通信。通过一个简单的文本输入、音频输出的轮次对话示例，详细解析了连接建立、消息格式、音频流接收、工具调用处理等核心环节。适合希望理解实时 AI 交互原理、构建自定义客户端或集成到现有系统的开发者。教程包含完整代码、环境配置、常见问题排查及生产环境优化建议，帮助开发者快速上手并规避典型陷阱。
- --
内容来源与验证
- 内容等级：PREMIUM_A
- 质量等级：S_CANDIDATE
- 难度：中级
- 验证等级：V1
- 原文作者：
- 原文日期：
- 来源许可：Apache-2.0
- 来源版本：
5cfe4263c07fa185d484b547b3ca2cf602ebb987 - 查看原文
站内还有更多完整实战教程
加入会员，解锁全部教程、完整代码与持续更新内容。
加入会员