Robotyka z przesyłaniem strumieniowym

Punkt końcowy modelu gemini-robotics-er-2-streaming-preview udostępnia dedykowany punkt końcowy streamingu, który integruje się z interfejsem Live API, umożliwiając dwukierunkową interakcję w czasie rzeczywistym między aplikacją a robotem. Dzięki temu nadaje się do agentów, którzy potrzebują szybkich pętli informacji zwrotnych i reaktywnych odpowiedzi na środowisko.

Wypróbuj w Google AI Studio Sklonuj przykładowe aplikacje z GitHub

Przypadki użycia

  • Koordynacja wielu robotów: wiele robotów, które komunikują stan zadania i delegują podzadania w ramach wspólnej sesji.
  • Ciągłe monitorowanie: Roboty, które obserwują scenę i wywołują działania gdy wystąpią określone zdarzenia, np. gdy pojemnik osiągnie określony poziom napełnienia.
  • Magazyn i logistyka: agenci kompletujący i pakujący, którzy wizualnie sprawdzają produkty, śledzą postępy w pakowaniu i reagują na błędy.

Specyfikacja techniczna

W tabeli poniżej znajdziesz specyfikację techniczną interfejsu Live API:

Kategoria Szczegóły
Metody wprowadzania Dźwięk (surowy 16-bitowy dźwięk PCM, 16 kHz, little-endian), obrazy (JPEG <= 1 FPS), tekst
Metody wyjściowe Tekst
Protokół Połączenie WebSocket ze stanem (WSS)

Tworzenie konfiguracji agenta

Każdy agent robotyki utworzony za pomocą interfejsu Live API wykonuje 3 kroki:

  1. Deklarowanie możliwości robota jako narzędzi. Każda czynność, którą może wykonać robot – nawigacja, chwytanie, mówienie – staje się deklaracją funkcji z nazwą, opisem i schematem parametrów. W przypadku działań fizycznych należy użyć "behavior": "BLOCKING" , aby model czekał na zakończenie działania robota przed wybraniem następnego kroku.
  2. Przesyłanie strumieniowe danych wejściowych z różnych metod do trwałej sesji. Otwórz sesję live.connect i pozostaw ją otwartą przez cały czas trwania zadania. Wysyłaj klatki wideo, dźwięk lub tekst, gdy tylko dotrą z czujników robota.
  3. Obsługa wywołań narzędzi w pętli odbierania. Za każdym razem, gdy model wybierze działanie, wysyła komunikat tool_call. Pętla odbierania wykonuje funkcję w pakiecie SDK robota i odsyła tool_response. Sesja pozostaje otwarta, a model wybiera następne działanie na podstawie wyniku.

W sekcjach poniżej dowiesz się, jak zastosować te kroki w 3 typowych wzorcach: podstawowej pętli agenta, proaktywnym monitorowaniu sceny za pomocą sygnału kontrolnego oraz przekierowywaniu mowy przez TTS jako narzędzie.

Orkiestrowanie robota za pomocą wywoływania funkcji

W przykładzie poniżej pokazujemy wszystkie 3 kroki połączone w jednym skrypcie w Pythonie.

Krok 1 – definicje narzędzi – deklaruje możliwości robota jako deklaracje funkcji. Funkcja navigate używa "behavior": "BLOCKING" , aby model czekał na dotarcie robota do punktu orientacyjnego przed wywołaniem innego narzędzia. Aby udostępnić dodatkowe możliwości robota, dodaj więcej deklaracji funkcji do tej samej listy.

Krok 2 – pomocnicy danych wejściowych – pokazuje 3 funkcje, które przesyłają strumieniowo dane wejściowe z różnych metod do sesji: send_text na potrzeby poleceń, send_image na potrzeby klatek z kamery z opcjonalnym promptem tekstowym oraz send_audio na potrzeby surowego dźwięku PCM z mikrofonu.

Krok 3 – pętla odbierania – działa równolegle i obsługuje 2 rodzaje wiadomości: wiadomości server_content (tekstowe dane wyjściowe modelu) i wiadomości tool_call (model żądający działania robota). Gdy nadejdzie wywołanie narzędzia, pętla wywoła execute_tool – element zastępczy, który zastąpisz prawdziwym pakietem SDK robota – a następnie odeśle tool_response , aby model mógł wybrać następne działanie.

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())

Pętla odbierania pozostaje aktywna po każdej odpowiedzi narzędzia. Model tworzy i modyfikuje plan długoterminowy bez konieczności kodowania całej sekwencji działań z wyprzedzeniem.

Proaktywne rozumowanie przestrzenno-czasowe

Interfejs Live API przesyła strumieniowo wideo, ale same klatki wideo nie wywołują nowej tury rozumowania. Aby wywołać odpowiedź modelu, klatkom wideo musi towarzyszyć prompt tekstowy lub dźwiękowy. Więcej informacji znajdziesz w sekcji Możliwości interfejsu Live API.

Aby włączyć proaktywne rozumowanie, zaimplementuj pakiet podtrzymujący: okresowo wysyłaj najnowszą klatkę z aparatu, a następnie krótki prompt tekstowy, który zmusi model do sprawdzenia sceny i podjęcia wyraźnej decyzji. Dane wejściowe wideo są ograniczone do 1 klatki na sekundę.

Dodaj ten współprogram obok pętli odbierania z poprzedniej sekcji. Działa ona jako osobne zadanie asyncio w tej samej sesji:

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)

Podczas działań robota nie musisz wstrzymywać sygnału kontrolnego. Jeśli używasz go jako niejawnego detektora sukcesu, jego działanie pozwala modelowi na ciągłe obserwowanie działania w toku (śledzenie, czy chwyt jest bezpieczny, czy nalewanie jest prawidłowe lub czy obiekt prawidłowo się układa) i reagowanie w momencie, gdy wynik stanie się jasny.

Wiadomości sygnału kontrolnego działają jak tury użytkownika i przerywają generowanie modelu w toku. Aby dowiedzieć się, jak interfejs Live API obsługuje to zachowanie, zapoznaj się z przewodnikiem po interfejsie Live API dotyczącym przerw.

Wyjście audio przez zewnętrzny TTS

Gemini Robotics ER 2 zwraca tekst. Twoja aplikacja przekierowuje ukończone odpowiedzi do osobnego dostawcy TTS (np. Gemini TTS) za pomocą wstrzykniętego wywołania zwrotnego. Dzięki temu masz kontrolę nad opóźnieniem mowy, wyborem głosu i zachowaniem podczas przerw oraz możesz wymieniać backendy TTS bez zmiany logiki agenta.

Możesz też zadeklarować TTS jako narzędzie, aby model traktował „powiedz coś” tak samo jak „przesuń ramię”. Dodaj tę deklarację funkcji do listy tools z pierwszej sekcji:

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"],
        },
    },
]

Dzięki opakowaniu TTS w deklarację funkcji model obsługuje mowę za pomocą tej samej ścieżki wywołania narzędzia co inne działania robota. Twoja aplikacja realizuje wywołanie za pomocą wstrzykniętego wywołania zwrotnego.

Przykłady na GitHubie

Pełne przykłady działania, w tym wersja demonstracyjna robota Spot z przekąskami i Tinybot pan-tilt hello world, znajdziesz w sekcji Przykłady interfejsu Robotics Live API.

Co dalej?