Interfejs Live API do robotyki

Punkt końcowy modelu gemini-robotics-er-2-streaming-preview udostępnia dedykowany punkt końcowy przesyłania strumieniowego, 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ę on do agentów, którzy potrzebują szybkich pętli informacji zwrotnych i reaktywnych odpowiedzi na środowisko.

Wypróbuj w Google AI Studio Skopiuj przykładowe aplikacje z GitHuba

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 typu pick-and-pack, którzy wizualnie weryfikują przedmioty, ś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 z zachowaniem stanu (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 wielu źródeł 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 heartbeat oraz przekierowywaniu mowy przez TTS jako narzędzie.

Orkiestrowanie robota za pomocą wywoływania funkcji

W tym przykładzie pokazano 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 źródeł do sesji: send_text dla poleceń, send_image dla klatek z kamery z opcjonalnym promptem tekstowym oraz send_audio dla 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 zmusza 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)

Nie musisz wstrzymywać sygnału heartbeat podczas działań robota. Jeśli używasz go jako niejawnego detektora sukcesu, jego działanie pozwala modelowi na ciągłe obserwowanie postępów działania – sprawdzanie, 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.

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 przerwania, a także możesz zmieniać 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 działające przykłady, w tym wersja demonstracyjna robota Spot, który przynosi przekąski, oraz Tinybot pan-tilt hello world, znajdziesz w sekcji Przykłady interfejsu Robotics Live API.

Co dalej?