はじめに
日本時間 11/19 に Gemini 3 Pro がリリースされました!
Gemini 3 Pro から Streaming function call arguments 機能が追加されました。
実際に従来の挙動との違いを試してみました!
Streaming function call arguments とは
通常、Geminiがツールを実行しようとする際、以下のようなJSONを生成します。
{
"name": "save_article",
"args": {
"title": "...",
"content": "非常に長い本文..."
}
}
従来は、このJSONが「閉じカッコ」まで完成するのを待ってから、SDKがクライアントにデータを渡していました。
しかし、新機能のフラグを有効にすることで、JSONの断片(Partial Arguments)がリアルタイムに届くようになります。
関数を呼び出す際に発生するレイテンシを軽減するのに役立ちます。
実装方法
FunctionCallingConfig に stream_function_call_arguments=True を設定するだけです。
config=types.GenerateContentConfig(
tools=[tool_config],
tool_config=types.ToolConfig(
function_calling_config=types.FunctionCallingConfig(
mode=types.FunctionCallingConfigMode.AUTO,
# これをTrueにする
stream_function_call_arguments=True,
)
),
)
検証
「2000文字以上の記事を書いて保存するツール」をGeminiに実行させ、ストリーミングのON/OFFでどう挙動が変わるか確認してみました。
import time
from google import genai
from google.genai import types
client = genai.Client(
vertexai=True, project=PROJECT_ID, location="global"
)
save_article_declaration = types.FunctionDeclaration(
name="save_article",
description="詳細な長文記事をデータベースに保存します。",
parameters={
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "記事のタイトル"
},
"content": {
"type": "string",
"description": "記事の本文すべて。非常に詳細に記述し、少なくとも2000文字以上の長さが必要です。"
}
},
"required": ["title", "content"],
},
)
tool_config = types.Tool(function_declarations=[save_article_declaration])
# 検証用関数
def run_experiment(stream_args: bool):
mode_str = "ON (ストリーミング)" if stream_args else "OFF (一括受信)"
print(f"\n🔴 実験開始: 引数ストリーミング = {mode_str}")
print("-" * 60)
start_time = time.perf_counter()
chunk_count = 0
response_stream = client.models.generate_content_stream(
model="gemini-3-pro-preview",
contents="インターネットの歴史について、非常に詳細な記事(2000文字以上)を執筆し、ツールを使って保存してください。",
config=types.GenerateContentConfig(
tools=[tool_config],
tool_config=types.ToolConfig(
function_calling_config=types.FunctionCallingConfig(
mode=types.FunctionCallingConfigMode.AUTO,
stream_function_call_arguments=stream_args,
)
),
),
)
for chunk in response_stream:
if chunk.function_calls:
fc = chunk.function_calls[0]
chunk_count += 1
# ストリーミングONの場合: partial_args (断片) を確認
if stream_args and fc.partial_args:
for arg in fc.partial_args:
value = arg.string_value if arg.string_value else ""
display_value = value.replace("\n", "\\n")
print(f"⏱️ チャンク {chunk_count:03}: Path='{arg.json_path}', 文字数={len(value)}, 内容='{display_value}'")
# ストリーミングOFFの場合: args (完成品) を確認
elif not stream_args and fc.args:
title = fc.args.get('title', 'No Title')
content = fc.args.get('content', '')
print(f"📦 完成データ受信: タイトル='{title}'")
print(f"📄 本文文字数: {len(content)}文字")
print(f"📝 本文冒頭: {content[:50]}...")
end_time = time.perf_counter()
print("-" * 60)
print(f"✅ 完了タイム: {end_time - start_time:.4f}秒 / 受信パケット数: {chunk_count}")
# --- 実行 ---
# 1. ストリーミング ON
run_experiment(stream_args=True)
# 2. ストリーミング OFF
run_experiment(stream_args=False)
ストリーミング OFF の場合
通常通り、全て完了するまでの約30秒間は何も表示されなかったです。
🔴 実験開始: 引数ストリーミング = OFF (一括受信) ------------------------------------------------------------ 📦 完成データ受信: タイトル='インターネットの全史:冷戦の産物から人類の神経系へ' 📄 本文文字数: 4214文字 📝 本文冒頭: インターネットの全史:冷戦の産物から人類の神経系へ 序章:デジタル社会の基盤 今日、私たちの生活... ------------------------------------------------------------ ✅ 完了タイム: 29.1465秒 / 受信パケット数: 1
ストリーミング ON の場合
実行直後から、以下のようにテキストが生成されていく様子を確認できました。
ツール引数の中身が流れてきます。
🔴 実験開始: 引数ストリーミング = ON (ストリーミング) ------------------------------------------------------------ ⏱️ チャンク 002: Path='$.content', 文字数=125, 内容='インターネットの壮大な歴史:軍事防衛から世界を変えるインフラへ\n\n序章:現代の神経系\n\n今日、私たちが「インターネット」と呼ぶこの巨大なネットワークは、電気や水道と同じような生活に不可欠なインフラストラクチャとなっています。スマートフォンで瞬時に地球' ⏱️ チャンク 003: Path='$.content', 文字数=105, 内容='の裏側のニュースを知り、ビデオ会議で遠隔地の人々と顔を合わせ、膨大な知識のデータベースにアクセスする。これらは今や日常の光景ですが、その起源を辿ると、冷戦時代の軍事的な緊張と、科学者たちの純粋な知的好奇心が交差' ・・・ ⏱️ チャンク 035: Path='$.content', 文字数=24, 内容='。その歴史はまだ書き続けられている途中なのです。' ⏱️ チャンク 036: Path='$.content', 文字数=0, 内容='' ⏱️ チャンク 037: Path='$.title', 文字数=31, 内容='インターネットの壮大な歴史:軍事防衛から世界を変えるインフラへ' ⏱️ チャンク 038: Path='$.title', 文字数=0, 内容='' ------------------------------------------------------------ ✅ 完了タイム: 39.4254秒 / 受信パケット数: 39
Path=の部分で引数の「どこ」に当たるデータなのかがパス($.content)で示されます。
まとめ
Geminiの Streaming function call arguments は、Function Calling の「生成待ち時間の長さ(空白時間)」を解消する機能となります。
ただし、受け取るデータが断片化されるため、クライアント側での結合処理の実装が必要になると思います。
リアルタイム性を重視する場合や長い引数生成をする際は、このストリーミング機能を使うという使い分けをしていくのかなと思いました。