Lesson 19: Connecting BonicBot with an LLM

python3 -m venv bonicbot-env
source bonicbot-env/bin/activate   # On Windows: bonicbot-env\Scripts\activate
pip install bonicbot-bridge opencv-python numpy openai pydantic piper-tts pyaudio

import json
import sys
import time
from pathlib import Path
from typing import List, Literal, Optional
 
import pyaudio
from openai import OpenAI, APIError, AuthenticationError, RateLimitError
from pydantic import BaseModel, Field, ValidationError
from piper import PiperVoice, SynthesisConfig
from piper.download_voices import download_voice
from bonicbot_bridge import BonicBot
from bonicbot_bridge.exceptions import BonicBotError
 
 
# ============================================================================
# CONFIG
# ============================================================================
HOST = 'localhost'  # replace with your BonicBot's IP address
 
# OpenRouter's API is OpenAI-compatible, so the same client library works —
# just point base_url at OpenRouter instead of OpenAI's servers.
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
 
# Optional. Lets this app show up (attributed) in OpenRouter's own
# dashboards/rankings — neither header is required for requests to work.
APP_URL = "https://example.com/bonicbot-lessons"   # replace with your own
APP_TITLE = "BonicBot the Improviser"
 
# Paste your key from https://openrouter.ai/keys here. Keep this file out
# of any public repo or shared folder if you hardcode it this way.
OPENROUTER_API_KEY = "your-key-here"
 
# NVIDIA's Nemotron 3 Ultra, free on OpenRouter (the `:free` variant).
# Free models rotate on OpenRouter fairly often — check
# https://openrouter.ai/models?fmt=free if this slug ever stops resolving.
MODEL_NAME = "nvidia/nemotron-3-ultra-550b-a55b:free"
 
PIPER_VOICE_NAME = "en_US-lessac-medium"
PIPER_VOICES_DIR = Path(__file__).resolve().parent / "piper_voices"
 
# How many rounds of tool-calling we allow for a single typed instruction
# before giving up. A simple instruction resolves in 1-2 rounds; this is
# just a safety cap against a runaway back-and-forth.
MAX_TOOL_ROUNDS = 6
 
# --- Neck: yaw only, a continuous angle rather than named positions -------
# -90 = fully left, 0 = center, 90 = fully right (the SDK's documented
# neck yaw joint limit).
NECK_YAW_RANGE = (-90.0, 90.0)
 
# --- Arm: named presets, pre-checked against the SDK's documented joint
# limits (shoulder pitch -45..180 deg, elbow 0..50 deg) so Nemotron can only
# request a pose that's physically safe to send straight to move_*_arm().
# Identical table to Lesson 18 on purpose. ----------------------------------
ArmPose = Literal[
    "rest", "wave", "reach_forward", "shrug",
    "droop", "celebrate", "thinking", "hug_self",
]
ARM_PRESETS = {
    # name             (shoulder_deg, elbow_deg)  -- meaning
    "rest":             (0.0,  0.0),   # neutral, arm relaxed at the side
    "wave":             (90.0, 30.0),  # greeting / hello / goodbye
    "reach_forward":    (90.0, 0.0),   # offering, pointing, handing something over
    "shrug":            (45.0, 50.0),  # unsure, confused, "I don't know"
    "droop":            (-40.0, 15.0), # sad, tired, deflated
    "celebrate":        (170.0, 10.0), # excited, proud, cheering
    "thinking":         (60.0, 50.0),  # curious, pondering, hand drawn in
    "hug_self":         (30.0, 50.0),  # shy, nervous, self-comforting
}
SHOULDER_RANGE = (0, 180)   # from the SDK's documented joint limits
ELBOW_RANGE = (0, 50)
 
# --- Move: precise motion, exact distance/angle instead of speed+duration --
# bot.drive_distance's own guard rejects anything past 10.0 m (raises
# PreciseMotionError), so that's the hard bound here too. Rotation has no
# documented guard, so two full turns either way is a generous sanity bound.
DRIVE_DISTANCE_RANGE = (-10.0, 10.0)     # meters, negative = backward
ROTATE_ANGLE_RANGE = (-720.0, 720.0)     # degrees, negative = left
 
PRECISE_ENGINE = "internal"   # no mapping/nav2 stack needed for open-loop moves
DEFAULT_DRIVE_SPEED = 0.2     # m/s
DEFAULT_ROTATE_SPEED = 30.0   # deg/s
 
 
def estimate_timeout(magnitude, speed, min_timeout=5.0, buffer_factor=1.5, buffer_seconds=5.0):
    """A blocking precise-motion call needs a timeout long enough to actually
    finish the move, or the SDK will cut it off early -- exactly the kind of
    'moved partway, then stopped' the fixed-duration approach ran into.
    Estimating from distance/angle and speed instead of guessing a constant
    avoids that."""
    if speed <= 0:
        return min_timeout
    return max(min_timeout, abs(magnitude) / speed * buffer_factor + buffer_seconds)
 
SYSTEM_PROMPT = (
    "You are the onboard controller for a small robot named BonicBot. You "
    "can ONLY affect the robot by calling the tools you've been given — "
    "there is no other way to move it or make it speak. Given a short "
    "instruction, call whichever tools it needs, in the order the actions "
    "should happen — you may call more than one tool in the same turn. "
    "The neck moves to a yaw angle in degrees, from -90 (fully left) to 90 "
    "(fully right), with 0 as straight ahead — a small glance is closer to "
    "20-30 degrees, a full look to one side is closer to 60-90. BonicBot "
    "cannot look up or down, cannot use grippers, and cannot strafe "
    "sideways — only turn in place. Keep duration_seconds between "
    "0.5 and 5.0. Use the speak tool at whatever point in the sequence a "
    "spoken line actually belongs, not just at the very end. Once the "
    "instruction has been fully carried out, reply with a short plain-text "
    "line and no further tool calls."
)
 
 
# ============================================================================
# TOOL DEFINITIONS — these are Nemotron's only way to affect the real world.
# Each one's JSON Schema mirrors the same field constraints Lesson 18
# expressed as a Pydantic schema (the enums and duration bounds are
# identical), just reshaped for tool calling.
# ============================================================================
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "neck_gesture",
            "description": (
                "Turn BonicBot's neck to a specific yaw angle in degrees, "
                "hold it briefly, then it returns to center (0) on its own. "
                "Negative degrees turn left, positive degrees turn right, "
                "0 is straight ahead. Use for attention, curiosity, or "
                "acknowledging someone off to one side."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "neck_yaw_deg": {
                        "type": "number",
                        "minimum": -90,
                        "maximum": 90,
                        "description": "Yaw angle in degrees: negative = left, positive = right, 0 = center.",
                    },
                    "duration_seconds": {
                        "type": "number",
                        "minimum": 0.5,
                        "maximum": 5.0,
                        "description": "How long to hold the angle before returning to center.",
                    },
                },
                "required": ["neck_yaw_deg"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "arm_gesture",
            "description": (
                "Move one arm into a named emotional pose, hold it, then "
                "return it to rest. Poses: rest (neutral), wave (greeting), "
                "reach_forward (offering/pointing), shrug (unsure), droop "
                "(sad/tired), celebrate (excited/proud), thinking "
                "(curious/pondering), hug_self (shy/nervous)."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "arm_side": {"type": "string", "enum": ["left", "right"]},
                    "arm_pose": {"type": "string", "enum": list(ARM_PRESETS.keys())},
                    "duration_seconds": {
                        "type": "number",
                        "minimum": 0.5,
                        "maximum": 5.0,
                    },
                },
                "required": ["arm_side", "arm_pose"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "move_gesture",
            "description": (
                "Drive the body forward/backward or turn it left/right in "
                "place, for a fixed duration. BonicBot cannot strafe "
                "sideways — only drive straight or turn in place."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "move_action": {
                        "type": "string",
                        "enum": ["move_forward", "move_backward", "turn_left", "turn_right"],
                    },
                    "duration_seconds": {
                        "type": "number",
                        "minimum": 0.5,
                        "maximum": 5.0,
                    },
                },
                "required": ["move_action"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "speak",
            "description": "Say one short, friendly sentence out loud through BonicBot's speaker.",
            "parameters": {
                "type": "object",
                "properties": {
                    "speech_text": {"type": "string", "maxLength": 200},
                },
                "required": ["speech_text"],
            },
        },
    },
]
 
 
# ============================================================================
# PYDANTIC ARG MODELS — a JSON Schema enum keeps Nemotron *mostly* honest,
# but arguments are still just a string it generated. These give the same
# defense-in-depth validation Lesson 18 did before anything reaches the SDK.
# ============================================================================
class NeckArgs(BaseModel):
    neck_yaw_deg: float = Field(ge=NECK_YAW_RANGE[0], le=NECK_YAW_RANGE[1])
    duration_seconds: float = Field(default=1.5, ge=0.5, le=5.0)
 
 
class ArmArgs(BaseModel):
    arm_side: Literal["left", "right"]
    arm_pose: ArmPose
    duration_seconds: float = Field(default=1.5, ge=0.5, le=5.0)
 
MoveAction = Literal["move_forward", "move_backward", "turn_left", "turn_right"]
 
class MoveArgs(BaseModel):
    move_action: MoveAction
    duration_seconds: float = Field(default=1.5, ge=0.5, le=5.0)
 
 
class SpeakArgs(BaseModel):
    speech_text: str = Field(max_length=200)
 
 
TOOL_ARG_MODELS = {
    "neck_gesture": NeckArgs,
    "arm_gesture": ArmArgs,
    "move_gesture": MoveArgs,
    "speak": SpeakArgs,
}
 
 
# ============================================================================
# GESTURE FUNCTIONS — identical to Lesson 18, SDK controls only, every
# angle clamped to safe ranges regardless of what already passed validation.
# ============================================================================
def clamp(value, min_val, max_val):
    return max(min_val, min(max_val, value))
 
 
def neck_gesture(bot, neck_yaw_deg, duration_seconds):
    neck_yaw_deg = clamp(neck_yaw_deg, *NECK_YAW_RANGE)
    bot.set_neck(neck_yaw_deg)
    time.sleep(duration_seconds)
    if neck_yaw_deg != 0.0:
        bot.set_neck(0.0)
 
 
def arm_gesture(bot, arm_side, arm_pose, duration_seconds):
    shoulder_deg, elbow_deg = ARM_PRESETS[arm_pose]
    shoulder_deg = clamp(shoulder_deg, *SHOULDER_RANGE)
    elbow_deg = clamp(elbow_deg, *ELBOW_RANGE)
    move = bot.move_left_arm if arm_side == "left" else bot.move_right_arm
 
    move(shoulder_deg, elbow_deg, wait=True)
    time.sleep(duration_seconds)
    if arm_pose != "rest":
        move(0.0, 0.0, wait=True)
 
 
def move_gesture(bot, move_action, duration_seconds):
    if move_action == "move_forward":
        bot.move_forward(speed=DEFAULT_DRIVE_SPEED, duration=duration_seconds)
    elif move_action == "move_backward":
        bot.move_backward(speed=DEFAULT_DRIVE_SPEED, duration=duration_seconds)
    elif move_action == "turn_left":
        bot.turn_left(speed=DEFAULT_ROTATE_SPEED, duration=duration_seconds)
    elif move_action == "turn_right":
        bot.turn_right(speed=DEFAULT_ROTATE_SPEED, duration=duration_seconds)
 
 
def speak_step(piper_voice, pyaudio_instance, text):
    print(f"  🗣️  BonicBot says: \"{text}\"")
    syn_config = SynthesisConfig(length_scale=1.0, volume=1.0)
    stream = None
    try:
        for chunk in piper_voice.synthesize(text, syn_config=syn_config):
            if stream is None:
                stream = pyaudio_instance.open(
                    format=pyaudio_instance.get_format_from_width(chunk.sample_width),
                    channels=chunk.sample_channels,
                    rate=chunk.sample_rate,
                    output=True,
                )
            stream.write(chunk.audio_int16_bytes)
    finally:
        if stream is not None:
            stream.stop_stream()
            stream.close()
 
 
# ============================================================================
# EXECUTING A SINGLE TOOL CALL — validate, then perform, then report back.
# Whatever this returns becomes the "tool" message Nemotron sees next, so a
# validation failure or a BonicBotError becomes something it can react
# to on its very next turn, instead of silently vanishing.
# ============================================================================
def execute_tool_call(bot, piper_voice, pyaudio_instance, name, raw_arguments):
    model_cls = TOOL_ARG_MODELS.get(name)
    if model_cls is None:
        return f"error: unknown tool '{name}'"
 
    try:
        args = model_cls.model_validate_json(raw_arguments)
    except ValidationError as e:
        return f"error: invalid arguments for {name}: {e}"
 
    try:
        if name == "neck_gesture":
            print(f"  🤖 bot.set_neck({args.neck_yaw_deg})")
            neck_gesture(bot, args.neck_yaw_deg, args.duration_seconds)
        elif name == "arm_gesture":
            method = "move_left_arm" if args.arm_side == "left" else "move_right_arm"
            print(f"  🤖 bot.{method}(...)  # {args.arm_pose}")
            arm_gesture(bot, args.arm_side, args.arm_pose, args.duration_seconds)
        elif name == "move_gesture":
            print(f"  🤖 bot.{args.move_action}(...)")
            move_gesture(bot, args.move_action, args.duration_seconds)
        elif name == "speak":
            speak_step(piper_voice, pyaudio_instance, args.speech_text)
        return "ok"
    except BonicBotError as e:
        return f"error: robot rejected this action: {e}"
 
 
# How many times we'll retry a single model call if OpenRouter hands back a
# 200 OK with no usable choice at all -- a known hiccup with free, shared
# models under load. This is separate from MAX_TOOL_ROUNDS below: it's not
# about the conversation going on too long, it's about one specific call
# coming back empty.
MAX_EMPTY_RESPONSE_RETRIES = 2
EMPTY_RESPONSE_RETRY_DELAY = 1.5  # seconds
 
 
def call_model(client, messages):
    """Calls the model and guards against a 200 OK response that still has
    no usable choice in it -- free, community-hosted models occasionally do
    this under load instead of raising a proper error. Returns None if every
    retry comes back empty, so the caller can bail out of this instruction
    cleanly instead of crashing on response.choices[0]."""
    for attempt in range(1, MAX_EMPTY_RESPONSE_RETRIES + 2):
        response = client.chat.completions.create(
            model=MODEL_NAME,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
            parallel_tool_calls=True,
            temperature=0.3,
        )
        if response.choices:
            return response
 
        if attempt <= MAX_EMPTY_RESPONSE_RETRIES:
            print(f"  ⚠️ Empty response from OpenRouter (attempt {attempt}) — retrying...")
            time.sleep(EMPTY_RESPONSE_RETRY_DELAY)
 
    return None
 
 
# ============================================================================
# THE AGENT LOOP — Nemotron picks tools, we run them, it sees the results and
# decides whether to keep going or wrap up. This replaces Lesson 18's
# "generate the whole plan, then perform it" flow with something closer to
# a live back-and-forth.
# ============================================================================
def run_agent_turn(client, bot, piper_voice, pyaudio_instance, instruction):
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": instruction},
    ]
 
    for round_num in range(1, MAX_TOOL_ROUNDS + 1):
        response = call_model(client, messages)
        if response is None:
            print("  ⚠️ OpenRouter/Nemotron kept returning empty responses for this "
                  "instruction -- a known hiccup with free, shared models under load. "
                  "Try again in a moment, or check openrouter.ai/models?fmt=free for a "
                  "less-loaded free model to swap into MODEL_NAME.")
            return
 
        message = response.choices[0].message
        messages.append(message.model_dump(exclude_none=True))
 
        if not message.tool_calls:
            if message.content:
                print(f"  💬 Nemotron: {message.content}")
            return
 
        # Tool calls in one round may be requested "in parallel" by the API,
        # but a physical robot can't actually do two blocking motions at
        # once — they're executed one at a time, in the order Nemotron listed
        # them, and each result goes back before the next round starts.
        for tool_call in message.tool_calls:
            result = execute_tool_call(
                bot, piper_voice, pyaudio_instance,
                tool_call.function.name,
                tool_call.function.arguments,
            )
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })
 
    print("  ⚠️ Hit the tool-round limit — stopping this instruction.")
 
 
# ============================================================================
# SETUP HELPERS — OpenRouter client + Piper voice, external to the BonicBot SDK
# ============================================================================
def ensure_openrouter_client():
    if OPENROUTER_API_KEY == "your-key-here":
        print("⚠️ Paste your real key into OPENROUTER_API_KEY near the top of the script.")
        print("   Get one from https://openrouter.ai/keys")
        sys.exit(1)
    return OpenAI(
        base_url=OPENROUTER_BASE_URL,
        api_key=OPENROUTER_API_KEY,
        default_headers={"HTTP-Referer": APP_URL, "X-Title": APP_TITLE},
    )
 
 
def ensure_piper_voice_ready(voice_name, voices_dir):
    voices_dir.mkdir(parents=True, exist_ok=True)
    model_path = voices_dir / f"{voice_name}.onnx"
    config_path = voices_dir / f"{voice_name}.onnx.json"
    if model_path.exists() and config_path.exists():
        return model_path, config_path
    print(f"Downloading voice '{voice_name}'...")
    download_voice(voice_name, voices_dir)
    return model_path, config_path
 
 
# ============================================================================
# MAIN
# ============================================================================
def main():
    client = ensure_openrouter_client()
 
    model_path, config_path = ensure_piper_voice_ready(PIPER_VOICE_NAME, PIPER_VOICES_DIR)
    piper_voice = PiperVoice.load(model_path, config_path=config_path)
    pyaudio_instance = pyaudio.PyAudio()
 
    try:
        with BonicBot(host=HOST, port=9090, timeout=10) as bot:
            print("✅ Connected. Type an instruction, or 'quit' to stop.")
            print('   Example: "turn left, then wave and say hi"\n')
 
            while True:
                instruction = input("Instruction> ").strip()
                if not instruction:
                    continue
                if instruction.lower() in {"quit", "exit"}:
                    break
 
                try:
                    run_agent_turn(client, bot, piper_voice, pyaudio_instance, instruction)
                except AuthenticationError:
                    print("⚠️ OpenRouter rejected the API key — check OPENROUTER_API_KEY near the top of the script.")
                    break
                except RateLimitError:
                    print("⚠️ Hit OpenRouter's free-tier rate limit (20/min, 50-1000/day) — wait a bit and try again.")
                except APIError as e:
                    print(f"⚠️ OpenRouter API error: {e}")
 
    except BonicBotError as e:
        print(f"⚠️ Robot error: {e}")
    finally:
        pyaudio_instance.terminate()
 
 
if __name__ == "__main__":
    main()

✅ Connected. Type an instruction, or 'quit' to stop.
   Example: "turn left, then wave and say hi"

Instruction> glance to the right and say hi

  🤖 bot.set_neck(45.0)
  🗣️  BonicBot says: "Hi there!"
  💬 Nemotron: Done — glanced right and said hello.