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.

Provo në Google AI Studio, klono aplikacione shembull nga GitHub

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:

Kategoria Detajet
Modalitetet e të dhënave Audio (audio PCM 16-bit i papërpunuar, 16kHz, little-endian), imazhe (JPEG <= 1FPS), tekst
Modalitetet e prodhimit Tekst
Protokolli Lidhje me Stateful WebSocket (WSS)

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

Laku i pranimit mbetet aktiv pas çdo përgjigjeje të mjetit. Modeli ndërton dhe rishikon një plan afatgjatë pa e koduar paraprakisht të gjithë sekuencën e veprimeve.

Arsyetim proaktiv hapësinor-kohor

API-ja Live transmeton video, por vetëm kornizat e videos nuk shkaktojnë një kthesë të re arsyetimi. Kornizat e videos duhet të shoqërohen nga një mesazh teksti ose audio për të shkaktuar një përgjigje modeli. Shihni aftësitë e API-t Live për sfond.

Për të aktivizuar arsyetimin proaktiv, zbatoni një “heartbath” : dërgoni periodikisht kuadrin më të fundit të kamerës, të ndjekur nga një mesazh i shkurtër me tekst që e detyron modelin të inspektojë skenën dhe të marrë një vendim të qartë. Shpejtësia e të dhënave video është e kufizuar në një kuadër për sekondë.

Shtojeni këtë korutinë së bashku me ciklin e marrjes nga seksioni i mëparshëm. Ajo ekzekutohet si një detyrë e veçantë asyncio në të njëjtën seancë:

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)

Nuk keni nevojë të ndaloni rrahjet e zemrës gjatë veprimeve të robotit. Kur përdoret si një detektor implicit suksesi , mbajtja e tij në punë i lejon modelit të vëzhgojë vazhdimisht veprimin në zhvillim e sipër - duke ndjekur nëse një kapje është e sigurt, një derdhje është në shënjestër ose një objekt po vendoset siç duhet - dhe të reagojë në momentin që rezultati bëhet i qartë.

Dalja e audios përmes TTS të jashtëm

Gemini Robotics ER 2 kthen tekst. Aplikacioni juaj i drejton përgjigjet e përfunduara te një ofrues i veçantë TTS (siç është Gemini TTS ) nëpërmjet një rikthimi të injektuar. Kjo e mban nën kontrollin tuaj vonesën e të folurit, përzgjedhjen e zërit dhe sjelljen e ndërprerjes, dhe ju lejon të ndërroni backend-et TTS pa ndryshuar logjikën e agjentit.

Gjithashtu mund të deklaroni TTS si një mjet në mënyrë që modeli ta trajtojë "thuaj diçka" njësoj si "lëviz krahun". Shtoni deklaratën e mëposhtme të funksionit në listën tuaj tools nga seksioni i parë:

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

Duke e mbështjellë TTS-në në një deklaratë funksioni, modeli e trajton të folurit përmes të njëjtës rrugë thirrjeje mjeti si çdo veprim tjetër i robotit. Aplikacioni juaj e përmbush thirrjen me një thirrje mbrapsht të injektuar.

Shembuj në GitHub

Për shembuj të plotë funksionalë, duke përfshirë demonstrimin e marrjes së ushqimeve të robotit Spot dhe funksionin Tinybot pan-tilt hello world, shihni shembujt e Robotics Live API .

Çfarë vjen më pas