Gemini Live + Reachy Mini：打造可对话的桌面机器人

┌─────────────┐
│  Your voice │  Microphone audio (16-bit PCM, 16 kHz)
└──────┬──────┘
       ▼
┌─────────────────────────────────────┐
│  fastrtc  (low-latency WebRTC I/O)  │
│  ─ streams audio to/from the LLM    │
│  ─ resamples between sample rates   │
└──────┬──────────────────┬───────────┘
       │                  │
       ▼                  ▼
┌──────────────┐   ┌──────────────────┐
│  Gemini Live │   │  OpenAI Realtime │   (pick one via MODEL_NAME)
│  Handler     │   │  Handler         │
└──────┬───────┘   └──────┬───────────┘
       │                  │
       ▼                  ▼
┌─────────────────────────────────────┐
│  Tool dispatch layer                │
│  ─ dance, play_emotion, camera,     │
│    move_head, head_tracking, ...    │
└──────┬──────────────────────────────┘
       ▼
┌─────────────────────────────────────┐
│  MovementManager  (60 Hz loop)      │
│  ─ sequential primary moves         │
│  ─ additive secondary offsets       │
│    (speech wobble + face tracking)  │
│  ─ idle breathing                   │
└──────┬──────────────────────────────┘
       ▼
┌─────────────┐
│ Reachy Mini │  Robot hardware / simulator
└─────────────┘

# 1. Microphone → Gemini
async def receive(self, frame):
    pcm_bytes = audio_to_int16(frame).tobytes()
    await self.session.send_realtime_input(
        audio=types.Blob(data=pcm_bytes, mime_type="audio/pcm;rate=16000")
    )

# 2. Gemini → Speaker
async def _run_live_session(self):
    async with client.aio.live.connect(model=..., config=...) as session:
        async for response in session.receive():
            if response.server_content and response.server_content.model_turn:
                for part in response.server_content.model_turn.parts:
                    audio_array = np.frombuffer(part.inline_data.data, dtype=np.int16)
                    await self.output_queue.put((24000, audio_array))

            if response.tool_call:
                await self._handle_tool_call(response)

LLM says: "dance(name='macarena')"
  → BackgroundToolManager starts a task
  → Task calls MovementManager.queue_move(MacarenaMove)
  → Result sent back to the LLM so it can narrate what happened

async def _video_sender_loop(self):
    while not self._stop_event.is_set():
        frame = self.deps.camera_worker.get_latest_frame()
        _, buffer = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 70])
        await self.session.send_realtime_input(
            video=types.Blob(data=buffer.tobytes(), mime_type="image/jpeg")
        )
        await asyncio.sleep(1.0)

# Clone the repo
git clone https://github.com/pollen-robotics/reachy_mini_conversation_app.git
cd reachy_mini_conversation_app

# Create a virtual environment (macOS example)
uv venv --python python3.12 .venv
source .venv/bin/activate

# Install dependencies
uv sync

uv sync --extra mediapipe_vision   # Lightweight head tracking
uv sync --extra yolo_vision        # YOLO-based face detection
uv sync --extra local_vision       # On-device VLM (SmolVLM2, GPU recommended)
uv sync --extra all_vision         # Everything

cp .env.example .env

# Your Gemini API key — that's all you need to get started
GEMINI_API_KEY=your-gemini-api-key-here

# Navigate to wherever you cloned/installed the Reachy Mini SDK
cd path/to/reachy_mini
source reachy_mini_env/bin/activate

# Physical robot — auto-detects USB connection
reachy-mini-daemon

# Or simulation mode
reachy-mini-daemon --simulation

reachy-mini-conversation-app

reachy-mini-conversation-app --gradio

# With MediaPipe head tracking
reachy-mini-conversation-app --head-tracker mediapipe

# Audio-only (no camera)
reachy-mini-conversation-app --no-camera

# Verbose logging
reachy-mini-conversation-app --debug

# Connect to a specific robot on the network
reachy-mini-conversation-app --robot-name my-reachy

profiles/
├── default/
│   ├── instructions.txt   # System prompt
│   └── tools.txt          # Which tools are enabled
├── mars_rover/
│   ├── instructions.txt
│   └── tools.txt
├── noir_detective/
│   ├── instructions.txt
│   └── tools.txt
└── ...

mkdir profiles/pirate_captain

## IDENTITY
You are Captain Byte, a swashbuckling robot pirate who speaks in nautical
metaphors and ends every sentence with "Arrr" or a pirate-themed quip.

## RESPONSE RULES
Keep responses to 1-2 sentences. Be helpful first, pirate second.
Always refer to the user as "matey" or "landlubber".

dance
play_emotion
move_head
camera
head_tracking

# In your .env file
REACHY_MINI_CUSTOM_PROFILE="pirate_captain"

# instructions.txt
[identities/witty_identity]
[passion_for_lobster_jokes]
You love to dance and will look for any excuse to bust a move.

# profiles/example/sweep_look.py
from reachy_mini_conversation_app.tools.core_tools import Tool

class SweepLookTool(Tool):
    name = "sweep_look"
    description = "Slowly look around the room in a sweeping motion."

    async def run(self, args, deps):
        # Queue a sequence of head movements...
        return {"status": "done", "description": "Finished looking around"}

dance
play_emotion
sweep_look    # Your custom tool

live_config = types.LiveConnectConfig(
    response_modalities=[types.Modality.AUDIO],
    system_instruction=types.Content(parts=[types.Part(text=instructions)]),
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Kore"),
        ),
    ),
    tools=[{"function_declarations": declarations}],
    input_audio_transcription=types.AudioTranscriptionConfig(),
    output_audio_transcription=types.AudioTranscriptionConfig(),
)

"You've been idle for a while. Feel free to get creative — dance, 
show an emotion, look around, do nothing, or just be yourself!"