配置 Gemini 功能

本文档介绍了如何在使用 Gemini Live API 时配置 Gemini 模型的各种功能。您可以配置工具使用（例如函数调用和接地）以及实时音频功能（例如共情对话和主动音频）。

配置工具使用

多种工具与支持 Gemini Live API 的各种版本的模型兼容，包括：

如需启用特定工具以在返回的回答中使用，请在初始化模型时将该工具的名称添加到 tools 列表中。以下部分提供了有关如何在代码中使用每个内置工具的示例。

重要提示： Gemini API 不支持在同一设置请求中将搜索工具（例如 googleSearch）与非搜索工具（例如函数调用或 Gemini Enterprise Agent Platform RAG 引擎 retrieval 工具）相结合。仅当多个工具都是搜索工具时，才支持使用多个工具。

函数调用

如果您希望模型与您管理的外部系统或 API 互动，请使用 函数调用。此功能适用于检查数据库、发送电子邮件或与自定义 API 互动等任务。 Gemini Live API 支持异步函数调用。如需了解详情， 请参阅使用 Gemini Live API 进行异步函数调用。

模型会生成函数调用，您的应用会执行代码并将结果发送回模型。

所有函数都必须在会话开始时声明，方法是将工具定义作为 LiveConnectConfig 消息的一部分发送。

如需启用函数调用，请在设置消息的 tools 列表中添加 function_declarations：

Python

import asyncio

from google import genai
from google.genai.types import (
    Content,
    LiveConnectConfig,
    Part,
)

# Initialize the client.
client = genai.Client(
    enterprise=True,
    project="GOOGLE_CLOUD_PROJECT",  # Replace with your project ID
    location="LOCATION",  # Replace with your location
)

MODEL_ID = "gemini-3.8-live"


def get_current_weather(location: str) -> str:
    """Example method. Returns the current weather.

    Args:
        location: The city and state, e.g. San Francisco, CA
    """
    weather_map: dict[str, str] = {
        "Boston, MA": "snowing",
        "San Francisco, CA": "foggy",
        "Seattle, WA": "raining",
        "Austin, TX": "hot",
        "Chicago, IL": "windy",
    }
    return weather_map.get(location, "unknown")


async def main():
    config = LiveConnectConfig(
        response_modalities=["AUDIO"],
        tools=[get_current_weather],
    )

    async with client.aio.live.connect(
        model=MODEL_ID,
        config=config,
    ) as session:
        text_input = "Get the current weather in Boston."
        print(f"Input: {text_input}")

        await session.send_realtime_input(text=text_input)
        )

        async for message in session.receive():
            if message.tool_call:
                function_responses = []
                for function_call in message.tool_call.function_calls:
                    print(f"FunctionCall > {function_call}")
                    # Execute the tool and send the response back to the model.
                    result = get_current_weather(**function_call.args)
                    function_responses.append(
                        {
                            "name": function_call.name,
                            "response": {"result": result},
                            "id": function_call.id,
                        }
                    )
                if function_responses:
                    await session.send_tool_response(function_responses=function_responses)


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

如需查看在系统说明中使用函数调用的示例，请参阅 最佳实践示例。

Model Context Protocol (MCP) 工具

Google Gen AI SDK 不会自动在 Gemini Live API 会话中执行 Model Context Protocol (MCP) 工具调用。如需在 Gemini Live API 会话中使用 MCP 工具，您可以使用以下方法之一：

  • 直接管理 MCP 工具调用：如果您直接使用 Google Gen AI SDK ，则您的应用必须管理 MCP 客户端会话，在模型请求时执行工具 调用，并使用 send_tool_response返回结果。
  • 使用智能体开发套件 (ADK)：使用 ADK 中的 McpToolset类，该类为 Gemini Live API 提供了内置的 MCP 集成。ADK 会自动从 MCP 服务器发现工具，将其架构转换为函数声明，在双向流期间运行 MCP 工具调用，并将工具响应发送回模型。

将 MCP 工具与 ADK 结合使用

您可以使用 ADK 按照以下步骤将 MCP 工具连接到 Gemini Live API：

  1. 使用 mcp extra 安装 google-adk 软件包：

    pip install "google-adk[mcp]"

  2. 创建一个名为 live_mcp_example.py 的脚本，该脚本使用 StreamableHTTPConnectionParams 将 McpToolset 连接到 MCP 服务，并使用 runner.run_live() 流式传输事件。

    替换以下内容：

    • PROJECT_ID：您的 Cloud de Confiance by S3NS 项目 ID。
    • MCP_SERVER_URL：您的可流式 HTTP MCP 服务器网址（例如 http://127.0.0.1:8765/mcp）。
    import asyncio
import contextlib
import os

from google.adk.agents import Agent, LiveRequestQueue, RunConfig
from google.adk.agents.run_config import StreamingMode
from google.adk.runners import InMemoryRunner
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams
from google.genai import types

os.environ.setdefault("GOOGLE_GENAI_USE_ENTERPRISE", "TRUE")
os.environ.setdefault("GOOGLE_CLOUD_PROJECT", "PROJECT_ID")
os.environ.setdefault("GOOGLE_CLOUD_LOCATION", "us-central1")

async def main() -> None:
  # 1. Connect to the Streamable HTTP MCP service
  mcp_toolset = McpToolset(
      connection_params=StreamableHTTPConnectionParams(
          url="MCP_SERVER_URL",
      )
  )

  # 2. Create the ADK Agent with a Gemini Live model and the MCP toolset
  agent = Agent(
      name="mcp_live_agent",
      model="gemini-3.8-live",
      instruction="You are a helpful assistant. Use the provided MCP tools when needed.",
      tools=[mcp_toolset],
  )

  runner = InMemoryRunner(agent=agent, app_name="mcp_live_demo")
  session = await runner.session_service.create_session(
      app_name="mcp_live_demo",
      user_id="user_1",
  )

  # 3. Configure bidirectional streaming and send a prompt
  live_request_queue = LiveRequestQueue()
  run_config = RunConfig(
      streaming_mode=StreamingMode.BIDI,
      response_modalities=[types.Modality.AUDIO],
      output_audio_transcription=types.AudioTranscriptionConfig(),
  )

  live_request_queue.send_content(
      types.Content(
          role="user",
          parts=[types.Part.from_text(text="What is the weather in Paris?")],
      )
  )

  # 4. Stream Live events (ADK automatically executes MCP tools)
  awaiting_post_tool_response = False
  saw_function_response = False
  has_model_output = False

  try:
    async with contextlib.aclosing(
        runner.run_live(
            user_id="user_1",
            session_id=session.id,
            live_request_queue=live_request_queue,
            run_config=run_config,
        )
    ) as live_events:
      async for event in live_events:
        for call in event.get_function_calls():
          print(f"[MCP Tool Call] {call.name}({call.args})")
          awaiting_post_tool_response = True
          saw_function_response = False
          has_model_output = False

        for resp in event.get_function_responses():
          print(f"[MCP Tool Response] {resp.name} -> {resp.response}")
          saw_function_response = True

        if event.output_transcription and event.output_transcription.text:
          if not awaiting_post_tool_response or saw_function_response:
            has_model_output = True
            awaiting_post_tool_response = False
          if event.output_transcription.finished:
            print(f"[Transcript] {event.output_transcription.text}")

        # Wait for the final response turn after tool execution completes
        if event.turn_complete:
          if awaiting_post_tool_response or not has_model_output:
            continue
          live_request_queue.close()
          break
  finally:
    await mcp_toolset.close()
    await runner.close()

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

如果您希望模型通过将回答锚定到可验证的信息来源来提供更准确、更具真实性的回答，请使用“依托 Google 搜索进行接地”。此功能适用于在网络上搜索等任务。

与函数调用不同，服务器端集成会自动处理信息的检索。

如需启用“依托 Google 搜索进行接地”，请在设置消息的 tools 列表中添加 google_search：

Python

import asyncio

from google import genai
from google.genai.types import (
    Content,
    LiveConnectConfig,
    Part,
)

# Initialize the client.
client = genai.Client(
    enterprise=True,
    project="GOOGLE_CLOUD_PROJECT",  # Replace with your project ID
    location="LOCATION",  # Replace with your location
)

MODEL_ID = "gemini-3.8-live"


async def main():
    config = LiveConnectConfig(
        response_modalities=["AUDIO"],
        tools=[{"google_search": {}}],
    )

    async with client.aio.live.connect(
        model=MODEL_ID,
        config=config,
    ) as session:
        text_input = "What is the current weather in Toronto, Canada?"
        print(f"Input: {text_input}")

        await session.send_realtime_input(text=text_input)

        async for message in session.receive():
            # Consume the messages from the model.
            # In native audio, the model response is in audio format.
            pass


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

配置共情对话

共情对话配置仅适用于 gemini-live-2.5-flash-native-audio。本部分不适用于 gemini-3.8-live。

启用共情对话 后，模型会尝试根据用户的语音语调和情感表达来理解和回应。

如需启用共情对话，请在设置消息中将 enable_affective_dialog 设置为 true：

Python

config = LiveConnectConfig(
    response_modalities=["AUDIO"],
    enable_affective_dialog=True,
)

配置主动音频

主动音频配置仅适用于 gemini-live-2.5-flash-native-audio。gemini-3.8-live 中包含主动音频。

借助主动音频，您可以控制模型何时响应。例如，您可以问问 Gemini，让它仅在收到提示或讨论特定主题时才做出响应。如需观看 主动音频的视频演示，请参阅 Gemini LiveAPI 原生音频预览。

如需启用主动音频，请在设置消息中配置 proactivity 字段，并将 proactive_audio 设置为 true：

Python

config = LiveConnectConfig(
    response_modalities=["AUDIO"],
    proactivity=ProactivityConfig(proactive_audio=True),
)

对话示例

以下示例展示了与 Gemini 进行烹饪相关对话的可能情形：

Prompt: "You are an AI assistant in Italian cooking; only chime in when the topic is about Italian cooking."

Speaker A: "I really love cooking!" (No response from Gemini.)

Speaker B: "Oh yes, me too! My favorite is French cuisine." (No response from
Gemini.)

Speaker A: "I really like Italian food; do you know how to make a pizza?"

(Italian cooking topic triggers response from Gemini.)
Gemini Live API: "I'd be happy to help! Here's a recipe for a pizza."

常见使用场景

使用主动音频时，Gemini 的行为如下：

  • 以最小延迟响应：Gemini 会在用户 说完话后做出响应，从而减少中断，并帮助 Gemini 在发生中断时不会丢失 上下文。
  • 避免中断：主动音频可帮助 Gemini 避免受到背景噪音或外部闲聊的干扰， 并防止 Gemini 在对话期间引入外部闲聊时做出响应。
  • 处理中断：如果用户需要在 Gemini 回答期间中断，主动音频可让 Gemini 更轻松地进行适当的后向通道处理（即处理适当的中断），而不是像用户使用填充词（例如 umm 或 uhh）时那样。
  • 共同聆听音频：Gemini 可以共同聆听不是说话者声音的音频文件，然后在对话的后续部分回答有关该音频文件的问题。

结算

在 Gemini 聆听对话时，系统会收取输入音频 token 费用。

对于输出音频 token，只有在 Gemini 回答时才会收费。如果 Gemini 不回应或保持静默，则不会收取输出音频 token 的费用。

如需了解详情，请参阅 Agent Platform 价格。

后续步骤

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