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 contenitore raggiunge un determinato livello di riempimento.
  • Magazzino e logistica: agenti di prelievo e imballaggio che verificano visivamente gli articoli, monitorano l'avanzamento dell'imballaggio e correggono gli errori.

Specifiche tecniche

La seguente tabella descrive 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)

Crea una configurazione agentica

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

  1. Dichiarare 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 dei 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 fotogrammi video, audio o testo man mano che arrivano dai sensori del robot.
  3. Gestire le chiamate allo strumento 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 sull'SDK del robot e restituisce un 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 ciclo 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, le definizioni degli strumenti, dichiara le funzionalità del robot come dichiarazioni di funzioni. La funzione navigate utilizza "behavior": "BLOCKING", quindi il modello attende 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, ovvero gli helper di input, mostra tre funzioni che trasmettono in streaming diversi input di modalità 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: messaggi server_content (l'output di testo del modello) e messaggi tool_call (il modello che richiede un'azione del robot). Quando arriva una chiamata allo strumento, il ciclo chiama execute_tool, uno stub che sostituisci con l'SDK del robot reale, quindi invia un 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 crea e rivede un piano a lungo termine senza che tu debba codificare in anticipo l'intera sequenza di azioni.

Ragionamento spaziotemporale proattivo

L'API Live esegue lo streaming del video, ma i fotogrammi video da soli non attivano un nuovo turno di ragionamento. I fotogrammi video devono essere accompagnati da un prompt di testo o audio per attivare una risposta del modello. Per ulteriori dettagli, consulta la sezione Funzionalità dell'API Live.

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

Implementare l'heartbeat

La coroutine heartbeat viene eseguita come attività asyncio separata nella stessa sessione. Ha come target una cadenza di 1 Hz (che corrisponde al limite di frequenza di input video) mentre attende il completamento di ogni turno (er_turn_done) per evitare di interrompere il ragionamento in corso:

async def heartbeat(session, camera, er_turn_done: asyncio.Event):
    TARGET_INTERVAL_SEC = 1.0

    while True:
        start_time = asyncio.get_running_loop().time()

        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."
            )
        )

        # Wait for the model to finish responding before sending the next heartbeat
        await er_turn_done.wait()
        er_turn_done.clear()

        # Sleep only the remaining time to maintain ~1 Hz cadence
        elapsed = asyncio.get_running_loop().time() - start_time
        remaining = TARGET_INTERVAL_SEC - elapsed
        if remaining > 0:
            await asyncio.sleep(remaining)

Aggiorna il ciclo di ricezione

Per segnalare quando il modello ha completato il suo turno, aggiorna receive_loop per impostare er_turn_done:

# In receive_loop: signal when the model finishes its turn
if sc.turn_complete:
    er_turn_done.set()

Uscita audio tramite TTS esterno

Gemini Robotics ER 2 restituisce del testo. La tua applicazione indirizza le risposte completate a un fornitore TTS separato (ad esempio Gemini TTS) tramite un callback inserito. In questo modo, la latenza del parlato, la selezione della voce e il comportamento di interruzione rimangono sotto il tuo controllo e puoi scambiare i backend TTS senza modificare la logica dell'agente.

Puoi anche dichiarare la sintesi vocale come strumento in modo che il modello tratti "di' qualcosa" allo stesso modo di "muovi 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 sintesi vocale tramite lo stesso percorso di chiamata dello strumento di qualsiasi altra azione del robot. La tua applicazione soddisfa la chiamata con un callback inserito.

Esempi su GitHub

Per esempi di funzionamento completi, tra cui la demo di recupero dello snack del robot Spot e il movimento panoramico e verticale di Tinybot Hello World, vedi Esempi di API Robotics Live.

Passaggi successivi