Robotikë me transmetim

Pika fundore e modelit gemini-robotics-er-2-streaming-preview ekspozon një pikë fundore të dedikuar transmetimi që integrohet me Live API , duke mundësuar ndërveprim në kohë reale, dypalësh midis aplikacionit tuaj dhe robotit. Kjo e bën atë të përshtatshëm për agjentët që kanë nevojë për sythe të shpejta reagimi dhe përgjigje reaktive ndaj mjedisit.

Rastet e përdorimit

• Koordinimi i shumë robotëve : Robotë të shumtë që komunikojnë gjendjen e detyrës dhe delegojnë nëndetyra përmes një sesioni të përbashkët.
• Monitorim i vazhdueshëm : Robotë që vëzhgojnë një skenë dhe aktivizojnë veprime kur ndodhin ngjarje specifike, siç është një enë që arrin një nivel mbushjeje.
• Magazina dhe logjistika : Agjentë të marrjes dhe paketimit që verifikojnë artikujt vizualisht, ndjekin progresin e paketimit dhe rikuperohen nga gabimet.

Specifikimet teknike

Tabela e mëposhtme përshkruan specifikimet teknike për Live API:

Ndërtoni një strukturë agjentësh

Çdo agjent robotik i ndërtuar në Live API ndjek tre hapa:

1. Deklaroni aftësitë e robotit si mjete. Çdo veprim që roboti mund të kryejë — të lundrojë, të kapë, të flasë — bëhet një deklaratë funksioni me një emër, përshkrim dhe skemë parametrash. Veprimet fizike duhet të përdorin "behavior": "BLOCKING" në mënyrë që modeli të presë që roboti të përfundojë përpara se të zgjedhë hapin tjetër.
2. Transmetoni të dhëna multimodale në një seancë të përhershme. Hapni një seancë live.connect dhe mbajeni të hapur për gjithë kohëzgjatjen e detyrës. Dërgoni korniza video, audio ose tekst ndërsa ato mbërrijnë nga sensorët e robotit tuaj.
3. Trajtoni thirrjet e mjeteve në një cikël pranimi. Sa herë që modeli zgjedh një veprim, ai dërgon një mesazh tool_call . Cikli juaj i pranimit ekzekuton funksionin kundër SDK-së së robotit tuaj dhe dërgon mbrapsht një tool_response . Sesioni mbetet i hapur dhe modeli zgjedh veprimin tjetër bazuar në rezultat.

Seksionet e mëposhtme tregojnë se si t'i zbatoni këto hapa në tre modele të zakonshme: një lak agjenti bazë, monitorim proaktiv i skenës me një rrahje zemre dhe drejtim i të folurit përmes TTS si mjet.

Orkestro një robot përmes thirrjes së funksioneve

Shembulli i mëposhtëm tregon të tre hapat e lidhur së bashku në një skript të vetëm Python.

Hapi 1 — përkufizimet e mjeteve — deklaron aftësitë e robotit si deklarata funksioni. Funksioni navigate përdor "behavior": "BLOCKING" në mënyrë që modeli të presë që roboti të arrijë në pikën e referencës përpara se të thërrasë një mjet tjetër. Shtoni më shumë deklarata funksioni në të njëjtën listë për të ekspozuar aftësi shtesë të robotit.

Hapi 2 — ndihmësit e të dhënave hyrëse — tregon tre funksione që transmetojnë të dhëna hyrëse të modaliteteve të ndryshme në sesion: send_text për komandat, send_image për kornizat e kamerave me një kërkesë teksti opsionale dhe send_audio për audio PCM të papërpunuar nga një mikrofon.

Hapi 3 — cikli i marrjes — ekzekutohet njëkohësisht dhe trajton dy lloje mesazhesh: mesazhet server_content (dalja e tekstit të modelit) dhe mesazhet tool_call (modeli që kërkon një veprim të robotit). Kur mbërrin një thirrje mjeti, cikli thërret execute_tool — një cung që e zëvendësoni me SDK-në tuaj të vërtetë të robotit — pastaj dërgon mbrapsht një tool_response në mënyrë që modeli të mund të zgjedhë veprimin tjetër.

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.