gemini-robotics-er-2-streaming-preview モデル エンドポイントは、Live
API と統合された専用の
ストリーミング エンドポイントを公開し、アプリケーションとロボット間のリアルタイムの
双方向通信を可能にします。これにより、迅速なフィードバック ループと環境への反応型レスポンスを必要とするエージェントに適しています。
ユースケース
- マルチロボット連携: 共有セッションを通じてタスクの状態を通信し、 サブタスクを委任する複数のロボット。
- 継続的なモニタリング: シーンを観察し、コンテナが満杯になるなど、特定のイベントが発生したときにアクションをトリガーするロボット。
- 倉庫と物流: 品物を 目視で確認し、梱包の進捗状況を追跡し、エラーから復旧するピッキングと梱包のエージェント。
技術仕様
次の表に、Live API の技術仕様の概要を示します。
| カテゴリ | 詳細 |
|---|---|
| 入力モダリティ | 音声（RAW 16 ビット PCM 音声、16kHz、リトル エンディアン）、画像（JPEG <= 1FPS）、テキスト |
| 出力モダリティ | テキスト |
| プロトコル | ステートフル WebSocket 接続（WSS） |
エージェントの設定を構築する
Live API で構築されたすべてのロボット工学エージェントは、次の 3 つのステップに従います。
- ロボットの機能をツールとして宣言する。ロボットが実行できる各アクション（移動、把握、発話）は、名前、説明、パラメータ スキーマを持つ関数宣言になります。物理アクションでは、
"behavior": "BLOCKING"を使用する必要があります。これにより、ロボットが完了するまでモデルが待機してから 次のステップを選択します。 - マルチモーダル入力を永続的なセッションにストリーミングする。
live.connectセッションを開き、タスクの実行中は開いたままにします。ロボットのセンサーから到着した動画フレーム、音声、テキストを送信します。 - 受信ループでツール呼び出しを処理する。モデルがアクションを選択するたびに、
tool_callメッセージが送信されます。受信ループは、ロボット SDK に対して関数を実行し、tool_responseを返します。セッションは開いたままになり、モデルは結果に基づいて次のアクションを選択します。
以降のセクションでは、これらのステップを 3 つの一般的なパターン（ベースライン エージェント ループ、ハートビートを使用したプロアクティブなシーン モニタリング、ツールとしての TTS を介した音声のルーティング）に適用する方法について説明します。
関数呼び出しでロボットをオーケストレートする
次の例は、3 つのステップすべてが 1 つの Python スクリプトにまとめられています。
ステップ 1（ツールの定義）では、ロボットの機能が関数宣言として宣言されます。navigate 関数は "behavior": "BLOCKING" を使用するため、
モデルはロボットがウェイポイントに到達するまで待機してから別のツールを呼び出します。
同じリストに関数宣言を追加して、ロボットの追加機能を公開します。
ステップ 2（入力ヘルパー）では、さまざまなモダリティ入力をセッションにストリーミングする 3 つの関数を示します。send_text はコマンド用、send_image はカメラフレーム用（オプションのテキスト プロンプト付き）、send_audio はマイクからの RAW PCM 音声用です。
ステップ 3（受信ループ）は同時に実行され、server_content メッセージ（モデルのテキスト出力）と tool_call メッセージ（ロボット アクションをリクエストするモデル）の 2 種類のメッセージを処理します。ツール呼び出しが到着すると、ループは execute_tool（実際のロボット SDK に置き換えるスタブ）を呼び出し、tool_response を返して、モデルが次のアクションを選択できるようにします。
import asyncio
from google import genai
from google.genai import types
MODEL = "gemini-robotics-er-2-streaming-preview"
# ── Tool definitions ─────────────────────────────────────────────────────────
tools = [
{
"function_declarations": [
{
"name": "navigate",
"description": "Navigate the robot to a named waypoint.",
"behavior": "BLOCKING",
"parameters": {
"type": "OBJECT",
"properties": {"name": {"type": "STRING"}},
"required": ["name"],
},
},
# Add more function definitions here
]
}
]
# ── Stub tool executor (replace with real robot SDK calls) ───────────────────
def execute_tool(name: str, args: dict) -> dict:
print(f" [Tool] {name}({args})")
return {"status": "success"}
# ── Input helpers ────────────────────────────────────────────────────────────
def send_text(session, text: str):
"""Send a text turn."""
return session.send_client_content(
turns=types.Content(role="user", parts=[types.Part(text=text)]),
turn_complete=True,
)
def send_image(session, image_bytes: bytes, prompt: str = ""):
"""Send a JPEG image with an optional text prompt."""
parts = [
types.Part(
inline_data=types.Blob(data=image_bytes, mime_type="image/jpeg")
)
]
if prompt:
parts.append(types.Part(text=prompt))
return session.send_client_content(
turns=types.Content(role="user", parts=parts),
turn_complete=True,
)
def send_audio(session, audio_chunk: bytes):
"""Stream a chunk of raw PCM audio (16-bit, 16 kHz, mono)."""
return session.send_realtime_input(
media=types.Blob(data=audio_chunk, mime_type="audio/pcm;rate=16000")
)
# ── Receive loop ─────────────────────────────────────────────────────────────
async def receive_loop(session):
"""Print model text and handle tool calls until the session ends."""
async for message in session.receive():
if message.server_content:
sc = message.server_content
if sc.model_turn and sc.model_turn.parts:
for part in sc.model_turn.parts:
if part.text:
print(f"Model: {part.text}", end="", flush=True)
if sc.turn_complete:
print("\n[Turn Complete]")
elif message.tool_call:
responses = []
for call in message.tool_call.function_calls:
print(f"\n[Tool Call] {call.name}({call.args})")
result = execute_tool(call.name, call.args)
responses