स्ट्रीमिंग के साथ रोबोटिक्स

gemini-robotics-er-2-streaming-preview मॉडल एंडपॉइंट, एक खास स्ट्रीमिंग एंडपॉइंट दिखाता है. यह एंडपॉइंट, Live API के साथ इंटिग्रेट होता है. इससे आपके ऐप्लिकेशन और रोबोट के बीच रीयल-टाइम में, दोनों दिशाओं में इंटरैक्शन हो सकता है. यह उन एजेंट के लिए सही है जिन्हें फ़ीडबैक लूप की ज़रूरत होती है और जो आस-पास के माहौल के हिसाब से तुरंत जवाब देते हैं.

इस्तेमाल के उदाहरण

• एक से ज़्यादा रोबोट के बीच तालमेल: एक से ज़्यादा रोबोट, जो टास्क की स्थिति के बारे में बताते हैं और शेयर किए गए सेशन के ज़रिए सबटास्क सौंपते हैं.
• लगातार निगरानी: रोबोट, किसी सीन को देखते हैं और खास इवेंट होने पर कार्रवाइयां ट्रिगर करते हैं. जैसे, किसी कंटेनर का पूरा भर जाना.
• वेयरहाउस और लॉजिस्टिक्स: पिक-ऐंड-पैक एजेंट, जो आइटम की पुष्टि विज़ुअल तरीके से करते हैं, पैकिंग की प्रोसेस को ट्रैक करते हैं, और गड़बड़ियों को ठीक करते हैं.

तकनीकी जानकारी

यहां दी गई टेबल में, Live API की तकनीकी जानकारी दी गई है:

एजेंटिक सेटअप बनाना

Live API पर बने हर रोबोटिक्स एजेंट के लिए, ये तीन चरण पूरे करने होते हैं:

1. रोबोट की क्षमताओं को टूल के तौर पर एलान करना. रोबोट की हर कार्रवाई — नेविगेट करना, पकड़ना, बोलना — एक फ़ंक्शन एलान बन जाती है. इसमें नाम, ब्यौरा, और पैरामीटर स्कीमा शामिल होता है. फ़िज़िकल ऐक्शन के लिए, "behavior": "BLOCKING" का इस्तेमाल करना ज़रूरी है, ताकि मॉडल, रोबोट के काम पूरा करने के बाद ही अगला चरण चुन सके.
2. मल्टीमॉडल इनपुट को परसिस्टेंट सेशन में स्ट्रीम करना. live.connect सेशन खोलें और इसे टास्क पूरा होने तक खुला रखें. वीडियो फ़्रेम, ऑडियो या टेक्स्ट को अपने रोबोट के सेंसर से मिलने पर भेजें.
3. रिसीव लूप में टूल कॉल को मैनेज करना. मॉडल, हर बार कोई कार्रवाई चुनने पर, tool_call मैसेज भेजता है. आपका रिसीव लूप, आपके रोबोट एसडीके के ख़िलाफ़ फ़ंक्शन को एक्ज़ीक्यूट करता है और tool_response वापस भेजता है. सेशन खुला रहता है और मॉडल, नतीजे के आधार पर अगली कार्रवाई चुनता है.

यहां दिए गए सेक्शन में, इन चरणों को तीन सामान्य पैटर्न पर लागू करने का तरीका बताया गया है: बेसलाइन एजेंट लूप, हार्टबीट के साथ सीन की प्रोऐक्टिव मॉनिटरिंग, और टीटीएस को टूल के तौर पर इस्तेमाल करके स्पीच को रूट करना.

फ़ंक्शन कॉल करके रोबोट को ऑर्केस्ट्रेट करना

यहां दिए गए उदाहरण में, तीनों चरणों को एक ही Python स्क्रिप्ट में एक साथ दिखाया गया है.

पहला चरण — टूल की परिभाषाएं — रोबोट की क्षमताओं को फ़ंक्शन एलान के तौर पर दिखाता है. The navigate फ़ंक्शन, "behavior": "BLOCKING" का इस्तेमाल करता है, ताकि the model, रोबोट के वेपॉइंट तक पहुंचने के बाद ही कोई दूसरा टूल कॉल करे. रोबोट की अन्य क्षमताओं को दिखाने के लिए, उसी सूची में ज़्यादा फ़ंक्शन एलान जोड़ें.

दूसरा चरण — इनपुट हेल्पर — तीन ऐसे फ़ंक्शन दिखाता है जो सेशन में अलग-अलग मोडैलिटी इनपुट स्ट्रीम करते हैं: कमांड के लिए send_text, कैमरे के फ़्रेम के लिए send_image जिसमें टेक्स्ट प्रॉम्प्ट भी शामिल किया जा सकता है, और माइक्रोफ़ोन से मिलने वाले रॉ पीसीएम ऑडियो के लिए send_audio.

तीसरा चरण — रिसीव लूप — एक साथ चलता है और दो तरह के मैसेज को मैनेज करता है: server_content मैसेज (मॉडल का टेक्स्ट आउटपुट) और tool_call मैसेज (मॉडल, रोबोट से किसी कार्रवाई का अनुरोध कर रहा है). जब कोई टूल कॉल आता है, तो लूप, execute_tool को कॉल करता है. यह एक स्टब है जिसे आपको अपने असली रोबोट एसडीके से बदलना होता है. इसके बाद, यह 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(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())

हर टूल के जवाब के बाद, रिसीव लूप चालू रहता है. मॉडल, कार्रवाई के पूरे क्रम को पहले से कोड किए बिना, लंबे समय की योजना बनाता है और उसमें बदलाव करता है.

स्पेशल-टेम्परल रीज़निंग की प्रोऐक्टिव सुविधा

Live API, वीडियो स्ट्रीम करता है. हालांकि, सिर्फ़ वीडियो फ़्रेम से रीज़निंग का नया चरण ट्रिगर नहीं होता. मॉडल से जवाब पाने के लिए, वीडियो फ़्रेम के साथ टेक्स्ट या ऑडियो प्रॉम्प्ट होना ज़रूरी है. ज़्यादा जानकारी के लिए, Live API की क्षमताएं देखें.

प्रोऐक्टिव रीज़निंग की सुविधा चालू करने के लिए, हार्टबीट लागू करें: समय-समय पर, कैमरे का सबसे नया फ़्रेम भेजें. इसके बाद, एक छोटा टेक्स्ट प्रॉम्प्ट भेजें, ताकि मॉडल सीन की जांच करे और साफ़ तौर पर फ़ैसला ले. वीडियो इनपुट की दर, एक फ़्रेम प्रति सेकंड तक सीमित है.

पिछले सेक्शन के रिसीव लूप के साथ, यह कोरोटीन जोड़ें. यह उसी सेशन में, अलग asyncio टास्क के तौर पर चलता है:

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)

रोबोट की कार्रवाइयों के दौरान, आपको हार्टबीट को रोकने की ज़रूरत नहीं है. इंप्लिसिट सक्सेस डिटेक्टर के तौर पर इस्तेमाल करने पर, इसे चालू रखने से मॉडल, प्रोसेस में मौजूद कार्रवाई को लगातार देख सकता है. जैसे, यह ट्रैक करना कि पकड़ मजबूत है या नहीं, कोई चीज़ सही जगह पर डाली जा रही है या नहीं, या कोई ऑब्जेक्ट सही तरीके से सेट हो रहा है या नहीं. साथ ही, नतीजा साफ़ होने पर तुरंत जवाब दिया जा सकता है.

हार्टबीट मैसेज, उपयोगकर्ता के टर्न के तौर पर काम करते हैं और मॉडल जनरेशन की प्रोसेस में रुकावट डालते हैं. यह समझने के लिए कि Live API इस व्यवहार को कैसे मैनेज करता है, रुकावटों के बारे में Live API की गाइड देखें.

बाहरी टीटीएस के ज़रिए ऑडियो आउटपुट

Gemini Robotics ER 2, टेक्स्ट दिखाता है. आपका ऐप्लिकेशन, पूरे हो चुके जवाबों को इंजेक्ट किए गए कॉलबैक के ज़रिए, किसी दूसरे टीटीएस प्रोवाइडर (जैसे, Gemini TTS) को रूट करता है. इससे, स्पीच की लेटेन्सी, आवाज़ का चुनाव, और रुकावट का व्यवहार आपके कंट्रोल में रहता है. साथ ही, एजेंट लॉजिक में बदलाव किए बिना, टीटीएस बैकएंड को बदला जा सकता है.

टीटीएस को टूल के तौर पर भी एलान किया जा सकता है, ताकि मॉडल, "कुछ बोलो" को "हाथ घुमाओ" की तरह ही ट्रीट करे. पहले सेक्शन की tools सूची में, यह फ़ंक्शन एलान जोड़ें:

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

टीटीएस को फ़ंक्शन एलान में रैप करके, मॉडल, स्पीच को उसी टूल-कॉल पाथ के ज़रिए मैनेज करता है जिस तरह किसी अन्य रोबोट कार्रवाई को मैनेज करता है. आपका ऐप्लिकेशन, इंजेक्ट किए गए कॉलबैक के ज़रिए कॉल को पूरा करता है.

GitHub पर मौजूद उदाहरण

Spot रोबोट के स्नैक-फ़ेच डेमो और Tinybot के पैन-टिल्ट हैलो वर्ल्ड सहित, काम करने वाले पूरे उदाहरण देखने के लिए, Robotics Live API के उदाहरण देखें.

आगे क्या करना है

• वीडियो को समझना — मोमेंट ढूंढना और प्रोग्रेस क्लासिफ़िकेशन.
• टास्क ऑर्केस्ट्रेशन — स्ट्रीमिंग के बिना, लंबे समय तक चलने वाले टास्क.
• Live API की खास जानकारी — Live API का पूरा दस्तावेज़.