Gemini の機能を構成する

このドキュメントでは、Gemini Live API を使用する際に Gemini モデルのさまざまな機能を構成する方法について説明します。関数呼び出しやグラウンディングなどのツールの使用、アフェクティブ ダイアログやプロアクティブ音声などのライブ音声機能を構成できます。

ツールの使用を構成する

さまざまなツールが Gemini Live API 対応モデルのさまざまなバージョンと互換性があります。次に例を示します。

返されたレスポンスで使用する特定のツールを有効にするには、モデルを初期化するときに tools リストにツールの名前を含めます。以降のセクションでは、コードで各組み込みツールを使用する方法の例を示します。

重要: Gemini API は、同じ設定リクエストで検索ツール(googleSearch など)と検索以外のツール(関数呼び出しや Gemini Enterprise Agent Platform RAG Engine 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 を使用して結果を返す必要があります。
  • Agent Development Kit(ADK)を使用する: ADK の McpToolset クラスを使用します。このクラスには、Gemini Live API 用の組み込み MCP 統合が用意されています。ADK は MCP サーバーからツールを自動的に検出し、スキーマを関数宣言に変換し、双方向ストリーム中に MCP ツール呼び出しを実行して、ツール レスポンスをモデルに送り返します。

ADK で MCP ツールを使用する

ADK を使用して MCP ツールを Gemini Live API に接続する手順は次のとおりです。

  1. mcp エクストラを使用して google-adk パッケージをインストールします。

    pip install "google-adk[mcp]"
    
  2. StreamableHTTPConnectionParams を使用して McpToolset を MCP サービスに接続し、runner.run_live() を使用してイベントをストリーミングする live_mcp_example.py という名前のスクリプトを作成します。

    次のように置き換えます。

    • PROJECT_ID: 実際の Cloud de Confiance by S3NS プロジェクト ID。
    • MCP_SERVER_URL: Streamable HTTP MCP サーバーの URL(例: 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 が適切にバックチャネル(適切な割り込みが処理される)を処理しやすくなります。
  • 音声を一緒に聴く: Gemini は、発言者の音声ではない音声ファイルを一緒に聴き、会話の後半でその音声ファイルに関する質問に後から答えることができます。

課金

Gemini が会話をリッスンしている間は、入力音声トークンが課金されます。

出力音声トークンについては、Gemini が応答した場合にのみ課金されます。Gemini が応答しない場合や無音状態が続く場合、出力音声トークンに対する課金は発生しません。

詳細については、Agent Platform の料金をご覧ください。

次のステップ

Gemini Live API の使用について詳しくは、以下をご覧ください。