Live API

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

本文档概述了 Live API，并介绍了以下主题：

如需了解如何开始交互式对话或 API 可以使用哪些工具，请参阅以下文档：

支持的模型

您可以将 Live API 与 Google Gen AI SDK 和 Vertex AI Studio 搭配使用。某些功能（例如文本输入和输出）仅在使用 Gen AI SDK 时可用。

您可以将 Live API 与以下模型搭配使用：

模型版本 可用性级别
gemini-live-2.5-flash 非公开正式版*
gemini-live-2.5-flash-preview-native-audio 公开预览版

*如需申请访问权限，请与您的 Google 客户支持团队代表联系。

如需了解更多信息（包括技术规范和限制），请参阅 Live API 参考文档。

入门示例

如需开始使用 Live API，您可以参阅以下笔记本教程、演示应用或指南。您可以使用 Python SDK 或通过直接进行 WebSocket 调用来与 API 进行交互。

选项 说明 优点 使用场景
Gen AI SDK 一个高级 Python 库，可简化与 Live API 的交互。 更易于使用，可抽象出低级连接细节，并与其他 Google Cloud 服务良好集成。 建议大多数 Python 开发者使用，尤其是在现有 Python 应用中进行快速原型设计和集成时。
WebSockets 一种低级别通信协议，用于与 API 端点进行直接双向通信。 与语言无关（可从任何支持 WebSocket 的语言中使用），并可对数据流进行精细控制。 非常适合非 Python 环境或需要对数据流和连接管理进行最大程度控制的应用。

笔记本教程

从 GitHub 下载这些笔记本教程，或在您选择的环境中打开它们。

演示应用和指南

实时 API 功能

Live API 具有以下功能：

  • 实时多模态理解： 与 Gemini 对话，讨论它在视频 Feed 中或通过屏幕共享看到的内容，利用对流式音频和视频的内置支持。
  • 内置工具使用：将工具（例如函数调用和基于 Google 搜索的 Grounding）集成到对话中，以实现更实用、更动态的互动。
  • 低延迟互动：与 Gemini 进行低延迟的类人互动。
  • 多语言支持： 支持 24 种语言的对话。
  • （仅限正式版）支持预配的吞吐量：使用固定费用、固定期限的订阅，为 Vertex AI 上受支持的生成式 AI 模型（包括 Live API）预留吞吐量。

带有 Live API 的 Gemini 2.5 Flash 还包含原生音频，以公开预览版的形式提供。原生音频引入了以下功能：

如需详细了解原生音频，请参阅内置工具。

支持的音频格式

Live API 支持以下音频格式：

  • 输入音频：16 kHz 的原始 16 位 PCM 音频，小端
  • 输出音频：24 kHz 的原始 16 位 PCM 音频，采用小端序

根据音频输入获取文本回答

如需发送音频并接收文字回复，请将音频转换为 16 位 PCM、16kHz、单声道格式。以下示例展示了如何读取 WAV 文件并以正确的格式发送该文件：

Python

# Test file: https://storage.googleapis.com/generativeai-downloads/data/16000.wav
# Install helpers for converting files: pip install librosa soundfile

import asyncio
import io
from pathlib import Path
from google import genai
from google.genai import types
import soundfile as sf
import librosa

client = genai.Client(
    vertexai=True,
    project=GOOGLE_CLOUD_PROJECT,
    location=GOOGLE_CLOUD_LOCATION,
)
model = "gemini-live-2.5-flash"
config = {"response_modalities": ["TEXT"]}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:

        buffer = io.BytesIO()
        y, sr = librosa.load("sample.wav", sr=16000)
        sf.write(buffer, y, sr, format="RAW", subtype="PCM_16")
        buffer.seek(0)
        audio_bytes = buffer.read()

        # If already in correct format, you can use this:
        # audio_bytes = Path("sample.pcm").read_bytes()

        await session.send_realtime_input(
            audio=types.Blob(data=audio_bytes, mime_type="audio/pcm;rate=16000")
        )

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

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

根据文本输入获取语音回答

如需发送文本输入并接收合成语音响应，请使用以下示例：

Python

import asyncio
import numpy as np
from IPython.display import Audio, Markdown, display
from google import genai
from google.genai.types import (
  Content,
  LiveConnectConfig,
  HttpOptions,
  Modality,
  Part,
  SpeechConfig,
  VoiceConfig,
  PrebuiltVoiceConfig,
)

client = genai.Client(
  vertexai=True,
  project=GOOGLE_CLOUD_PROJECT,
  location=GOOGLE_CLOUD_LOCATION,
)

voice_name = "Aoede"

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

async with client.aio.live.connect(
  model="gemini-live-2.5-flash",
  config=config,
) as session:
  text_input = "Hello? Gemini are you there?"
  display(Markdown(f"**Input:** {text_input}"))

  await session.send_client_content(
      turns=Content(role="user", parts=[Part(text=text_input)]))

  audio_data = []
  async for message in session.receive():
      if (
          message.server_content.model_turn
          and message.server_content.model_turn.parts
      ):
          for part in message.server_content.model_turn.parts:
              if part.inline_data:
                  audio_data.append(
                      np.frombuffer(part.inline_data.data, dtype=np.int16)
                  )

  if audio_data:
      display(Audio(np.concatenate(audio_data), rate=24000, autoplay=True))

如需查看有关发送文本的更多示例，请参阅我们的入门指南。

转录音频

如需转写输入和输出音频，请使用以下示例启用转写：

Python

import asyncio
from google import genai
from google.genai import types

client = genai.Client(
    vertexai=True,
    project=GOOGLE_CLOUD_PROJECT,
    location=GOOGLE_CLOUD_LOCATION,
)
model = "gemini-live-2.5-flash"

config = {
    "response_modalities": ["AUDIO"],
    "input_audio_transcription": {},
    "output_audio_transcription": {}
}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        message = "Hello? Gemini are you there?"

        await session.send_client_content(
            turns={"role": "user", "parts": [{"text": message}]}, turn_complete=True
        )

        async for response in session.receive():
            if response.server_content.model_turn:
                print("Model turn:", response.server_content.model_turn)
            if response.server_content.input_transcription:
                print("Input transcript:", response.server_content.input_transcription.text)
            if response.server_content.output_transcription:
                print("Output transcript:", response.server_content.output_transcription.text)

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

WebSockets

# Set model generation_config
CONFIG = {
    'response_modalities': ['AUDIO'],
}

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {bearer_token[0]}",
}

# Connect to the server
async with connect(SERVICE_URL, additional_headers=headers) as ws:
    # Setup the session
    await ws.send(
        json.dumps(
            {
                "setup": {
                    "model": "gemini-2.0-flash-live-preview-04-09",
                    "generation_config": CONFIG,
                    'input_audio_transcription': {},
                    'output_audio_transcription': {}
                }
            }
        )
    )

    # Receive setup response
    raw_response = await ws.recv(decode=False)
    setup_response = json.loads(raw_response.decode("ascii"))

    # Send text message
    text_input = "Hello? Gemini are you there?"
    display(Markdown(f"**Input:** {text_input}"))

    msg = {
        "client_content": {
            "turns": [{"role": "user", "parts": [{"text": text_input}]}],
            "turn_complete": True,
        }
    }

    await ws.send(json.dumps(msg))

    responses = []
    input_transcriptions = []
    output_transcriptions = []

    # Receive chucks of server response
    async for raw_response in ws:
        response = json.loads(raw_response.decode())
        server_content = response.pop("serverContent", None)
        if server_content is None:
            break

        if (input_transcription := server_content.get("inputTranscription")) is not None:
            if (text := input_transcription.get("text")) is not None:
                input_transcriptions.append(text)
        if (output_transcription := server_content.get("outputTranscription")) is not None:
            if (text := output_transcription.get("text")) is not None:
                output_transcriptions.append(text)

        model_turn = server_content.pop("modelTurn", None)
        if model_turn is not None:
            parts = model_turn.pop("parts", None)
            if parts is not None:
                for part in parts:
                    pcm_data = base64.b64decode(part["inlineData"]["data"])
                    responses.append(np.frombuffer(pcm_data, dtype=np.int16))

        # End of turn
        turn_complete = server_content.pop("turnComplete", None)
        if turn_complete:
            break

    if input_transcriptions:
        display(Markdown(f"**Input transcription >** {''.join(input_transcriptions)}"))

    if responses:
        # Play the returned audio message
        display(Audio(np.concatenate(responses), rate=24000, autoplay=True))

    if output_transcriptions:
        display(Markdown(f"**Output transcription >** {''.join(output_transcriptions)}"))

更多信息

如需详细了解如何使用 Live API，请参阅：