API en vivo para robótica

El extremo del modelo gemini-robotics-er-2-streaming-preview expone un extremo de transmisión dedicado que se integra con la API Live, lo que permite la interacción bidireccional en tiempo real entre tu aplicación y el robot. Esto lo hace adecuado para agentes que necesitan bucles de retroalimentación rápidos y respuestas reactivas al entorno.

Probar en Google AI Studio Clonar apps de ejemplo desde GitHub

Casos de uso

  • Coordinación de varios robots: Varios robots que comunican el estado de la tarea y delegan subtareas a través de una sesión compartida.
  • Supervisión continua: Robots que observan una escena y activan acciones cuando ocurren eventos específicos, como un contenedor que alcanza un nivel de llenado.
  • Almacén y logística: Agentes de selección y empaquetado que verifican visualmente los artículos , hacen un seguimiento del progreso del empaquetado y se recuperan de los errores.

Especificaciones técnicas

En la siguiente tabla, se describen las especificaciones técnicas de la API de Live:

Categoría Detalles
Modalidades de entrada Audio (audio PCM sin procesar de 16 bits, 16 kHz, little-endian), imágenes (JPEG <= 1 FPS) y texto
Modalidades de salida Texto
Protocolo Conexión WebSocket con estado (WSS)

Crea una configuración de agente

Cada agente de robótica creado en la API de Live sigue tres pasos:

  1. Declara las capacidades del robot como herramientas. Cada acción que puede realizar el robot (navegar, agarrar, hablar) se convierte en una declaración de función con un nombre, una descripción y un esquema de parámetros. Las acciones físicas deben usar "behavior": "BLOCKING" para que el modelo espere a que el robot termine antes de elegir el siguiente paso.
  2. Transmite la entrada multimodal a una sesión persistente. Abre una sesión de live.connect y mantenla abierta durante la vida útil de la tarea. Envía fotogramas de video, audio o texto a medida que llegan de los sensores del robot.
  3. Controla las llamadas a herramientas en un bucle de recepción. Cada vez que el modelo selecciona una acción, envía un mensaje tool_call. El bucle de recepción ejecuta la función en el SDK del robot y envía una tool_response. La sesión permanece abierta y el modelo elige la siguiente acción según el resultado.

En las siguientes secciones, se muestra cómo aplicar estos pasos a tres patrones comunes: un bucle de agente de línea de base, la supervisión proactiva de escenas con un latido y el enrutamiento de voz a través de TTS como herramienta.

Organiza un robot a través de llamadas a funciones

En el siguiente ejemplo, se muestran los tres pasos conectados en una sola secuencia de comandos de Python.

El paso 1 (definiciones de herramientas) declara las capacidades del robot como declaraciones de funciones. La navigate función usa "behavior": "BLOCKING" para que el modelo espere a que el robot llegue al punto de referencia antes de llamar a otra herramienta. Agrega más declaraciones de funciones en la misma lista para exponer capacidades adicionales del robot.

El paso 2 (helpers de entrada) muestra tres funciones que transmiten diferentes entradas de modalidad a la sesión: send_text para comandos, send_image para fotogramas de cámara con una instrucción de texto opcional y send_audio para audio PCM sin procesar de un micrófono.

El paso 3 (el bucle de recepción) se ejecuta de forma simultánea y controla dos tipos de mensajes: mensajes server_content (la salida de texto del modelo) y mensajes tool_call (el modelo que solicita una acción del robot). Cuando llega una llamada a herramienta, el bucle llama a execute_tool (un código auxiliar que reemplazas por el SDK real del robot) y, luego, envía una tool_response para que el modelo pueda seleccionar la siguiente acción.

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

El bucle de recepción permanece activo después de cada respuesta de la herramienta. El modelo construye y revisa un plan de horizonte largo sin que codifiques toda la secuencia de acciones con anticipación.

Razonamiento espacio-temporal proactivo

La API de Live transmite video, pero los fotogramas de video por sí solos no activan un nuevo turno de razonamiento. Los fotogramas de video deben ir acompañados de una instrucción de texto o audio para activar una respuesta del modelo. Consulta las capacidades de la API de Live para obtener información general.

Para habilitar el razonamiento proactivo, implementa un latido: envía periódicamente el fotograma de la cámara más reciente seguido de una instrucción de texto breve que obliga al modelo a inspeccionar la escena y tomar una decisión explícita. La entrada de video tiene un límite de velocidad de un fotograma por segundo.

Agrega esta corrutina junto con el bucle de recepción de la sección anterior. Se ejecuta como una tarea asyncio independiente en la misma sesión:

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)

No es necesario pausar el latido durante las acciones del robot. Cuando se usa como un detector de éxito implícito, mantenerlo en ejecución permite que el modelo observe continuamente la acción en curso (hacer un seguimiento de si un agarre es seguro, si un vertido está en el objetivo o si un objeto se está asentando correctamente) y reaccione en el momento en que el resultado se vuelva claro.

Salida de audio a través de TTS externo

Gemini Robotics ER 2 muestra texto. Tu aplicación enruta las respuestas completadas a un proveedor de TTS independiente (como Gemini TTS) a través de una devolución de llamada insertada. Esto mantiene bajo tu control la latencia de voz, la selección de voz y el comportamiento de interrupción, y te permite intercambiar backends de TTS sin cambiar la lógica del agente.

También puedes declarar TTS como una herramienta para que el modelo trate "decir algo" de la misma manera que "mover el brazo". Agrega la siguiente declaración de función a tu lista tools de la primera sección:

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

Cuando se ajusta TTS en una declaración de función, el modelo controla la voz a través de la misma ruta de llamada a herramienta que cualquier otra acción del robot. Tu aplicación cumple con la llamada con una devolución de llamada insertada.

Ejemplos en GitHub

Para obtener ejemplos de trabajo completos, incluida la demostración de búsqueda de bocadillos del robot Spot y el saludo al mundo de Tinybot pan-tilt, consulta Ejemplos de la API de Live de Robotics.

¿Qué sigue?