「Gemini Live API for Robotics」最前線：AIとロボットのリアルタイム対話が未来を創る！

import asyncio
from google import genai
from google.genai import types
import os # 環境変数からAPIキーを取得するために追加

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})")
    # ここに実際のロボットSDK呼び出しを実装します
    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.
        # 例: send_text(session, "Go to the kitchen.")
        # 例: await asyncio.sleep(10) # ロボットが動作するのを待つ
        # 例: send_image(session, camera.get_latest_frame(), "What do you see?")
        recv_task.cancel()

if __name__ == "__main__":
    asyncio.run(main())