Lesson 17: Writing BonicBot’s Performance Script

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

import json
import shutil
import subprocess
import sys
import time
from typing import List, Literal, Optional
 
import ollama
from pydantic import BaseModel, Field, ValidationError, model_validator
 
 
OLLAMA_MODEL = "qwen3.5:0.8b"
 
SERVER_START_TIMEOUT = 15.0
SERVER_POLL_INTERVAL = 0.5
 
MAX_VALIDATION_ATTEMPTS = 3
 
CHAT_OPTIONS = {"temperature": 0.2, "top_p": 0.8, "num_predict": 500}
CHAT_KEEP_ALIVE = "30m"
 
NeckPose = Literal["look_left", "look_center", "look_right"]
 
ArmPose = Literal[
    "rest", "wave", "reach_forward", "shrug",
    "droop", "celebrate", "thinking", "hug_self",
]
 
MoveAction = Literal["move_forward", "move_backward", "turn_left", "turn_right"]
 
SYSTEM_PROMPT = (
    "You are a choreographer for a small robot named BonicBot. Given a short "
    "instruction, write 2 to 5 steps mixing these action types:\n"
    "- speak: say one short, friendly sentence\n"
    "- neck: turn the head using neck_pose = look_left, look_center, or look_right\n"
    "- arm: move one arm using arm_side = left or right, and arm_pose, one of:\n"
    "    rest (neutral), wave (greeting), reach_forward (offering/pointing),\n"
    "    shrug (unsure), droop (sad/tired), celebrate (excited/proud),\n"
    "    thinking (curious/pondering), hug_self (shy/nervous)\n"
    "- move: drive the body using move_action = move_forward, move_backward, "
    "turn_left, or turn_right. For move_forward or move_backward, set "
    "move_distance_meters (0.1 to 3.0) for how far to travel. For turn_left "
    "or turn_right, set turn_degrees (1 to 360) for how far to rotate — e.g. "
    "a quarter turn is 90, a full turn is 360. The system converts this to "
    "real motor timing automatically; never invent a duration yourself for "
    "a move step.\n"
    "Only ever use these exact literal values, and keep duration_seconds "
    "(for speak/neck/arm steps) between 0.5 and 5.0. BonicBot cannot look "
    "up or down, cannot use grippers, and cannot strafe sideways — only "
    "turn in place.\n"
    "IMPORTANT: 'reach_forward' is an ARM pose (one arm reaches out), not the "
    "same thing as the robot actually driving forward. If the instruction is "
    "literally about the robot moving/walking/driving/going somewhere (e.g. "
    "'move forward', 'go forward', 'drive back', 'turn around'), you MUST "
    "include a 'move' step with the matching move_action — don't substitute "
    "an arm gesture for it, and don't skip it.\n"
    "Each step also has an optional run_with_next flag (default false). Set "
    "it to true on the FIRST of two steps that should happen at the exact "
    "same time as the step right after it — for example if an arm gesture "
    "comes before a speak step and they should happen together, set "
    "run_with_next=true on that arm step, not on the speak step that "
    "follows it. The flag always goes on the earlier step, never the later "
    "one. Leave it false for steps that should happen one after another, "
    "which is most of the time. Never set run_with_next to true on the "
    "last step in a plan, and never pair more than two steps together at "
    "once.\n"
    "Simultaneity isn't only signaled by the literal phrase 'at the same "
    "time' — continuous/ongoing phrasing also means two steps should "
    "overlap. If the instruction says something like 'keep moving forward "
    "while you talk', 'say hi and keep moving', 'wave as you turn', or "
    "'while driving forward, say...', set run_with_next=true on the "
    "earlier step so it overlaps with the one after it, the same as you "
    "would for 'at the same time'. Only leave run_with_next false when the "
    "instruction genuinely describes one thing happening, then another — "
    "e.g. 'turn around and walk away' is sequential (finish turning, THEN "
    "walk), not simultaneous.\n"
    "Base every plan strictly on the new instruction you're given — never "
    "reuse a previous example's wording or steps unless the new instruction "
    "genuinely calls for the same reaction."
)
 
 
class ChoreographyStep(BaseModel):
    action: Literal["speak", "neck", "arm", "move"]
 
    speech_text: Optional[str] = Field(default=None, max_length=200)
    neck_pose: Optional[NeckPose] = None
    arm_side: Optional[Literal["left", "right"]] = None
    arm_pose: Optional[ArmPose] = None
    move_action: Optional[MoveAction] = None
    turn_degrees: Optional[float] = Field(default=None, ge=1, le=360)
    move_distance_meters: Optional[float] = Field(default=None, ge=0.1, le=3.0)
 
    duration_seconds: float = Field(default=1.5, ge=0.5, le=5.0)
    run_with_next: bool = Field(default=False)
 
    @model_validator(mode="after")
    def fields_match_action(self):
        if self.action == "speak" and not self.speech_text:
            raise ValueError("a 'speak' step needs speech_text")
        if self.action == "neck" and not self.neck_pose:
            raise ValueError("a 'neck' step needs neck_pose")
        if self.action == "arm" and not (self.arm_side and self.arm_pose):
            raise ValueError("an 'arm' step needs arm_side and arm_pose")
        if self.action == "move":
            if not self.move_action:
                raise ValueError("a 'move' step needs move_action")
            if self.move_action in ("turn_left", "turn_right") and self.turn_degrees is None:
                raise ValueError("a turning move step needs turn_degrees")
            if self.move_action in ("move_forward", "move_backward") and self.move_distance_meters is None:
                raise ValueError("a forward/backward move step needs move_distance_meters")
        return self
 
 
class Choreography(BaseModel):
    title: str
    steps: List[ChoreographyStep] = Field(min_length=1, max_length=6)
 
    @model_validator(mode="after")
    def run_with_next_pairs_are_valid(self):
        last_index = len(self.steps) - 1
        for i, step in enumerate(self.steps):
            if step.run_with_next:
                if i == last_index:
                    raise ValueError(
                        "the last step can't set run_with_next=true — there's no next step to "
                        "pair with. If this step should run together with the one before it, "
                        "set run_with_next=true on that EARLIER step instead, and set this "
                        "last step's run_with_next back to false."
                    )
                if self.steps[i + 1].run_with_next:
                    raise ValueError(
                        "run_with_next can only pair two adjacent steps at a time. Keep "
                        "run_with_next=true on just one of these two steps and set the "
                        "other one back to false."
                    )
        return self
 
 
FEW_SHOT_EXAMPLES = [
    (
        "act proud",
        {
            "title": "Proud",
            "steps": [
                {"action": "speak", "speech_text": "I did it! I'm so proud of that.", "duration_seconds": 1.5},
                {"action": "neck", "neck_pose": "look_center", "duration_seconds": 1.0},
                {"action": "arm", "arm_side": "right", "arm_pose": "celebrate", "duration_seconds": 1.5},
                {"action": "move", "move_action": "move_forward", "move_distance_meters": 0.3},
            ],
        },
    ),
    (
        "act sad",
        {
            "title": "Sad",
            "steps": [
                {"action": "neck", "neck_pose": "look_left", "duration_seconds": 1.0},
                {"action": "arm", "arm_side": "left", "arm_pose": "droop", "duration_seconds": 2.0},
                {"action": "speak", "speech_text": "I'm feeling a little sad today.", "duration_seconds": 1.5},
            ],
        },
    ),
    (
        "greet someone shyly",
        {
            "title": "Shy Greeting",
            "steps": [
                {"action": "neck", "neck_pose": "look_right", "duration_seconds": 1.0},
                {"action": "arm", "arm_side": "left", "arm_pose": "hug_self", "duration_seconds": 1.5},
                {"action": "speak", "speech_text": "Oh... um, hi there.", "duration_seconds": 1.5},
            ],
        },
    ),
    (
        "move forward",
        {
            "title": "Move Forward",
            "steps": [
                {"action": "speak", "speech_text": "Moving forward now.", "duration_seconds": 1.0},
                {"action": "move", "move_action": "move_forward", "move_distance_meters": 0.5},
            ],
        },
    ),
    (
        "back away nervously",
        {
            "title": "Back Away",
            "steps": [
                {"action": "neck", "neck_pose": "look_left", "duration_seconds": 1.0},
                {"action": "move", "move_action": "move_backward", "move_distance_meters": 0.4},
                {"action": "speak", "speech_text": "Sorry, I need a little space.", "duration_seconds": 1.5},
            ],
        },
    ),
    (
        "turn around and walk away",
        {
            "title": "Turn And Go",
            "steps": [
                {"action": "speak", "speech_text": "Okay, I'll head off now.", "duration_seconds": 1.5},
                {"action": "move", "move_action": "turn_right", "turn_degrees": 180},
                {"action": "move", "move_action": "move_forward", "move_distance_meters": 0.5},
            ],
        },
    ),
    (
        "wave hello and say hi at the same time",
        {
            "title": "Wave And Greet",
            "steps": [
                {"action": "speak", "speech_text": "Hi there, so good to see you!", "run_with_next": True, "duration_seconds": 1.5},
                {"action": "arm", "arm_side": "right", "arm_pose": "wave", "duration_seconds": 1.5},
                {"action": "neck", "neck_pose": "look_center", "duration_seconds": 1.0},
            ],
        },
    ),
    (
        "celebrate and say congratulations at the same time",
        {
            "title": "Celebrate And Congratulate",
            "steps": [
                {"action": "arm", "arm_side": "right", "arm_pose": "celebrate", "run_with_next": True, "duration_seconds": 1.5},
                {"action": "speak", "speech_text": "Wow, congratulations! That's amazing news!"},
                {"action": "neck", "neck_pose": "look_center", "duration_seconds": 1.0},
            ],
        },
    ),
]
 
for _instruction, _output in FEW_SHOT_EXAMPLES:
    Choreography.model_validate(_output)
 
_FEW_SHOT_INSTRUCTIONS = {instr for instr, _ in FEW_SHOT_EXAMPLES}
_FEW_SHOT_OUTPUTS = [out for _, out in FEW_SHOT_EXAMPLES]
 
 
def ensure_ollama_installed():
    if shutil.which("ollama") is None:
        print("⚠️ Install Ollama once from https://ollama.com/download, then re-run.")
        sys.exit(1)
 
 
def ensure_ollama_running():
    try:
        ollama.list()
        return
    except Exception:
        pass
    print("Starting Ollama server...")
    subprocess.Popen(["ollama", "serve"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    start = time.time()
    while time.time() - start < SERVER_START_TIMEOUT:
        try:
            ollama.list()
            print("✅ Ollama server is up.")
            return
        except Exception:
            time.sleep(SERVER_POLL_INTERVAL)
 
 
def ensure_model_pulled(model_name):
    try:
        ollama.show(model_name)
        return
    except Exception:
        pass
    print(f"Downloading model '{model_name}'...")
    for _ in ollama.pull(model_name, stream=True):
        pass
    print(f"✅ Model '{model_name}' ready.")
 
 
def warm_up_model(model_name):
    print("Warming up model...")
    ollama.chat(
        model=model_name,
        messages=[{"role": "user", "content": "hello"}],
        options=CHAT_OPTIONS,
        keep_alive=CHAT_KEEP_ALIVE,
        think=False,
    )
    print("✅ Model warm.")
 
 
def ask_choreographer(instruction, max_attempts=MAX_VALIDATION_ATTEMPTS):
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for example_instruction, example_output in FEW_SHOT_EXAMPLES:
        messages.append({"role": "user", "content": example_instruction})
        messages.append({"role": "assistant", "content": json.dumps(example_output)})
    messages.append({"role": "user", "content": instruction})
 
    for attempt in range(1, max_attempts + 1):
        print(f"  🧠 thinking (attempt {attempt}/{max_attempts}): ", end="", flush=True)
        t0 = time.time()
        raw_content = ""
        for chunk in ollama.chat(
            model=OLLAMA_MODEL,
            format=Choreography.model_json_schema(),
            messages=messages,
            options=CHAT_OPTIONS,
            keep_alive=CHAT_KEEP_ALIVE,
            think=False,
            stream=True,
        ):
            piece = chunk["message"]["content"]
            raw_content += piece
            print(piece, end="", flush=True)
        print(f"\n  ⏱  inference took {time.time() - t0:.2f}s")
 
        try:
            choreography = Choreography.model_validate_json(raw_content)
        except ValidationError as e:
            print(f"⚠️ Attempt {attempt}/{max_attempts} didn't pass validation:")
            print(e)
            feedback = str(e)
        else:
            if choreography.model_dump() in _FEW_SHOT_OUTPUTS and instruction not in _FEW_SHOT_INSTRUCTIONS:
                print(f"⚠️ Attempt {attempt}/{max_attempts} returned a few-shot example verbatim, not a new plan.")
                feedback = (
                    "That response was one of the earlier example plans copied verbatim, "
                    "not a new plan for THIS instruction. Write a plan specific to: "
                    f"'{instruction}'."
                )
            else:
                return choreography
 
        if attempt == max_attempts:
            print("⚠️ Out of attempts — nothing will be performed.")
            return None
 
        print("↻ Sending the issue back to the model to retry...\n")
        messages.append({"role": "assistant", "content": raw_content})
        messages.append({
            "role": "user",
            "content": (
                f"That plan didn't work:\n{feedback}\n"
                "Fix ONLY what's wrong and send a complete, corrected "
                "Choreography that satisfies the schema."
            ),
        })
 
    return None
 
 
def main():
    ensure_ollama_installed()
    ensure_ollama_running()
    ensure_model_pulled(OLLAMA_MODEL)
    warm_up_model(OLLAMA_MODEL)
 
    print("✅ Ready. Type an instruction, or 'quit' to stop.")
    print('   Example: "act shy and say hello"\n')
 
    while True:
        instruction = input("Instruction> ").strip()
        if not instruction:
            continue
        if instruction.lower() in {"quit", "exit"}:
            break
 
        choreography = ask_choreographer(instruction)
        if choreography is None:
            continue
 
        print(f"\n📋 {choreography.title}")
        print(choreography.model_dump_json(indent=2))
        print()
 
 
if __name__ == "__main__":
    main()
    

✅ Ready. Type an instruction, or 'quit' to stop.
   Example: "act shy and say hello"

Instruction> act shy and say hello

📋 Shy Hello
{
  "title": "Shy Hello",
  "steps": [
    { "action": "neck", "speech_text": null, "neck_pose": "look_right", "arm_side": null, "arm_pose": null, "move_action": null, "duration_seconds": 1.0 },
    { "action": "arm", "speech_text": null, "neck_pose": null, "arm_side": "left", "arm_pose": "hug_self", "move_action": null, "duration_seconds": 1.5 },
    { "action": "speak", "speech_text": "Oh... um, hello there.", "neck_pose": null, "arm_side": null, "arm_pose": null, "move_action": null, "duration_seconds": 1.5 }
  ]
}

🎬 Performing: Shy Hello

  🤖 bot.look_right()
  🤖 bot.move_left_arm(shoulder=30.0, elbow=50.0)  # hug_self
  🗣️  BonicBot says: "Oh... um, hello there."

✅ Performance complete.

try:
    return Choreography.model_validate_json(raw_content)
except ValidationError as e:
    ...