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 = execute_tool(call.name, call.args)
responses.append(
types.FunctionResponse(
name=call.name,
response=result,
id=call.id,
)
)
await session.send_tool_response(function_responses=responses)
# ── Main ─────────────────────────────────────────────────────────────────────
async def main():
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
config = types.LiveConnectConfig(
response_modalities=["TEXT"],
tools=tools,
system_instruction=types.Content(
parts=[types.Part(text="You are a robot controller. Use tools to execute commands.")]
),
)
async with client.aio.live.connect(model=MODEL, config=config) as session:
recv_task = asyncio.create_task(receive_loop(session))
# Connect robot perception callbacks and user inputs to the helpers above.
recv_task.cancel()
asyncio.run(main())
接收循环在每次工具响应后保持活跃状态。模型会构建和修订长期计划，而无需您提前对整个操作序列进行编码。
主动时空推理
Live API 会流式传输视频，但仅视频帧不会触发新的推理轮次。视频帧必须附带文本或音频提示，才能触发模型响应。如需了解背景信息，请参阅 Live API 功能。
如需启用主动推理，请实现心跳：定期发送 最新的相机帧，然后发送简短的文本提示，强制模型 检查场景并做出明确的决策。视频输入速率限制为每秒一帧。
将此协程添加到上一部分中的接收循环旁边。它在同一会话中作为单独的 asyncio 任务运行：
async def heartbeat(session, camera): # camera is your robot camera API
while True:
frame = await camera.latest_jpeg()
await session.send_realtime_input(
video=types.Blob(data=frame, mime_type="image/jpeg")
)
await session.send_realtime_input(
text=(
"[HEARTBEAT] If no task is active, call 'ack' and wait for user"
" input. If a task is active: observe the scene. If the current"
" step is progressing correctly, call 'ack'. If the current step"
" is complete, call 'run_instruction' with the next step. If the"
" overall goal is achieved, call 'reset' and inform the user."
)
)
await asyncio.sleep(1)
您无需在机器人操作期间暂停心跳。当用作 隐式成功检测器时，保持其运行可让模型持续 观察正在进行的操作（跟踪抓取是否安全、倾倒是否准确或对象是否正确放置），并在结果明确时立即做出反应。
通过外部 TTS 进行音频输出
Gemini Robotics ER 2 会返回文本。您的应用通过注入的回调将完成的响应路由到单独的 TTS 提供商（例如 Gemini TTS）。 这样，您就可以控制语音延迟、语音选择和中断行为，并且无需更改智能体逻辑即可交换 TTS 后端。
您还可以将 TTS 声明为工具，以便模型将“说些什么”与“移动手臂”视为相同。 将以下函数声明添加到第一部分中的 tools 列表：
TOOLS = [
{
"name": "send_message",
"description": (
"Speak a message aloud via TTS, then deliver it to the"
" specified target. Use target='user' to speak directly"
" to the user, or a peer agent name (e.g., 'duo') to"
" communicate with another robot."
),
"parameters": {
"type": "object",
"properties": {
"target": {
"type": "string",
"description": "Recipient: 'user' or a peer agent name.",
},
"message": {
"type": "string",
"description": "The message to speak and deliver.",
},
},
"required": ["target", "message"],
},
},
]
通过将 TTS 封装在函数声明中，模型会通过与任何其他机器人操作相同的工具调用路径来处理语音。您的应用通过注入的回调来完成调用。
GitHub 示例
如需查看完整的示例，包括 Spot 机器人零食获取演示和 Tinybot 平移倾斜 Hello World，请参阅 Robotics Live API 示例。
后续步骤
- 视频理解 - 时刻查找和进度分类。
- 任务编排 - 无需流式传输的长期任务。
- Live API 概览 - 完整的 Live API 文档。