Gemini 2.5 Pro 预览版现已可用于生产环境！了解详情

此页面由 Cloud Translation API 翻译。

Live API

Live API 支持与 Gemini 建立低延迟的双向语音和视频互动。借助 Live API，您可以为最终用户提供自然的、类似人类的语音对话体验，并能够使用语音指令中断模型的回答。该模型可以处理文本、音频和视频输入，并提供文本和音频输出。

您可以在 Google AI Studio 中试用 Live API。

使用 Live API

本部分介绍了如何将 Live API 与我们的某个 SDK 搭配使用。如需详细了解底层 WebSockets API，请参阅 WebSockets API 参考文档。

发送和接收短信

import asyncio
from google import genai

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

config = {"response_modalities": ["TEXT"]}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        while True:
            message = input("User> ")
            if message.lower() == "exit":
                break
            await session.send_client_content(
                turns={"role": "user", "parts": [{"text": message}]}, turn_complete=True
            )

            async for response in session.receive():
                if response.text is not None:
                    print(response.text, end="")

if __name__ == "__main__":
    asyncio.run(main())

接收音频

以下示例展示了如何接收音频数据并将其写入 .wav 文件。

import asyncio
import wave
from google import genai

client = genai.Client(api_key="GEMINI_API_KEY", http_options={'api_version': 'v1alpha'})
model = "gemini-2.0-flash-live-001"

config = {"response_modalities": ["AUDIO"]}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        wf = wave.open("audio.wav", "wb")
        wf.setnchannels(1)
        wf.setsampwidth(2)
        wf.setframerate(24000)

        message = "Hello? Gemini are you there?"
        await session.send_client_content(
            turns={"role": "user", "parts": [{"text": message}]}, turn_complete=True
        )

        async for idx,response in async_enumerate(session.receive()):
            if response.data is not None:
                wf.writeframes(response.data)

            # Un-comment this code to print audio data info
            # if response.server_content.model_turn is not None:
            #      print(response.server_content.model_turn.parts[0].inline_data.mime_type)

        wf.close()

if __name__ == "__main__":
    asyncio.run(main())

音频格式

Live API 支持以下音频格式：

输入音频格式：16kHz 小端字节序的原始 16 位 PCM 音频
输出音频格式：24kHz 小端字节序的原始 16 位 PCM 音频

流式传输音频和视频

系统指令

借助系统说明，您可以根据自己的特定需求和使用情形来控制模型的行为。系统说明可以在设置配置中设置，并将在整个会话期间保持有效。

from google.genai import types

config = {
    "system_instruction": types.Content(
        parts=[
            types.Part(
                text="You are a helpful assistant and answer in a friendly tone."
            )
        ]
    ),
    "response_modalities": ["TEXT"],
}

增量内容更新

使用增量更新发送文本输入、建立会话上下文或恢复会话上下文。对于简短的情境，您可以发送精细互动来表示确切的事件顺序：

PythonJSON

turns = [
    {"role": "user", "parts": [{"text": "What is the capital of France?"}]},
    {"role": "model", "parts": [{"text": "Paris"}]},
]

await session.send_client_content(turns=turns, turn_complete=False)

turns = [{"role": "user", "parts": [{"text": "What is the capital of Germany?"}]}]

await session.send_client_content(turns=turns, turn_complete=True)

{
  "clientContent": {
    "turns": [
      {
        "parts":[
          {
            "text": ""
          }
        ],
        "role":"user"
      },
      {
        "parts":[
          {
            "text": ""
          }
        ],
        "role":"model"
      }
    ],
    "turnComplete": true
  }
}

对于较长的上下文，建议提供单个消息摘要，以便释放上下文窗口以供后续互动。

更改语音

Live API 支持以下语音：Puck、Charon、Kore、Fenrir、Aoede、Leda、Orus 和 Zephyr。

如需指定语音，请在会话配置中在 speechConfig 对象中设置语音名称：

PythonJSON

from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Kore")
        )
    )
)

{
  "voiceConfig": {
    "prebuiltVoiceConfig": {
      "voiceName": "Kore"
    }
  }
}

使用函数调用

您可以使用 Live API 定义工具。如需详细了解函数调用，请参阅函数调用教程。

必须在会话配置中定义工具：

config = types.LiveConnectConfig(
    response_modalities=["TEXT"],
    tools=[set_light_values]
)

async with client.aio.live.connect(model=model, config=config) as session:
    await session.send_client_content(
        turns={
            "role": "user",
            "parts": [{"text": "Turn the lights down to a romantic level"}],
        },
        turn_complete=True,
    )

    async for response in session.receive():
        print(response.tool_call)

从单个提示中，模型可以生成多个函数调用以及串联其输出所需的代码。此代码在沙盒环境中执行，会生成后续的 BidiGenerateContentToolCall 消息。执行会暂停，直到每个函数调用的结果都可用，以确保顺序处理。

客户端应响应 BidiGenerateContentToolResponse。

音频输入和音频输出会对模型使用函数调用功能的能力产生负面影响。

处理中断

用户可以随时中断模型的输出。当语音活动检测 (VAD) 检测到中断时，系统会取消并舍弃正在生成的音频。只有已发送到客户端的信息会保留在会话历史记录中。然后，服务器会发送 BidiGenerateContentServerContent 消息来报告中断。

此外，Gemini 服务器会舍弃所有待处理的函数调用，并发送包含已取消调用的 ID 的 BidiGenerateContentServerContent 消息。

async for response in session.receive():
    if response.server_content.interrupted is not None:
        # The generation was interrupted

配置语音活动检测 (VAD)

默认情况下，该模型会自动对连续的音频输入流执行语音活动检测 (VAD)。您可以使用设置配置的 realtimeInputConfig.automaticActivityDetection 字段配置 VAD。

当音频流暂停超过一秒时（例如，由于用户关闭了麦克风），应发送 audioStreamEnd 事件以刷新所有缓存的音频。客户端可以随时恢复发送音频数据。

或者，您也可以在设置消息中将 realtimeInputConfig.automaticActivityDetection.disabled 设置为 true 以停用自动 VAD。在此配置中，客户端负责检测用户语音并在适当的时间发送 activityStart 和 activityEnd 消息。在此配置中，系统不会发送 audioStreamEnd。而是会通过 activityEnd 消息标记任何流中断。

我们将在未来几周内推出对此功能的 SDK 支持。

获取令牌数量

您可以在返回的服务器消息的 usageMetadata 字段中找到消耗的令牌总数。

from google.genai import types

async with client.aio.live.connect(
    model='gemini-2.0-flash-live-001',
    config=types.LiveConnectConfig(
        response_modalities=['AUDIO'],
    ),
) as session:
    # Session connected
    while True:
        await session.send_client_content(
            turns=types.Content(role='user', parts=[types.Part(text='Hello world!')])
        )
        async for message in session.receive():
            # The server will periodically send messages that include
            # UsageMetadata.
            if message.usage_metadata:
                usage = message.usage_metadata
                print(
                    f'Used {usage.total_token_count} tokens in total. Response token'
                    ' breakdown:'
                )
            for detail in usage.response_tokens_details:
                match detail:
                  case types.ModalityTokenCount(modality=modality, token_count=count):
                      print(f'{modality}: {count}')

            # For the purposes of this example, placeholder input is continually fed
            # to the model. In non-sample code, the model inputs would come from
            # the user.
            if message.server_content and message.server_content.turn_complete:
                break

配置会话恢复

如需防止在服务器定期重置 WebSocket 连接时会话终止，请在设置配置中配置 sessionResumption 字段。

传递此配置会导致服务器发送 SessionResumptionUpdate 消息，这些消息可用于通过将上次恢复令牌作为后续连接的 SessionResumptionConfig.handle 传递来恢复会话。

from google.genai import types

print(f"Connecting to the service with handle {previous_session_handle}...")
async with client.aio.live.connect(
    model="gemini-2.0-flash-live-001",
    config=types.LiveConnectConfig(
        response_modalities=["AUDIO"],
        session_resumption=types.SessionResumptionConfig(
            # The handle of the session to resume is passed here,
            # or else None to start a new session.
            handle=previous_session_handle
        ),
    ),
) as session:
    # Session connected
    while True:
        await session.send_client_content(
            turns=types.Content(
                role="user", parts=[types.Part(text="Hello world!")]
            )
        )
        async for message in session.receive():
            # Periodically, the server will send update messages that may
            # contain a handle for the current state of the session.
            if message.session_resumption_update:
                update = message.session_resumption_update
                if update.resumable and update.new_handle:
                    # The handle should be retained and linked to the session.
                    return update.new_handle

            # For the purposes of this example, placeholder input is continually fed
            # to the model. In non-sample code, the model inputs would come from
            # the user.
            if message.server_content and message.server_content.turn_complete:
                break

在会话断开连接之前接收消息

服务器发送 GoAway 消息，表示当前连接即将终止。此消息包含 timeLeft，表示剩余时间，可让您在连接被终止为“ABORTED”之前采取进一步行动。

在生成完成后收到消息

服务器发送 generationComplete 消息，表示模型已完成生成回答。

启用上下文窗口压缩

如需延长会话时长并避免突然断开连接，您可以在会话配置中设置 contextWindowCompression 字段，以启用上下文窗口压缩。

在 ContextWindowCompressionConfig 中，您可以配置滑动窗口机制以及触发压缩的令牌数量。

from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    context_window_compression=(
        # Configures compression with default parameters.
        types.ContextWindowCompressionConfig(
            sliding_window=types.SlidingWindow(),
        )
    ),
)

更改媒体分辨率

您可以通过在会话配置中设置 mediaResolution 字段来指定输入媒体的媒体分辨率：

from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    media_resolution=types.MediaResolution.MEDIA_RESOLUTION_LOW,
)

限制

在规划项目时，请考虑 Live API 和 Gemini 2.0 的以下限制。

客户端身份验证

Live API 仅提供服务器到服务器身份验证，不建议直接供客户端使用。客户端输入应通过中间应用服务器路由，以便通过 Live API 进行安全身份验证。

会话时长

您可以通过启用会话压缩功能将会话时长延长到无限时长。不压缩时，仅音频会话的时长限制为 15 分钟，音频和视频会话的时长限制为 2 分钟。如果超出这些限制且未进行压缩，连接将会终止。

上下文窗口

会话的上下文窗口限制为 32,000 个 token。

第三方集成

对于网站和移动应用部署，您可以探索以下选项：