স্ট্রিমিং সহ রোবোটিক্স

gemini-robotics-er-2-streaming-preview মডেল এন্ডপয়েন্টটি একটি ডেডিকেটেড স্ট্রিমিং এন্ডপয়েন্ট প্রদান করে যা লাইভ এপিআই (Live API)- এর সাথে ইন্টিগ্রেট করে, এবং আপনার অ্যাপ্লিকেশন ও রোবটের মধ্যে রিয়েল-টাইম, দ্বিমুখী মিথস্ক্রিয়া সক্ষম করে। এটি এমন এজেন্টদের জন্য উপযুক্ত যাদের দ্রুত ফিডব্যাক লুপ এবং পরিবেশের প্রতি প্রতিক্রিয়াশীল সাড়া প্রয়োজন।

ব্যবহারের ক্ষেত্র

• বহু-রোবট সমন্বয় : একাধিক রোবট একটি সাধারণ সেশনের মাধ্যমে কাজের অবস্থা সম্পর্কে যোগাযোগ করে এবং উপ-কাজ অর্পণ করে।
• নিরবচ্ছিন্ন পর্যবেক্ষণ : এমন রোবট যা কোনো দৃশ্য পর্যবেক্ষণ করে এবং নির্দিষ্ট ঘটনা ঘটলে, যেমন কোনো পাত্র পূর্ণ হয়ে গেলে, সেই অনুযায়ী কাজ সম্পাদন করে।
• গুদাম ও লজিস্টিকস : পিক-অ্যান্ড-প্যাক এজেন্ট যারা পণ্য দৃশ্যত যাচাই করেন, প্যাকিংয়ের অগ্রগতি পর্যবেক্ষণ করেন এবং ভুল সংশোধন করেন।

প্রযুক্তিগত বিবরণ

নিম্নলিখিত সারণিতে লাইভ এপিআই-এর প্রযুক্তিগত বিবরণ তুলে ধরা হলো:

একটি এজেন্টিক সেটআপ তৈরি করুন

Live API-এর উপর ভিত্তি করে নির্মিত প্রতিটি রোবোটিক্স এজেন্ট তিনটি ধাপ অনুসরণ করে:

1. রোবটের সক্ষমতাগুলোকে টুল হিসেবে ঘোষণা করুন। রোবটের প্রতিটি কাজ—যেমন দিক নির্ণয় করা, ধরা, কথা বলা—একটি নাম, বিবরণ এবং প্যারামিটার স্কিমা সহ একটি ফাংশন ডিক্লারেশন হয়ে ওঠে। শারীরিক কাজগুলোর জন্য অবশ্যই "behavior": "BLOCKING" ব্যবহার করতে হবে, যাতে মডেলটি পরবর্তী পদক্ষেপ নেওয়ার আগে রোবটের কাজ শেষ হওয়ার জন্য অপেক্ষা করে।
2. একটি স্থায়ী সেশনে মাল্টিমোডাল ইনপুট স্ট্রিম করুন। একটি live.connect সেশন খুলুন এবং কাজটি চলাকালীন তা খোলা রাখুন। আপনার রোবটের সেন্সর থেকে ভিডিও ফ্রেম, অডিও বা টেক্সট আসা মাত্রই পাঠিয়ে দিন।
3. একটি রিসিভ লুপের মাধ্যমে টুল কলগুলো পরিচালনা করুন। প্রতিবার মডেল যখন কোনো অ্যাকশন নির্বাচন করে, তখন এটি একটি tool_call মেসেজ পাঠায়। আপনার রিসিভ লুপটি আপনার রোবট SDK-এর ফাংশনটি এক্সিকিউট করে এবং একটি tool_response ফেরত পাঠায়। সেশনটি খোলা থাকে এবং মডেলটি ফলাফলের উপর ভিত্তি করে পরবর্তী অ্যাকশনটি বেছে নেয়।

নিম্নলিখিত বিভাগগুলিতে তিনটি সাধারণ প্যাটার্নে এই পদক্ষেপগুলি কীভাবে প্রয়োগ করতে হয় তা দেখানো হয়েছে: একটি বেসলাইন এজেন্ট লুপ, হার্টবিটের মাধ্যমে সক্রিয় দৃশ্য পর্যবেক্ষণ, এবং একটি সরঞ্জাম হিসাবে টিটিএস-এর মাধ্যমে স্পিচ রাউটিং।

ফাংশন কলিংয়ের মাধ্যমে একটি রোবটকে পরিচালনা করুন

নিম্নলিখিত উদাহরণটি দেখায় কীভাবে এই তিনটি ধাপ একটি একক পাইথন স্ক্রিপ্টে একসাথে সংযুক্ত করা হয়েছে।

ধাপ ১ — টুল ডেফিনিশন — রোবটের সক্ষমতাগুলোকে ফাংশন ডিক্লারেশন হিসেবে ঘোষণা করে। navigate ফাংশনটি "behavior": "BLOCKING" ব্যবহার করে, ফলে মডেলটি অন্য কোনো টুল কল করার আগে রোবটের ওয়েপয়েন্টে পৌঁছানোর জন্য অপেক্ষা করে। রোবটের অতিরিক্ত সক্ষমতাগুলো প্রকাশ করার জন্য একই লিস্টে আরও ফাংশন ডিক্লারেশন যোগ করুন।

ধাপ ২ — ইনপুট হেল্পার — তিনটি ফাংশন দেখায় যা সেশনে বিভিন্ন মোডালিটির ইনপুট স্ট্রিম করে: কমান্ডের জন্য send_text , ঐচ্ছিক টেক্সট প্রম্পট সহ ক্যামেরা ফ্রেমের জন্য send_image , এবং মাইক্রোফোন থেকে র PCM অডিওর জন্য send_audio ।

ধাপ ৩ — রিসিভ লুপ — সমান্তরালভাবে চলে এবং দুই ধরনের বার্তা পরিচালনা করে: server_content বার্তা (মডেলের টেক্সট আউটপুট) এবং tool_call বার্তা (মডেলের পক্ষ থেকে কোনো রোবট অ্যাকশনের অনুরোধ)। যখন একটি টুল কল আসে, লুপটি execute_tool কল করে — এটি একটি স্টাব যা আপনি আপনার আসল রোবট SDK দিয়ে প্রতিস্থাপন করবেন — তারপর একটি 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())

প্রতিটি টুল প্রতিক্রিয়ার পর রিসিভ লুপটি সক্রিয় থাকে। আপনার সম্পূর্ণ অ্যাকশন সিকোয়েন্সটি আগে থেকে এনকোড করা ছাড়াই মডেলটি একটি দীর্ঘমেয়াদী পরিকল্পনা তৈরি ও সংশোধন করে।

সক্রিয় স্থানিক-কালিক যুক্তি

লাইভ এপিআই ভিডিও স্ট্রিম করে, কিন্তু শুধুমাত্র ভিডিও ফ্রেম একটি নতুন রিজনিং টার্ন ট্রিগার করে না। একটি মডেল রেসপন্স ট্রিগার করার জন্য ভিডিও ফ্রেমের সাথে অবশ্যই একটি টেক্সট বা অডিও প্রম্পট থাকতে হবে। এর প্রেক্ষাপট জানতে লাইভ এপিআই ক্যাপাবিলিটিজ দেখুন।

সক্রিয় সিদ্ধান্ত গ্রহণ সক্ষম করতে, একটি হার্টবিট প্রয়োগ করুন: পর্যায়ক্রমে সর্বশেষ ক্যামেরা ফ্রেমটি পাঠান এবং এর সাথে একটি সংক্ষিপ্ত টেক্সট প্রম্পট দিন, যা মডেলটিকে দৃশ্যটি পরিদর্শন করতে এবং একটি সুস্পষ্ট সিদ্ধান্ত নিতে বাধ্য করে। ভিডিও ইনপুটের হার প্রতি সেকেন্ডে একটি ফ্রেমে সীমিত।

পূর্ববর্তী বিভাগের রিসিভ লুপের পাশে এই কো-রুটিনটি যোগ করুন। এটি একই সেশনে একটি পৃথক 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)

রোবটের কার্যকলাপ চলাকালীন হার্টবিট থামানোর প্রয়োজন নেই। যখন এটিকে একটি পরোক্ষ সাফল্য শনাক্তকারী হিসেবে ব্যবহার করা হয়, তখন এটিকে চালু রাখলে মডেলটি চলমান কাজটি ক্রমাগত পর্যবেক্ষণ করতে পারে — যেমন, কোনো কিছু ধরা নিরাপদ কিনা, ঢালা সঠিক জায়গায় হচ্ছে কিনা, বা কোনো বস্তু ঠিকমতো বসছে কিনা — এবং ফলাফল স্পষ্ট হওয়ার মুহূর্তেই প্রতিক্রিয়া দেখাতে পারে।

বাহ্যিক টিটিএস এর মাধ্যমে অডিও আউটপুট

জেমিনি রোবোটিক্স ইআর ২ টেক্সট ফেরত দেয়। আপনার অ্যাপ্লিকেশন একটি ইনজেক্টেড কলব্যাকের মাধ্যমে সম্পন্ন হওয়া প্রতিক্রিয়াগুলোকে একটি পৃথক টিটিএস প্রোভাইডারের (যেমন জেমিনি টিটিএস ) কাছে পাঠিয়ে দেয়। এর ফলে স্পিচ ল্যাটেন্সি, ভয়েস সিলেকশন এবং ইন্টারাপশন আচরণ আপনার নিয়ন্ত্রণে থাকে এবং এজেন্ট লজিক পরিবর্তন না করেই আপনি টিটিএস ব্যাকএন্ড অদলবদল করতে পারেন।

আপনি TTS-কে একটি টুল হিসেবেও ঘোষণা করতে পারেন, যাতে মডেলটি 'say something' এবং 'move the arm'-কে একই ভাবে বিবেচনা করে। প্রথম সেকশন থেকে আপনার 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"],
        },
    },
]

TTS-কে একটি ফাংশন ডিক্লারেশনের মধ্যে রাখার মাধ্যমে, মডেলটি অন্য যেকোনো রোবট অ্যাকশনের মতোই একই টুল-কল পাথ ব্যবহার করে স্পিচ পরিচালনা করে। আপনার অ্যাপ্লিকেশনটি একটি ইনজেক্টেড কলব্যাকের মাধ্যমে এই কলটি সম্পন্ন করে।

গিটহাবে উদাহরণ

স্পট রোবটের স্নাক-ফেচ ডেমো এবং টাইনিবটের প্যান-টিল্ট হ্যালো ওয়ার্ল্ড সহ সম্পূর্ণ কার্যকরী উদাহরণের জন্য, রোবোটিক্স লাইভ এপিআই উদাহরণসমূহ দেখুন।

এরপর কী?

• ভিডিও অনুধাবন — মুহূর্ত শনাক্তকরণ ও অগ্রগতি শ্রেণিবিন্যাস।
• টাস্ক অর্কেস্ট্রেশন — স্ট্রিমিং ছাড়াই দীর্ঘমেয়াদী কাজ।
• লাইভ এপিআই ওভারভিউ — সম্পূর্ণ লাইভ এপিআই ডকুমেন্টেশন।