Robotika dengan streaming

Endpoint model gemini-robotics-er-2-streaming-preview mengekspos endpoint streaming khusus yang terintegrasi dengan Live API, sehingga memungkinkan interaksi dua arah secara real-time antara aplikasi Anda dan robot. Hal ini membuatnya cocok untuk agen yang memerlukan loop umpan balik cepat dan respons reaktif terhadap lingkungan.

Kasus penggunaan

• Koordinasi multi-robot: Beberapa robot yang mengomunikasikan status tugas dan mendelegasikan subtugas melalui sesi bersama.
• Pemantauan berkelanjutan: Robot yang mengamati suatu adegan dan memicu tindakan saat peristiwa tertentu terjadi, seperti saat kontainer mencapai tingkat pengisian tertentu.
• Gudang dan logistik: Agen pengambilan dan pengemasan yang memverifikasi item secara visual, melacak progres pengemasan, dan memulihkan dari error.

Spesifikasi teknis

Tabel berikut menguraikan spesifikasi teknis untuk Live API:

Membangun penyiapan agentic

Setiap agen robotik yang dibangun di Live API mengikuti tiga langkah:

1. Mendeklarasikan kemampuan robot sebagai alat. Setiap tindakan yang dapat dilakukan robot — bernavigasi, menggenggam, berbicara — menjadi deklarasi fungsi dengan nama, deskripsi, dan skema parameter. Tindakan fisik harus menggunakan "behavior": "BLOCKING" sehingga model menunggu robot selesai sebelum memilih langkah berikutnya.
2. Streaming input multimodal ke sesi persisten. Buka sesi live.connect dan biarkan sesi tetap terbuka selama tugas berlangsung. Mengirim frame video, audio, atau teks saat diterima dari sensor robot Anda.
3. Menangani panggilan alat dalam loop penerimaan. Setiap kali model memilih tindakan, model akan mengirim pesan tool_call. Loop penerimaan Anda menjalankan fungsi terhadap robot SDK Anda dan mengirim kembali tool_response. Sesi tetap terbuka, dan model memilih tindakan berikutnya berdasarkan hasilnya.

Bagian berikut menunjukkan cara menerapkan langkah-langkah ini ke tiga pola umum: loop agen dasar, pemantauan adegan proaktif dengan heartbeat, dan perutean ucapan melalui TTS sebagai alat.

Mengatur robot melalui panggilan fungsi

Contoh berikut menunjukkan ketiga langkah yang digabungkan dalam satu skrip Python.

Langkah 1 — definisi alat — mendeklarasikan kemampuan robot sebagai deklarasi fungsi. Fungsi navigate menggunakan "behavior": "BLOCKING" sehingga model menunggu robot mencapai titik jalan sebelum memanggil alat lain. Tambahkan lebih banyak deklarasi fungsi dalam daftar yang sama untuk mengekspos kemampuan robot tambahan.

Langkah 2 — helper input — menampilkan tiga fungsi yang mengalirkan input modalitas yang berbeda ke dalam sesi: send_text untuk perintah, send_image untuk frame kamera dengan perintah teks opsional, dan send_audio untuk audio PCM mentah dari mikrofon.

Langkah 3 — loop penerimaan — berjalan secara bersamaan dan menangani dua jenis pesan: pesan server_content (output teks model) dan pesan tool_call (model yang meminta tindakan robot). Saat panggilan alat tiba, loop memanggil execute_tool — stub yang Anda ganti dengan SDK robot asli — lalu mengirim kembali execute_tool sehingga model dapat memilih tindakan berikutnya.tool_response

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(