Gemini Live API教程：构建实时语音AI应用

本教程面向有编程基础的开发者，介绍如何使用 Google GenAI SDK 与 Gemini Live API 构建实时多模态应用。通过简单的文本转音频示例，逐步演示了 SDK 安装、API 密钥配置、客户端初始化、模型选择，以及基于异步任务的音频循环实现。教程还深入讲解了会话恢复机制，允许跨会话保持上下文。最后提供了后续学习资源，帮助开发者进一步探索可中断流式音频、视频处理等高级功能。

技术背景

传统 API 采用请求-响应模式，延迟高且无法支持实时交互。Live API 基于 WebSocket 提供低延迟双向流式通信，适用于语音助手、实时翻译、交互式教学等场景。其核心优势在于支持音频、视频帧和文本的多模态输入，并实时输出音频，实现自然的人机对话。主要限制包括：需要稳定的网络连接，且当前处于预览阶段，功能可能变化；TEXT 响应模态不再支持，需通过音频转录获取文本。

核心原理

Live API 通过 WebSocket 连接实现客户端与 Gemini 模型的持续双向数据流。客户端通过 send_realtime_input 发送文本、音频或视频帧，模型以流式方式返回音频块（PCM 数据）和元数据。LiveConnectConfig 配置响应模态（如 AUDIO）和会话恢复参数。异步任务模型允许并发发送和接收，实现实时交互。会话恢复通过 SessionResumptionConfig 传递上一会话的句柄，使模型在 24 小时内记住上下文。

架构分析

系统架构分为客户端和服务器端。客户端使用 google-genai SDK，核心组件包括：client.aio.live.connect 建立 WebSocket 会话；LiveConnectConfig 定义会话参数；send_realtime_input 发送输入；receive() 异步迭代接收响应。服务器端由 Gemini 模型处理多模态输入并生成音频输出。设计上，AudioLoop 类封装了发送和接收逻辑，但当前实现顺序执行，生产环境应使用独立异步任务并发处理。会话恢复机制通过 session_resumption_update 消息更新句柄，实现跨会话上下文保持。

完整实战教程

##### 版权所有 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 - 快速入门

<a target="_blank" href="https://colab.research.google.com/github/google-gemini/cookbook/blob/main/quickstarts/Get_started_LiveAPI.ipynb">

预览：Live API 目前处于预览阶段。

本笔记本演示了 Gemini Live API 的简单用法。有关新功能的概述，请参阅 Gemini Live API 文档。

本笔记本实现了一个简单的基于回合的聊天，您以文本形式发送消息，模型以音频回复。API 的功能远不止于此。这里的目标是用简单的代码进行演示。

API 的某些功能在 Colab 中无法使用，要试用它们，建议查看 GitHub 中的 Live API 示例。

如果您不需要代码，只想尝试多媒体流式传输，请使用 Google AI Studio 中的 Live API。

本教程末尾的 后续步骤 部分提供了其他资源的链接。

设置

安装 SDK

新的 Google Gen AI SDK 通过 Google AI for Developers 和 Vertex AI API 提供对 Gemini 3（及之前模型）的编程访问。除少数例外，在一个平台上运行的代码也将在另一个平台上运行。

有关此新 SDK 的更多详细信息，请参阅 文档 或 入门 笔记本。

%pip install -U -q google-genai

设置你的 API 密钥

要运行以下单元格，你的 API 密钥必须存储在名为 GEMINI_API_KEY 的 Colab 密钥中。如果你还没有 API 密钥，或者不确定如何创建 Colab 密钥，请参阅 [身份验证]

image](../quickstarts/Authentication.ipynb) 查看示例。

from google.colab import userdata
import os

os.environ['GEMINI_API_KEY'] = userdata.get('GEMINI_API_KEY')

初始化 SDK 客户端

客户端将从环境变量中读取你的 API 密钥。

from google import genai
from google.genai import types

client = genai.Client(api_key=os.environ['GEMINI_API_KEY'])

选择模型

Gemini 3.1 Flash Live 模型与 Live API 配合使用，可实现与 Gemini 的低延迟双向语音和视频交互。该模型可以处理文本、音频和视频帧（图像）输入，并提供音频输出（可选转录）。

MODEL = 'gemini-3.1-flash-live-preview'  # @param ['gemini-3.1-flash-live-preview'] {allow-input: true, isTemplate: true}

导入

导入所有必要的模块。

import asyncio
import base64
import contextlib
import os
import wave

from IPython.display import display, Audio

from google import genai
from google.genai import types

简单文本转音频

在 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

通过在 LiveConnectConfig 中设置 response_modalities=["AUDIO"] 来告诉模型返回音频，并使用 send_realtime_input 发送输入文本。当您从模型获得响应时，将数据写入 .wav 文件。

注意：TEXT 响应模态不再受支持。如果您需要音频旁边的文本，请使用启用了输出转录的 AUDIO。
config = types.LiveConnectConfig(
    response_modalities=["AUDIO"]
)

async with client.aio.live.connect(model=MODEL, config=config) as session:
    file_name = 'audio.wav'
    with wave_file(file_name) as wav:
        message = "Hello? Gemini are you there?"
        print("> ", message, "\n")
        await session.send_realtime_input(text=message)

        async for response in session.receive():
            if response.data is not None:
                wav.writeframes(response.data)
                print('.', end='')
            if response.server_content and response.server_content.turn_complete:
                break

display(Audio(file_name, autoplay=True))

迈向异步任务

Live API 的真正强大之处在于它的实时性和可中断性。在简单的步骤序列中，你无法获得这种全部能力。要真正使用这些功能，你需要将 send 和 receive 操作（以及其他操作）移入它们自己的 异步任务 中。

由于 Colab 的限制，本教程并未完全实现交互式异步任务，但它确实朝这个方向迈出了下一步：

设置一个快速日志记录器以便调试（切换到 setLevel('DEBUG') 以查看调试消息）。

import logging

logger = logging.getLogger('Live')
logger.setLevel('INFO')

下面的类实现了与 Live API 的交互。

async def async_enumerate(aiterable):
    n = 0
    async for item in aiterable:
        yield n, item
        n += 1


class AudioLoop:
    def __init__(self, turns=None, config=None):
        self.session = None
        self.index = 0
        self.turns = turns
        if config is None:
            config = types.LiveConnectConfig(
                response_modalities=["AUDIO"]
            )
        self.config = config

    async def run(self):
        logger.debug('connect')
        async with client.aio.live.connect(model=MODEL, config=self.config) as session:
            self.session = session

            async for sent in self.send():
                # Ideally send and recv would be separate tasks.
                await self.recv()

    async def _iter(self):
        for text in self.turns:
            print("message >", text)
            yield text

    async def send(self):
        async for text in self._iter():
            logger.debug('send')

            # Send the message to the model using send_realtime_input.
            await self.session.send_realtime_input(text=text)
            logger.debug('sent')
            yield text

    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 n, response in async_enumerate(self.session.receive()):
                logger.debug(f'got chunk: {str(response)}')

                if response.data is not None:
                    wav.writeframes(response.data)
                    print('.', end='')
                if response.server_content and response.server_content.turn_complete:
                    break

            print('\n<Turn complete>')

        display(Audio(file_name, autoplay=True))
        await asyncio.sleep(2)

这里有 3 个值得描述的方法：

run - 主循环

此方法：

send - 向 API 发送输入文本

send 方法遍历硬编码的轮次，通过 send_realtime_input 将每条消息发送给模型，并将控制权交还给运行循环，以便 recv 可以收集响应。

recv - 从 API 收集音频并播放

recv 方法在循环中收集音频块，并将它们写入 .wav 文件。一旦模型发送 turn_complete 信号，它就跳出循环，然后播放音频。

为了在 Colab 中保持简单，它在播放音频之前收集所有音频。其他示例 演示了如何在开始接收音频时立即播放（使用 PyAudio），以及如何中断模型（在单独的任务中实现输入和音频播放）。

运行

运行它：

await AudioLoop(['Hello', "What's your name?"]).run()

使用可恢复会话

会话恢复允许你通过发送上一个会话中获得的最后一个会话句柄，返回到与 Live API 的先前交互。

当你将会话设置为可恢复时，会话信息会在 Live API 上存储最多 24 小时。在这个时间窗口内，你可以恢复对话并引用之前与模型共享的信息。

下面的会话使用 AUDIO 输出并启用了输出转录，以便你可以在笔记本输出中阅读模型的响应。

辅助函数

async_main 按顺序运行一系列硬编码的轮次。每轮之后，它将音频响应写入 .wav 文件，内联渲染，并从 session_resumption_update 消息中捕获最新的会话句柄。

import asyncio


async def async_main(turns, last_handle=None):
    print(f"Connecting with handle: {last_handle}")
    config = types.LiveConnectConfig(
        response_modalities=["AUDIO"],
        output_audio_transcription=types.AudioTranscriptionConfig(),
        session_resumption=types.SessionResumptionConfig(
            handle=last_handle,
        )
    )

    async with client.aio.live.connect(model=MODEL, config=config) as session:
        for i, message in enumerate(turns):
            print(f"\n> {message}")
            await session.send_realtime_input(text=message)

            file_name = f"resumable_{i}.wav"
            with wave_file(file_name) as wav:
                async for response in session.receive():
                    if response.data is not None:
                        wav.writeframes(response.data)
                        print('.', end='')
                    # session_resumption_update is a top-level field on the response,
                    # not nested under server_content.
                    if response.session_resumption_update:
                        update = response.session_resumption_update
                        if update.resumable and update.new_handle:
                            last_handle = update.new_handle
                    if response.server_content:
                        if response.server_content.output_transcription:
                            print(response.server_content.output_transcription.text, end="")
                        if response.server_content.turn_complete:
                            break

            display(Audio(file_name, autoplay=True))
            await asyncio.sleep(2)

    return last_handle

运行第一个会话，包含两个问题。返回会话句柄，以便稍后恢复：

turns_session_1 = ["Hello, what's your name?", "What is the capital of Brazil?"]
last_handle = await async_main(turns_session_1)

用于恢复的会话句柄保存在 last_handle 中：

last_handle

现在启动一个新的 Live API 会话，指向之前的句柄。模型可以回忆起之前对话中的信息：

turns_session_2 = ["What was the second question I asked you in our previous conversation?"]
await async_main(turns_session_2, last_handle)

后续步骤

<a name="next_steps"></a>

本教程仅展示了使用 Python GenAI SDK 的 Live API 的基本用法。

常见问题

常见错误：1) API 密钥未设置或无效，导致认证失败，需确保环境变量 GEMINI_API_KEY 正确。2) 模型名称错误，需使用 gemini-3.1-flash-live-preview。3) 网络问题导致 WebSocket 断开，可增加重试机制。4) 音频播放问题，需确保环境支持音频输出。排查方法：启用日志记录（logger.setLevel('DEBUG')）查看详细错误；检查 API 密钥权限；确认网络防火墙允许 WebSocket 连接。生产优化：使用连接池、实现自动重连、优化音频缓冲以减少延迟。

企业应用扩展

进一步学习可参考：1) 实现可中断的流式音频和视频，使用 PyAudio 实时播放并支持打断。2) 探索视频理解、空间理解等高级示例。3) 直接使用 WebSocket 协议进行底层控制。4) 集成到语音助手、实时翻译等应用。5) 利用会话恢复实现多轮对话的持久化。

总结

本教程面向有编程基础的开发者，介绍如何使用 Google GenAI SDK 与 Gemini Live API 构建实时多模态应用。通过简单的文本转音频示例，逐步演示了 SDK 安装、API 密钥配置、客户端初始化、模型选择，以及基于异步任务的音频循环实现。教程还深入讲解了会话恢复机制，允许跨会话保持上下文。最后提供了后续学习资源，帮助开发者进一步探索可中断流式音频、视频处理等高级功能。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部语音与实时 API教程