Robotica con streaming

L'endpoint del modello gemini-robotics-er-2-streaming-preview espone un endpoint di streaming dedicato che si integra con l'API Live, consentendo l'interazione bidirezionale in tempo reale tra l'applicazione e il robot. Ciò lo rende adatto agli agenti che necessitano di cicli di feedback rapidi e risposte reattive all'ambiente.

Prova in Google AI Studio Clona app di esempio da GitHub

Casi d'uso

  • Coordinamento di più robot: più robot che comunicano lo stato delle attività e delegano le sottoattività tramite una sessione condivisa.
  • Monitoraggio continuo: robot che osservano una scena e attivano azioni quando si verificano eventi specifici, ad esempio quando un container raggiunge un livello di riempimento.
  • Magazzino e logistica: agenti di prelievo e imballaggio che verificano visivamente gli articoli , monitorano l'avanzamento dell'imballaggio e recuperano dagli errori.

Specifiche tecniche

La tabella seguente illustra le specifiche tecniche dell'API Live:

Categoria Dettagli
Modalità di input Audio (audio PCM a 16 bit non elaborato, 16 kHz, little-endian), immagini (JPEG <= 1 FPS), testo
Modalità di output Testo
Protocollo Connessione WebSocket con stato (WSS)

Creare una configurazione di agenti

Ogni agente di robotica creato sull'API Live segue tre passaggi:

  1. Dichiara le funzionalità del robot come strumenti. Ogni azione che il robot può eseguire (navigare, afferrare, parlare) diventa una dichiarazione di funzione con un nome, una descrizione e uno schema di parametri. Le azioni fisiche devono utilizzare "behavior": "BLOCKING" in modo che il modello attenda che il robot finisca prima di scegliere il passaggio successivo.
  2. Trasmetti l'input multimodale in una sessione persistente. Apri una sessione live.connect e mantienila aperta per tutta la durata dell'attività. Invia frame video, audio o testo non appena arrivano dai sensori del robot.
  3. Gestisci le chiamate di strumenti in un ciclo di ricezione. Ogni volta che il modello seleziona un'azione, invia un messaggio tool_call. Il ciclo di ricezione esegue la funzione rispetto all'SDK del robot e invia una tool_response. La sessione rimane aperta e il modello sceglie l'azione successiva in base al risultato.

Le sezioni seguenti mostrano come applicare questi passaggi a tre pattern comuni: un loop dell'agente di base, il monitoraggio proattivo della scena con un heartbeat e il routing della voce tramite TTS come strumento.

Orchestrare un robot tramite la chiamata di funzione

L'esempio seguente mostra tutti e tre i passaggi collegati in un unico script Python.

Il passaggio 1 (definizioni degli strumenti) dichiara le funzionalità del robot come dichiarazioni di funzioni. La funzione navigate utilizza "behavior": "BLOCKING" in modo che il modello attenda che il robot raggiunga il waypoint prima di chiamare un altro strumento. Aggiungi altre dichiarazioni di funzioni nello stesso elenco per esporre funzionalità aggiuntive del robot.

Il passaggio 2 (helper di input) mostra tre funzioni che trasmettono input di modalità diverse nella sessione: send_text per i comandi, send_image per i frame della videocamera con un prompt testuale facoltativo e send_audio per l'audio PCM non elaborato da un microfono.

Il passaggio 3 (il ciclo di ricezione) viene eseguito contemporaneamente e gestisce due tipi di messaggi: i messaggi server_content (l'output di testo del modello) e i messaggi tool_call (il modello che richiede un'azione del robot). Quando arriva una chiamata di strumenti, il ciclo chiama execute_tool, uno stub che sostituisci con l'SDK del robot reale, quindi invia una tool_response in modo che il modello possa selezionare l'azione successiva.

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

Il ciclo di ricezione rimane attivo dopo ogni risposta dello strumento. Il modello costruisce e rivede un piano a lungo termine senza che tu debba codificare in anticipo l'intera sequenza di azioni.

Ragionamento spaziale-temporale proattivo

L'API Live trasmette video, ma i frame video da soli non attivano un nuovo turno di ragionamento. I frame video devono essere accompagnati da un prompt di testo o audio per attivare una risposta del modello. Per informazioni di base, consulta Funzionalità dell'API Live.

Per attivare il ragionamento proattivo, implementa un heartbeat: invia periodicamente l' ultimo frame della videocamera seguito da un breve prompt testuale che costringe il modello a ispezionare la scena e a prendere una decisione esplicita. L'input video è limitato a un frame al secondo.

Aggiungi questa coroutine insieme al ciclo di ricezione della sezione precedente. Viene eseguita come attività asyncio separata nella stessa sessione:

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)

Non è necessario mettere in pausa l'heartbeat durante le azioni del robot. Se utilizzato come rilevatore di successo implicito, il modello può osservare continuamente l'azione in corso, monitorando se una presa è sicura, se un versamento è a segno o se un oggetto si sta sistemando correttamente, e reagire nel momento in cui il risultato diventa chiaro.

Uscita audio tramite TTS esterno

Gemini Robotics ER 2 restituisce testo. La tua applicazione instrada le risposte completate a un provider TTS separato (ad esempio Gemini TTS) tramite un callback inserito. In questo modo, puoi controllare la latenza della voce, la selezione della voce e il comportamento di interruzione e puoi sostituire i backend TTS senza modificare la logica dell'agente.

Puoi anche dichiarare TTS come strumento in modo che il modello tratti "dire qualcosa" come "spostare il braccio". Aggiungi la seguente dichiarazione di funzione all'elenco tools della prima sezione:

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

Se racchiudi TTS in una dichiarazione di funzione, il modello gestisce la voce tramite lo stesso percorso di chiamata di strumenti di qualsiasi altra azione del robot. La tua applicazione soddisfa la chiamata con un callback inserito.

Esempi su GitHub

Per esempi di lavoro completi, inclusa la demo di recupero degli snack del robot Spot e il saluto di Tinybot pan-tilt, consulta Esempi dell'API Live di robotica.

Passaggi successivi