Конечная точка модели gemini-robotics-er-2-streaming-preview предоставляет выделенную потоковую точку доступа, которая интегрируется с Live API , обеспечивая двустороннее взаимодействие в реальном времени между вашим приложением и роботом. Это делает ее подходящей для агентов, которым необходимы быстрые циклы обратной связи и реактивные реакции на окружающую среду.
Варианты использования
- Координация нескольких роботов : несколько роботов обмениваются информацией о состоянии задачи и делегируют подзадачи в рамках общей сессии.
- Непрерывный мониторинг : Роботы, которые наблюдают за происходящим и запускают действия при возникновении определенных событий, например, когда контейнер достигает определенного уровня заполнения.
- Складское хозяйство и логистика : Специалисты по комплектации и упаковке, которые визуально проверяют товары, отслеживают ход упаковки и устраняют ошибки.
Технические характеристики
В таблице ниже приведены технические характеристики Live API:
| Категория | Подробности |
|---|---|
| Входные модальности | Аудио (необработанный 16-битный PCM-аудио, 16 кГц, little-endian), изображения (JPEG <= 1 кадр/с), текст |
| Выходные модальности | Текст |
| Протокол | Соединение WebSocket с сохранением состояния (WSS) |
Создайте агентскую конфигурацию.
Каждый роботизированный агент, созданный на основе Live API, проходит три этапа:
- Объявите возможности робота как инструменты. Каждое действие, которое может выполнять робот — навигация, захват, речь — становится объявлением функции с именем, описанием и схемой параметров. Физические действия должны использовать
"behavior": "BLOCKING", чтобы модель ожидала завершения действия роботом, прежде чем выбрать следующий шаг. - Передавайте многомодальные входные данные в постоянную сессию. Откройте сессию
live.connectи держите ее открытой на протяжении всего выполнения задачи. Отправляйте видеокадры, аудио или текст по мере их поступления от датчиков вашего робота. - Обработка вызовов инструментов осуществляется в цикле приема. Каждый раз, когда модель выбирает действие, она отправляет сообщение
tool_call. Ваш цикл приема выполняет функцию в SDK вашего робота и отправляет обратно сообщениеtool_response. Сессия остается открытой, и модель выбирает следующее действие на основе результата.
В следующих разделах показано, как применить эти шаги к трем распространенным сценариям: базовый цикл работы агента, проактивный мониторинг сцены с использованием сигнала активности и маршрутизация речи через синтез речи в качестве инструмента.
Управляйте роботом посредством вызова функций.
В следующем примере показаны все три шага, объединенные в одном скрипте на 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())
После каждого ответа инструмента цикл приема остается активным. Модель строит и корректирует долгосрочный план без предварительного кодирования всей последовательности действий.
Проактивное пространственно-временное мышление
API Live передает видеопоток, но сами по себе видеокадры не запускают новый этап логического вывода. Для запуска ответа модели видеокадры должны сопровождаться текстовой или звуковой подсказкой. См. раздел «Возможности API Live» для получения более подробной информации.
Для обеспечения проактивного мышления реализуйте механизм "пульса" : периодически отправляйте последний кадр с камеры, за которым следует короткая текстовая подсказка, заставляющая модель осмотреть сцену и принять явное решение. Видеовход ограничен одним кадром в секунду.
Добавьте эту сопрограмму рядом с циклом приема из предыдущего раздела. Она будет выполняться как отдельная задача 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"],
},
},
]
Благодаря обертыванию функции преобразования текста в речь, модель обрабатывает речь по тому же пути вызова инструмента, что и любое другое действие робота. Ваше приложение выполняет вызов с помощью внедренной функции обратного вызова.
Примеры на GitHub
Полные рабочие примеры, включая демонстрацию робота Spot, доставляющего закуски, и пример "hello world" для робота Tinybot с функцией панорамирования и наклона, можно найти в примерах API Robotics Live .
Что дальше?
- Анализ видеоматериалов — выявление ключевых моментов и классификация хода событий.
- Оркестрация задач — выполнение задач с длительным горизонтом планирования без потоковой передачи данных.
- Обзор работающего API — полная документация по работающему API.