gemini-robotics-er-2-streaming-preview 模型端点公开了一个与 Live API 集成的专用流式传输端点，可实现应用与机器人之间的实时双向互动。因此，它非常适合需要快速反馈环和对环境做出反应式响应的代理。
使用场景
- 多机器人协调：多个机器人通过共享会话通信任务状态并委托子任务。
- 持续监控：观察场景并在发生特定事件（例如容器达到填充水平）时触发操作的机器人。
- 仓储和物流：可直观地验证商品、跟踪打包进度并从错误中恢复的拣货和打包代理。
技术规范
下表列出了 Live API 的技术规范：
| 类别 | 详细信息 |
|---|---|
| 输入模态 | 音频（原始 16 位 PCM 音频，16kHz，小端序）、图片（JPEG <= 1FPS）、文本 |
| 输出模态 | 文本 |
| 协议 | 有状态 WebSocket 连接 (WSS) |
构建代理式设置
基于 Live API 构建的每个机器人代理都遵循以下三个步骤：
- 将机器人功能声明为工具。机器人可以执行的每项操作（导航、抓取、说话）都成为一个函数声明，其中包含名称、说明和参数架构。实体操作必须使用
"behavior": "BLOCKING"，这样模型才能在机器人完成操作后选择下一步。 - 将多模态输入流式传输到持久会话中。打开
live.connect会话，并在任务的整个生命周期内保持该会话处于打开状态。发送从机器人传感器接收到的视频帧、音频或文本。 - 在接收循环中处理工具调用。每次模型选择操作时，都会发送
tool_call消息。您的接收循环针对机器人 SDK 执行该函数，并发送回tool_response。会话保持打开状态，模型会根据结果选择下一个操作。
以下部分展示了如何将这些步骤应用于三种常见模式：基准智能体循环、通过检测信号进行主动场景监控，以及通过 TTS 将语音作为工具进行路由。
通过函数调用编排机器人
以下示例展示了如何在一个 Python 脚本中将这三个步骤连接在一起。
第 1 步 - 工具定义 - 将机器人功能声明为函数声明。navigate 函数使用 "behavior": "BLOCKING"，因此模型会等待机器人到达途经点，然后再调用另一个工具。在同一列表中添加更多函数声明，以公开其他机器人功能。
第 2 步 - 输入辅助函数 - 显示了三个将不同模态输入流式传输到会话中的函数：send_text 用于命令，send_image 用于包含可选文本提示的相机帧，以及 send_audio 用于来自麦克风的原始 PCM 音频。
第 3 步（接收循环）同时运行，并处理两种类型的消息：server_content 消息（模型的文本输出）和 tool_call 消息（模型请求机器人执行操作）。当工具调用到达时，该循环会调用 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 =