API Live pour la robotique

Le point de terminaison de modèle gemini-robotics-er-2-streaming-preview expose un point de terminaison de streaming dédié qui s'intègre à l'API Live, ce qui permet une interaction bidirectionnelle en temps réel entre votre application et le robot. Il convient donc aux agents qui ont besoin de boucles de rétroaction rapides et de réponses réactives à l'environnement.

Essayer dans Google AI Studio Cloner des exemples d'applications depuis GitHub

Cas d'utilisation

  • Coordination de plusieurs robots : plusieurs robots qui communiquent l'état des tâches et délèguent des sous-tâches via une session partagée.
  • Surveillance continue : robots qui observent une scène et déclenchent des actions lorsque des événements spécifiques se produisent, par exemple lorsqu'un conteneur atteint un niveau de remplissage.
  • Entrepôt et logistique : agents de préparation et d'emballage qui vérifient visuellement les articles , suivent l'avancement de l'emballage et corrigent les erreurs.

Spécifications techniques

Le tableau suivant présente les spécifications techniques de l'API Live :

Catégorie Détails
Modes d'entrée Audio (audio PCM 16 bits brut, 16 kHz, little-endian), images (JPEG <= 1 FPS), texte
Modes de sortie Texte
Protocole Connexion WebSocket avec état (WSS)

Créer une configuration d'agent

Chaque agent de robotique créé sur l'API Live suit trois étapes :

  1. Déclarer les fonctionnalités du robot en tant qu'outils. Chaque action que le robot peut effectuer (naviguer, saisir, parler) devient une déclaration de fonction avec un nom, une description et un schéma de paramètres. Les actions physiques doivent utiliser "behavior": "BLOCKING" afin que le modèle attende que le robot ait terminé avant de choisir l'étape suivante.
  2. Diffuser une entrée multimodale dans une session persistante. Ouvrez une session live.connect et laissez-la ouverte pendant toute la durée de la tâche. Envoyez des images vidéo, de l'audio ou du texte à mesure qu'ils arrivent des capteurs de votre robot.
  3. Gérer les appels d'outils dans une boucle de réception. Chaque fois que le modèle sélectionne une action, il envoie un message tool_call. Votre boucle de réception exécute la fonction par rapport au SDK de votre robot et renvoie une tool_response. La session reste ouverte et le modèle choisit l'action suivante en fonction du résultat.

Les sections suivantes expliquent comment appliquer ces étapes à trois modèles courants : une boucle d'agent de base, une surveillance proactive de la scène avec un signal de présence et un routage de la parole via la synthèse vocale en tant qu'outil.

Orchestrer un robot via des appels de fonction

L'exemple suivant montre les trois étapes liées dans un seul script Python.

L'étape 1 (définitions d'outils) déclare les fonctionnalités du robot en tant que déclarations de fonction. La fonction navigate utilise "behavior": "BLOCKING" afin que le modèle attende que le robot atteigne le point de cheminement avant d'appeler un autre outil. Ajoutez d'autres déclarations de fonction dans la même liste pour exposer des fonctionnalités supplémentaires du robot.

L'étape 2 (assistants d'entrée) présente trois fonctions qui diffusent différentes entrées de modalités dans la session : send_text pour les commandes, send_image pour les images de la caméra avec un prompt textuel facultatif et send_audio pour l'audio PCM brut provenant d'un microphone.

L'étape 3 (boucle de réception) s'exécute simultanément et gère deux types de messages : les messages server_content (sortie textuelle du modèle) et les messages tool_call (le modèle demandant une action du robot). Lorsqu'un appel d'outil arrive, la boucle appelle execute_tool (un stub que vous remplacez par le SDK de votre robot réel), puis renvoie une tool_response afin que le modèle puisse sélectionner l'action suivante.

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

La boucle de réception reste active après chaque réponse de l'outil. Le modèle construit et révise un plan à long terme sans que vous ayez à encoder l'intégralité de la séquence d'actions à l'avance.

Raisonnement spatio-temporel proactif

L'API Live diffuse des vidéos, mais les images vidéo seules ne déclenchent pas de nouveau tour de raisonnement. Les images vidéo doivent être accompagnées d'une invite textuelle ou audio pour déclencher une réponse du modèle. Pour en savoir plus, consultez les fonctionnalités de l'API Live pour background.

Pour activer le raisonnement proactif, implémentez un signal de présence : envoyez régulièrement la dernière image de la caméra, suivie d'un court prompt textuel qui oblige le modèle à inspecter la scène et à prendre une décision explicite. L'entrée vidéo est limitée à une image par seconde.

Ajoutez cette coroutine à côté de la boucle de réception de la section précédente. Elle s'exécute en tant que tâche asyncio distincte dans la même session :

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)

Vous n'avez pas besoin de mettre en pause le signal de présence pendant les actions du robot. Lorsqu'il est utilisé comme détecteur de réussite implicite, le fait de le laisser s'exécuter permet au modèle d'observer en permanence l'action en cours (en vérifiant si une prise est sécurisée, si un versement est ciblé ou si un objet se dépose correctement) et de réagir dès que le résultat devient clair.

Sortie audio via une synthèse vocale externe

Gemini Robotics ER 2 renvoie du texte. Votre application achemine les réponses complètes vers un fournisseur de synthèse vocale distinct (tel que Gemini TTS) via un rappel injecté. Cela vous permet de contrôler la latence de la parole, la sélection de la voix et le comportement d'interruption, et de remplacer les backends de synthèse vocale sans modifier la logique de l'agent.

Vous pouvez également déclarer la synthèse vocale comme outil afin que le modèle traite "dire quelque chose" de la même manière que "déplacer le bras". Ajoutez la déclaration de fonction suivante à votre liste tools de la première section :

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

En encapsulant la synthèse vocale dans une déclaration de fonction, le modèle gère la parole via le même chemin d'appel d'outil que toute autre action du robot. Votre application traite l'appel avec un rappel injecté.

Exemples sur GitHub

Pour obtenir des exemples de travail complets, y compris la démonstration de récupération de collations du robot Spot et le Tinybot Hello World, consultez Exemples d'API Live pour la robotique.

Étape suivante