LiveAPI 工具快速入门指南

本教程面向有编程基础的开发者，深入讲解 Gemini 多模态 Live API 的工具使用，包括函数调用、代码执行和 Google 搜索。通过实际代码示例，展示如何配置工具、处理异步函数调用（阻塞、中断、等待空闲、静默模式），以及组合使用多个工具。教程强调代码执行作为核心工具，支持在单个 API 调用中集成多种工具，并提供了环境准备、实践步骤和常见问题排查，帮助开发者快速上手并应用于生产环境。

技术背景

Live API 是 Gemini 模型提供的实时多模态交互接口，支持音频、文本和工具调用。在早期模型中，工具使用受限，每次请求只能使用一个工具。Gemini 2.5 在 Live API 中引入了重大变化：所有工具（如 Google 搜索、代码执行、函数调用）都可以通过代码执行统一处理，并支持在单个 API 调用中组合多个工具。这解决了开发者需要多次调用 API 才能完成复杂任务的痛点，提高了效率和灵活性。适用场景包括实时语音助手、交互式数据分析、动态内容生成等。主要限制是 API 仍处于预览阶段，SDK 支持尚不完善，且需要处理异步事件流。

核心原理

Live API 基于 WebSocket 连接，客户端发送实时输入（文本或音频），模型返回流式响应。工具使用通过配置 tools 参数实现，每个工具定义在 function_declarations、code_execution 或 google_search 中。模型在生成响应时，可以决定调用工具，并通过 tool_call 对象返回调用请求。客户端需处理这些请求，执行相应操作，并通过 send_tool_response 发送结果。异步函数调用允许模型在等待工具结果时继续处理其他输入，通过 behavior 和 scheduling 控制阻塞行为。代码执行工具让模型编写并运行 Python 代码，结果作为 code_execution_result 返回。

架构分析

系统架构包括客户端、Live API 和工具执行环境。客户端使用 Google Gen AI SDK 建立 WebSocket 连接，发送配置和输入。API 服务器负责模型推理和工具调度。工具执行环境（如代码执行沙箱、Google 搜索服务）在服务器端运行。数据流：客户端发送输入 -> 模型生成响应（可能包含工具调用）-> 客户端处理工具调用并返回结果 -> 模型继续生成。关键设计决策：使用 v1alpha API 版本以启用实时功能；通过 response_modalities 控制输出类型；使用 asyncio 处理并发事件。代码中 Live 类封装了会话管理、消息队列和工具调用处理，实现了异步非阻塞交互。

完整实战教程

##### 版权所有 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.

Gemini - 多模态实时 API：工具使用

本笔记本提供了如何使用多模态 Live API 与 Gemini 2.5 结合工具的示例。

该 API 提供了 Google 搜索、代码执行和函数调用工具。早期的 Gemini 模型支持这些工具的版本。Gemini 2.5（在 Live API 中）最大的变化是，基本上所有工具都由代码执行处理。通过这一变化，您可以在单个 API 调用中使用多个工具，并且模型可以在单个代码执行块中使用多个工具。

本教程假设您熟悉 Live API，如本教程所述。

设置

安装 SDK

新的 Google Gen AI SDK 提供了对 Gemini 2.5（及之前模型）的编程访问，同时支持 Google AI for Developers 和 Vertex AI API。除少数例外，在一个平台上运行的代码在另一个平台上也能运行。这意味着你可以使用 Developer API 快速构建原型，然后无需重写代码即可将应用迁移到 Vertex AI。

关于这个新 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 密钥。

要使用实时 API，你需要将客户端版本设置为 v1alpha。

from google import genai

client = genai.Client(http_options={"api_version": "v1alpha"})

选择模型

选择最新的稳定版模型或预览版。请注意，Colab 不支持生成音频的音频预览模型，它只显示文本输出。

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

导入

import asyncio
import contextlib
import json
import wave

from IPython.display import display, Markdown, Audio, HTML

from google import genai
from google.genai import types

工具函数

你将使用 Live API 的音频输出，在 Colab 中听到它的最简单方式是将 PCM 数据写入为 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

使用日志记录器，这样可以更方便地开启或关闭调试消息。

import logging
logger = logging.getLogger('Live')
logger.setLevel('INFO')
#logger.setLevel('DEBUG')  # Switch between "INFO" and "DEBUG" to toggle debug messages.

开始使用

Live API 的大部分设置与入门教程类似。由于本教程不关注 API 的实时交互性，代码已简化：此代码使用 Live API，但仅发送单个文本提示，并监听单轮回复。

您可以在任何示例上设置 modality="TEXT" 以获取输出的文本版本。

n = 0
async def run(prompt, modality="AUDIO", tools=None):
  global n
  if tools is None:
    tools=[]

  config = {
          "tools": tools,
          "response_modalities": [modality]
  }

  async with client.aio.live.connect(model=MODEL_ID, config=config) as session:
    display(Markdown(data=prompt))
    display(Markdown(data='-------------------------------'))
    await session.send_realtime_input(
      text=prompt
    )

    audio = False
    filename = f'audio_{n}.wav'
    with wave_file(filename) as wf:
      async for response in session.receive():
        logger.debug(str(response))
        if response.server_content and response.server_content.model_turn and response.server_content.model_turn.parts and hasattr(response.server_content.model_turn.parts[0], 'text'):
          if text := response.server_content.model_turn.parts[0].text:
            display(Markdown(data=text))
            continue

        if response.server_content and response.server_content.model_turn and response.server_content.model_turn.parts and hasattr(response.server_content.model_turn.parts[0], 'data'):
          if data := response.server_content.model_turn.parts[0].data:
            print('.', end='')
            wf.writeframes(data)
            audio = True
            continue

        server_content = response.server_content
        if server_content is not None:
          handle_server_content(wf, server_content)
          continue

        tool_call = response.tool_call
        if tool_call is not None:
          await handle_tool_call(session, tool_call)


  if audio:
    display(Audio(filename, autoplay=True))
    n = n+1

由于本教程演示了多种工具，你需要更多代码来处理它返回的不同类型的对象。

def handle_server_content(wf, server_content):
  model_turn = server_content.model_turn
  if model_turn:
    for part in model_turn.parts:
      executable_code = part.executable_code
      if executable_code is not None:
        display(Markdown('-------------------------------'))
        display(Markdown(f'``` python\n{executable_code.code}\n```'))
        display(Markdown('-------------------------------'))

      code_execution_result = part.code_execution_result
      if code_execution_result is not None:
        display(Markdown('-------------------------------'))
        display(Markdown(f'``` \n{code_execution_result.output}\n```'))
        display(Markdown('-------------------------------'))

  grounding_metadata = getattr(server_content, 'grounding_metadata', None)
  if grounding_metadata is not None:
    display(
        HTML(grounding_metadata.search_entry_point.rendered_content))

  return

async def handle_tool_call(session, tool_call):
  print("Tool call:")
  function_responses = []
  for fc in tool_call.function_calls:
    function_response = types.FunctionResponse(
        id=fc.id,
        name=fc.name,
        response={"result": "ok"},
    )
    function_responses.append(function_response)
  print('>>> ', function_responses)
  await session.send_tool_response(function_responses=function_responses)

第一次尝试运行它：

await run(prompt="Hello?", tools=None, modality = "TEXT")

简单函数调用

API 的函数调用功能可以处理各种各样的函数。SDK 中的支持仍在构建中。因此，保持简单，只需发送一个最小的函数定义：仅函数名称。

请注意，在实时 API 中，函数调用与聊天轮次是独立的。在处理函数调用时，对话可以继续。

turn_on_the_lights = {'name': 'turn_on_the_lights'}
turn_off_the_lights = {'name': 'turn_off_the_lights'}

prompt = "Turn on the lights"

tools = [
    {'function_declarations': [turn_on_the_lights, turn_off_the_lights]}
]

await run(prompt, tools=tools, modality = "TEXT")

异步函数调用

异步函数调用 允许模型异步管理其函数调用，而不会阻塞用户输入。

您可以决定模型在函数调用结束时的行为：保持沉默、中断当前操作或等待完成当前任务。

接下来的单元格将使用稍微更新的代码来使用 Live API，以便会话保持打开 20 秒，并接受每 10 秒发送给模型的多个请求。如果您对此实现感到好奇，请展开下一个单元格。

# @title Live class with multiple messages (just run this cell)

import collections.abc
import inspect
from asyncio.exceptions import CancelledError
import traceback

class Live:
  def __init__(self, client):
    self.client = client


  async def run(self, config, functions=None, messages=None):
    self.config = config
    self.send_queue = asyncio.Queue()
    self.tool_call_queue = asyncio.Queue()

    try:
      async with (
            client.aio.live.connect(model=MODEL_ID, config=config) as session,
            asyncio.TaskGroup() as tg
      ):
        self.session = session
        recv_task = tg.create_task(self._recv())
        send_task = tg.create_task(self._send())
        tool_call_task = tg.create_task(self._run_tool_calls(functions))
        read_text= tg.create_task(self._read_text(messages))


        await read_text
        await asyncio.sleep(20) # Keeping the socket open for 20s to wait for the FC and different messages

        raise CancelledError
    except CancelledError:
      pass
    except ExceptionGroup as EG:
      traceback.print_exception(EG)

  async def _recv(self):
    try:
      mode = None
      while True:
        async for response in self.session.receive():
          logger.debug(str(response))
          if response.server_content and response.server_content.model_turn and response.server_content.model_turn.parts and hasattr(response.server_content.model_turn.parts[0], 'text'):
            if text := response.server_content.model_turn.parts[0].text:
              if mode != 'text':
                mode = 'text'
                print()
              print(text)
          else:
            if mode == 'text':
              mode = 'other'
              print()
            print(f'<<<  {response.model_dump_json(exclude_none=True)}\n')

          tool_call = response.tool_call
          if tool_call is not None:
            await self.tool_call_queue.put(tool_call)

    except asyncio.CancelledError:
      pass

  async def _send(self):
    while True:
      msg = await self.send_queue.get()
      print(f'>>> {repr(msg)}\n')
      await self.session.send_realtime_input(text=msg['parts'][0]['text'])

  async def _run_tool_calls(self, functions):
    while True:
      tool_call = await self.tool_call_queue.get()
      for fc in tool_call.function_calls:
        fun = functions[fc.name]
        called = fun(**fc.args)
        if inspect.iscoroutine(called):
          print(f'>> Starting {fc.name}\n')
          result = await called
          print(f'>> Done {fc.name} >>> {repr(result)}\n')
          result = self._wrap_function_result(fc, result)
          await self.session.send_tool_response(function_responses=[result])
        elif isinstance(called, collections.abc.AsyncIterable):
          async for result in called:
            result.will_continue=True
            result = self._wrap_function_result(fc, result)
            print(f">>> {repr(result)}\n")
            await self.session.send_tool_response(function_responses=[result])

          result = self._wrap_function_result(
              fc,
              types.FunctionResponse(will_continue=False)
          )
          print(f">>> {repr(result)}\n")
          await self.session.send_tool_response(
              function_responses=[result]
          )


        else:
          raise TypeError(f"expected {fc.name} to return a coroutine, or an "
                          f"AsyncIterable, got {type(fun)}")

  def _wrap_function_result(self, fc, result):
    if result is None:
      return types.FunctionResponse(
          name=fc.name,
          id=fc.id,
          response={'result': 'ok'}
      )
    elif isinstance(result, types.FunctionResponse):
      result.name = fc.name
      result.id = fc.id
      return result
    else:
      return types.FunctionResponse(
          name=fc.name,
          id=fc.id,
          response= {'result': result}
      )

  async def _read_text(self, messages):
    if messages:
        for n, message in enumerate(messages):
            await self.send_queue.put({
                'role': 'user',
                'parts': [{'text': message}]
            })
            if n+1 < len(messages):
              await asyncio.sleep(5)
    else:
        while True:
            message = await asyncio.to_thread(input, "message > ")
            if message.lower() == "q":
                break
            await self.send_queue.put({
                'role': 'user',
                'parts': [{'text': message}]
            })

默认行为：阻塞

我们先从默认行为开始。首先定义一个模拟天气函数，通过等待10秒来模拟计算时间。

默认行为类似于FIFO队列：函数调用被添加到队列中，任何后续请求都会在其后面排队（被阻塞），直到它处理完成。

# Mock function, takes 10s to process
async def get_weather_vegas():
  await asyncio.sleep(10)
  return {'weather': "Sunny, 42 degrees"}

# multiple prompts, they are going to be asked with 5s delay between each of them.
questions = [
    "What's the weather in Vegas?",
    "In the meantime tell me about the Paris casino"
]

await Live(client).run(
    messages=questions,
    functions={
        'get_weather_vegas': get_weather_vegas,
    },
    config={
        "response_modalities": ["TEXT"],
        "tools": [
            {
                'function_declarations': [
                    {'name': 'get_weather_vegas',  "behavior": "UNSPECIFIED"}, # This is default behavior, equivalent to BLOCKING
                ]
            }
        ]
    }
)

如你所见，模型立即调用了 get_weather_vegas 函数，但第二个问题被忽略了，因为模型仍在等待函数调用的结果。它只有在回答函数调用之后才开始回答第二个问题。

中断：停止当前操作并处理此结果

这次，behavior 设置为 NON_BLOCKING，这意味着它将使用异步函数调用。

当您这样做时，您需要定义模型在获得函数调用结果后将执行的操作。这可以在函数内部管理，或者在处理函数调用的脚本中管理（因为自动函数调用不可用），方法是在 FunctionResponse 中添加一个 scheduling 值。

这次，scheduling 行为是“Interrupt”，这意味着一旦收到响应，模型将停止当前正在说的内容，并立即处理该响应。

# Mock function, takes 10s to process
async def get_weather_vegas():
  await asyncio.sleep(10)
  return types.FunctionResponse(
      response={'weather': "Sunny, 42 degrees"},
      scheduling="INTERRUPT"
  )

# multiple prompts, they are going to be asked with 5s delay between each of them.
questions = [
    "What's the weather in Vegas?",
    "In the meantime tell me what you know about the Paris casino and all there's to do and see in it. Then continue to tell me about the Vegas casinos until I tell you to stom talking. Don't ask me, just talk non-stop"
    "Then can you tell me what's your favorite cirque du soleil show?"
]

await Live(client).run(
    messages=questions,
    functions={
        'get_weather_vegas': get_weather_vegas,
    },
    config={
        "response_modalities": ["TEXT"],
        "tools": [
            {
                'function_declarations': [
                    {'name': 'get_weather_vegas',  "behavior": "NON_BLOCKING"},
                ]
            }
        ]
    }
)

如你所见，这次模型通过说类似“拉斯维加斯的天气请求正在运行。完成后我会告诉你”这样的话来确认了我们的请求，然后继续处理你所要求的内容，接着当函数响应返回时，它停下了手头的事情，告诉我们天气情况，然后又继续谈论它之前的话题。

等待空闲：先完成手头工作，再处理这个结果

同样，behavior 被设置为 NON_BLOCKING，这意味着它将使用异步函数调用，你需要在 FunctionResponse 中添加一个 scheduling 值。

这次 scheduling 的行为是“When_idle”，这意味着模型将等待它说完，然后才告诉我们你要求的内容。

import time

# Mock function, takes 6s to process
async def get_weather_vegas():
  await asyncio.sleep(6)
  return types.FunctionResponse(
      response={'weather': "Sunny, 42 degres"},
      scheduling="WHEN_IDLE"
  )

# multiple prompts, they are going to be asked with 5s delay between each of them.
questions = [
    "What's the weather in Vegas?",
    "In the meantime, without using tools, tell me what you know about the Paris casino and all there's to do and see in it. Tell me about each casino on the strip!"
]

await Live(client).run(
    messages=questions,
    functions={
        'get_weather_vegas': get_weather_vegas,
    },
    config={
        "response_modalities": ["TEXT"],
        "tools": [
            {
                'function_declarations': [
                    {'name': 'get_weather_vegas',  "behavior": "NON_BLOCKING"},
                ]
            }
        ]
    }
)

如你所见，这一次，即使它在回答关于赌场的问题时收到了函数调用的响应（参见 >> Done get_weather_vegas >>> [...] response={'weather': 'Sunny, 42 degres'}) 这一行），它也等到当前回答结束后才告知天气情况。

静默：将所学留给自己

这次，behavior 再次设置为 NON_BLOCKING，这意味着它将使用异步函数调用，并且需要在 FunctionResponse 中提供 scheduling 值。

这次 scheduling 行为是“Silent”，这意味着模型不会告诉你函数调用何时完成，但它可能仍会在后续对话中使用该知识。

import time

# Mock function, takes 5s to process
async def get_weather_vegas():
  time.sleep(10)
  return types.FunctionResponse(
      response={'weather': "Sunny, 42 degres"},
      scheduling="SILENT"
  )

# multiple prompts, they are going to be asked with 5s delay between each of them.
questions = [
    "What's the weather in Vegas?",
    "In the meantime tell me about the Paris casino.",
    "Is the temperature over 40 degres?"
]

await Live(client).run(
    messages=questions,
    functions={
        'get_weather_vegas': get_weather_vegas,
    },
    config={
        "response_modalities": ["TEXT"],
        "tools": [
            {
                'function_declarations': [
                    {'name': 'get_weather_vegas',  "behavior": "NON_BLOCKING"},
                ]
            }
        ]
    }
)

这次，如你所见，当函数调用结束时模型没有做任何事，但当再次被问及同一件事时，它没有进行新的函数调用就回答了。

代码执行

code_execution 允许模型编写并运行 Python 代码。尝试用它解决一个模型无法凭记忆解答的数学问题：

prompt="Can you compute the largest prime palindrome under 100000."

tools = [
    {'code_execution': {}}
]

await run(prompt, tools=tools, modality="AUDIO")

组合式函数调用

组合式函数调用是指将用户自定义函数与 code_execution 工具结合使用的能力。模型会将它们编写成更大的代码块，然后暂停执行，等待您为每个调用发送回响应。

prompt="Can write some code to loop through and print integers from 1-20, and every time you hit a multiple of 3 turn on the lights, and every time you hit a multiple of 5 turn them off?"

tools = [
    {'code_execution': {}},
    {'function_declarations': [turn_on_the_lights, turn_off_the_lights]}
]

await run(prompt, tools=tools, modality="AUDIO")

Google 搜索

google_search 工具允许模型进行谷歌搜索。例如，尝试询问它关于训练数据中未包含的近期事件。

在 AUDIO 模式下，搜索仍会执行，但您不会看到详细结果：

prompt="When the latest Brazil vs. Argentina soccer match happened and what was the final score?"

tools = [
   {'google_search': {}}
]

await run(prompt, tools=tools, modality="AUDIO")

多工具

然而，新API最大的不同在于，你不再局限于每次请求只能使用一个工具。尝试将前面章节中的任务组合起来：

prompt = """\
  Hey, I need you to do three things for me.

  1. Then compute the largest prime plaindrome under 100000.
  2. Then use google search to lookup unformation about the largest earthquake in california the week of Dec 5 2024?
  3. Turn on the lights

  Thanks!
  """

tools = [
    {'google_search': {}},
    {'code_execution': {}},
    {'function_declarations': [turn_on_the_lights, turn_off_the_lights]}
]

await run(prompt, tools=tools, modality="AUDIO")

后续步骤

或者从 Cookbook 中查看 Gemini 2.5 的其他功能，特别是这个多工具示例以及关于 Gemini 空间能力 的示例。

常见问题

常见错误：1) API 密钥未设置或无效，导致认证失败；2) 模型 ID 错误或不可用；3) 工具定义格式不正确，导致模型无法识别；4) 异步函数调用未正确处理 FunctionResponse 的 scheduling 字段，导致行为不符合预期；5) 音频输出在 Colab 中无法播放，需检查 WAV 文件生成。排查方法：启用日志记录（设置 logger.setLevel('DEBUG')）查看详细响应；检查 API 响应中的错误信息；确保使用 v1alpha 版本；验证工具名称和参数匹配。生产环境优化：使用连接池管理 WebSocket；实现重试机制；对工具调用结果进行缓存；监控 API 使用量以控制成本。

企业应用扩展

进一步学习可参考官方 SDK 文档和 WebSocket 底层实现。探索更复杂的工具组合，如将代码执行与搜索结合进行实时数据分析。尝试使用 google_search 的 grounding 元数据增强回答的可信度。考虑将 Live API 集成到语音助手或实时交互应用中，利用异步函数调用实现流畅的用户体验。关注 Gemini 2.5 的空间理解能力，结合工具实现多模态交互。

总结

本教程面向有编程基础的开发者，深入讲解 Gemini 多模态 Live API 的工具使用，包括函数调用、代码执行和 Google 搜索。通过实际代码示例，展示如何配置工具、处理异步函数调用（阻塞、中断、等待空闲、静默模式），以及组合使用多个工具。教程强调代码执行作为核心工具，支持在单个 API 调用中集成多种工具，并提供了环境准备、实践步骤和常见问题排查，帮助开发者快速上手并应用于生产环境。

内容来源与验证

继续系统学习

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

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

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