Lesson 13: Giving BonicBot Different Personalities

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

import re
import shutil
import subprocess
import sys
import threading
import time
import difflib
from collections import Counter
from pathlib import Path
 
import numpy as np
import ollama
import pyaudio
import speech_recognition as sr
from faster_whisper import WhisperModel
from piper import PiperVoice, SynthesisConfig
from piper.download_voices import download_voice
 
 
from bonicbot_bridge import BonicBot
 
HOST = "[IP_ADDRESS]"
 
OLLAMA_MODEL = "qwen3.5:0.8b"
DEFAULT_VOICE = "en_US-lessac-medium"
PIPER_VOICES_DIR = Path(__file__).resolve().parent / "piper_voices"
PIPER_LENGTH_SCALE = 0.8
PIPER_VOLUME = 1.0
 
WHISPER_MODEL_SIZE = "small.en"
WHISPER_BEAM_SIZE = 1
 
SERVER_START_TIMEOUT = 15.0
SERVER_POLL_INTERVAL = 0.5
 
LISTEN_TIMEOUT = 15.0
PHRASE_TIME_LIMIT = 25.0
PAUSE_THRESHOLD = 0.6
NON_SPEAKING_DURATION = 0.4
ENERGY_THRESHOLD_CEILING = 4000
MAX_HISTORY_MESSAGES = 9
 
REPEAT_SIMILARITY_THRESHOLD = 0.85
SHORT_UTTERANCE_WORD_LIMIT = 6
EXIT_WORDS = {"goodbye", "bye", "stop", "exit", "quit"}
DEFAULT_PERSONA = "buddy"
 
# ----------------------------------------------------------------------
# GESTURE CONFIG
# ----------------------------------------------------------------------
GESTURES_ENABLED = True  # set False to run with no robot body attached
 
# Joint limits in degrees, straight from the SDK docs. Used to sanity-check
# any angle you tweak below so a typo can't try to drive a servo past what
# the hardware supports.
SHOULDER_LIMITS = (-45, 180)
ELBOW_LIMITS = (0, 50)
NECK_YAW_LIMITS = (-90, 90)
JOINT_LIMITS = {"shoulder": SHOULDER_LIMITS, "elbow": ELBOW_LIMITS}
 
# Rest angle for each joint type — used whenever a gesture releases back
# to a neutral pose. 0 is inside both the shoulder and elbow ranges above.
ARM_REST_ANGLES = {"shoulder": 0, "elbow": 0}
 
# ⚠️ ASSUMED joint_name strings for bot.servo.set_single_servo(joint_name, angle).
# The SDK docs confirm the method exists but don't spell out its exact
# joint_name values, and I couldn't reach github.com/api.github.com from
# this environment to check the source directly. Before relying on this,
# connect once and run `print(bot.servo.get_servo_limits())` (or
# get_servo_angles()) — its keys are almost certainly the real joint_name
# strings — and fix the values below to match if they differ.
SERVO_JOINT_NAMES = {
    ("left", "shoulder"): "left_shoulder",
    ("right", "shoulder"): "right_shoulder",
    ("left", "elbow"): "left_elbow",
    ("right", "elbow"): "right_elbow",
}
 
# Fallback gesture values used for any persona that doesn't override them.
# Each persona gestures with exactly ONE joint on ONE arm — never shoulder
# and elbow together, never both arms together.
DEFAULT_GESTURES = {
    "arm_side": "right",           # "left" or "right" — the one arm this persona uses
    "arm_joint": "shoulder",       # "shoulder" or "elbow" — the one joint this persona uses
    "arm_forward_angle": 60,       # angle for the "explaining" pose held while talking
    "neck_thinking_style": "bounce",   # "bounce" (fast, wide) or "tilt" (slow, small)
    "neck_thinking_pause": 0.6,        # seconds between each step of the neck animation
    "neck_thinking_amplitude": 35,     # degrees left/right of center while thinking
    "greeting_style": "arm_bob",   # "arm_bob" (single joint/arm) or "neck_sweep" (neck only)
}
 
THINKING_GESTURE_MIN_PAUSE = 0.3  # sanity floor so a bad persona config can't spin the neck servo
 
 
def clamp_angle(joint, angle):
    """Keep a persona-configured angle inside the hardware's real range for
    that joint type, so a tweak in PERSONAS can't send a bad value to the servo."""
    low, high = JOINT_LIMITS.get(joint, (-90, 180))
    return max(low, min(high, angle))
 
SAFETY_SUFFIX = (
    " Always answer the student's actual question directly and specifically "
    "in your very first sentence — never substitute a catchphrase, a "
    "reintroduction of yourself, or a repeat of an earlier reply for a real "
    "answer, and never reuse the same sentence twice in a row. No matter "
    "which persona you are playing, keep replies short — one to two "
    "sentences, meant to be spoken out loud. Never claim to be a real human "
    "or to have real feelings, never pretend the roleplay is literally true, "
    "and never give unsafe, harmful, illegal, or otherwise inappropriate "
    "advice. Stay in character, but stay safe and honest — if you don't know "
    "something, say so instead of guessing."
)
 
PERSONAS = {
    "buddy": {
        "display_name": "Best Friend",
        "aliases": ["buddy", "best friend", "bestie", "friend", "study buddy", "pal", "bro"],
        "base_prompt": (
            "You are BonicBot playing the user's chill, upbeat best friend and "
            "study buddy — casual, warm, and full of energy, like a friend "
            "hyping you up between classes. Use relaxed, casual language and "
            "light slang ('let's go', 'you got this', 'no worries'), keep it "
            "short and punchy, and always answer the actual question clearly "
            "and correctly before adding any hype."
        ),
        "greeting": (
            "Heyyy, it's your buddy BonicBot! Ready to figure stuff out "
            "together — what's up?"
        ),
        "voice": "en_US-amy-medium",
        # Energetic: gestures entirely with its RIGHT SHOULDER (never the
        # elbow, never the left arm) — a raised-shoulder "explaining" pose,
        # a couple of quick shoulder bobs as a hello, and a fast, wide
        # left-right neck bounce while thinking.
        "gestures": {
            "arm_side": "right",
            "arm_joint": "shoulder",
            "arm_forward_angle": 90,
            "neck_thinking_style": "bounce",
            "neck_thinking_pause": 0.2,
            "neck_thinking_amplitude": 20,
            "greeting_style": "arm_bob",
        },
    },
    "teacher": {
        "display_name": "Strict Professor",
        "aliases": ["teacher", "professor", "strict teacher", "prof", "sir", "principal"],
        "base_prompt": (
            "You are BonicBot playing a strict, no-nonsense professor — "
            "formal, precise, and impatient with vague answers. Speak "
            "crisply and seriously, address the student as 'student', and "
            "treat every question like it might appear on an exam, but "
            "always give the actual correct answer first, clearly, with zero "
            "fluff."
        ),
        "greeting": (
            "Attention. This is Professor BonicBot. State your question "
            "clearly, and I shall answer precisely."
        ),
        "voice": "en_US-lessac-medium",
        # Restrained: gestures entirely with its LEFT ELBOW (never the
        # shoulder, never the right arm) — a raised-forearm "explaining"
        # pose, like pointing at a chalkboard, plus a slow, small neck lean
        # while thinking. Greets with a neck-only sweep of the room — no
        # arm movement at all, to feel more formal than buddy's arm bob.
        "gestures": {
            "arm_side": "left",
            "arm_joint": "elbow",
            "arm_forward_angle": 40,
            "neck_thinking_style": "tilt",
            "neck_thinking_pause": 1.0,
            "neck_thinking_amplitude": 20,
            "greeting_style": "neck_sweep",
        },
    },
}
 
SWITCH_TRIGGERS = [
    "switch persona to", "switch to", "become a", "become the", "become",
    "be a ", "be the ", "turn into a", "turn into the", "turn into",
    "change persona to", "change to", "act like a", "act like the", "act like",
]
LIST_TRIGGERS = [
    "what personas", "which personas", "list personas", "who can you be",
    "what characters", "what other personas", "what personalities",
]
SWITCH_INTENT_WORDS = ("change", "become", "switch", "turn", "act", " be ")
 
EMOJI_PATTERN = re.compile(
    "[\U0001F300-\U0001FAFF\U00002600-\U000027BF\U0001F1E6-\U0001F1FF"
    "\U0001F900-\U0001F9FF\U00002190-\U000021FF\U00002B00-\U00002BFF]+",
    flags=re.UNICODE,
)
 
 
def clean_for_speech(text):
    text = EMOJI_PATTERN.sub("", text)
    text = re.sub(r"\*\*(.*?)\*\*", r"\1", text)
    text = re.sub(r"\*(.*?)\*", r"\1", text)
    text = re.sub(r"`(.*?)`", r"\1", text)
    text = re.sub(r"#+\s*", "", text)
    text = re.sub(r"[*_~]", "", text)
    return re.sub(r"\s+", " ", text).strip()
 
 
def ensure_ollama_installed():
    if shutil.which("ollama") is None:
        print("Install Ollama from https://ollama.com/download, then re-run this script.")
        sys.exit(1)
 
 
def ensure_ollama_running():
    try:
        ollama.list()
        return
    except Exception:
        pass
    subprocess.Popen(["ollama", "serve"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    start = time.time()
    while time.time() - start < SERVER_START_TIMEOUT:
        try:
            ollama.list()
            return
        except Exception:
            time.sleep(SERVER_POLL_INTERVAL)
 
 
def is_model_available(model_name):
    try:
        ollama.show(model_name)
        return True
    except Exception:
        return False
 
 
def ensure_model_pulled(model_name):
    if is_model_available(model_name):
        return
    print(f"Downloading model '{model_name}'...")
    last_status = None
    for progress in ollama.pull(model_name, stream=True):
        status = progress.get("status", "")
        if status and status != last_status:
            print(f"  {status}")
            last_status = status
 
 
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}'...")
    try:
        download_voice(voice_name, voices_dir)
    except Exception as e:
        print(f"Couldn't download voice: {e}")
        sys.exit(1)
    return model_path, config_path
 
 
def load_persona_voice(persona_key, voice_cache):
    voice_name = PERSONAS[persona_key].get("voice", DEFAULT_VOICE)
    if voice_name not in voice_cache:
        model_path, config_path = ensure_piper_voice_ready(voice_name, PIPER_VOICES_DIR)
        voice_cache[voice_name] = PiperVoice.load(model_path, config_path=config_path)
    return voice_cache[voice_name]
 
 
def speak(piper_voice, pyaudio_instance, text):
    print(f"BonicBot: {text}")
    speech_text = clean_for_speech(text)
    syn_config = SynthesisConfig(length_scale=PIPER_LENGTH_SCALE, volume=PIPER_VOLUME)
    stream = None
    try:
        for chunk in piper_voice.synthesize(speech_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()
 
 
NO_SPEECH_PROB_THRESHOLD = 0.82
AVG_LOGPROB_THRESHOLD = -1.5
HALLUCINATION_MAX_REPEAT = 3
HALLUCINATION_MIN_CHARS = 20
MIN_SPEECH_DURATION = 0.3
 
 
def _looks_like_hallucination(text):
    if len(text) < HALLUCINATION_MIN_CHARS:
        return False
    sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+", text) if s.strip()]
    if len(sentences) < 4:
        return False
    counts = Counter(sentences)
    if counts.most_common(1)[0][1] >= HALLUCINATION_MAX_REPEAT:
        return True
    return len(counts) / len(sentences) < 0.5
 
 
def listen(recognizer, mic, whisper_model):
    with mic as source:
        print("\n🎤 Listening... (say 'goodbye' to quit, or 'who can you be?')")
        try:
            audio = recognizer.listen(source, timeout=LISTEN_TIMEOUT, phrase_time_limit=PHRASE_TIME_LIMIT)
        except sr.WaitTimeoutError:
            return None
 
    audio_np = np.frombuffer(audio.get_raw_data(), dtype=np.int16).astype(np.float32) / 32768.0
    segments, _ = whisper_model.transcribe(
        audio_np, language="en", beam_size=WHISPER_BEAM_SIZE, vad_filter=True,
        vad_parameters={"min_silence_duration_ms": 500, "threshold": 0.35, "speech_pad_ms": 300},
    )
 
    kept = []
    speech_duration = 0.0
    for seg in segments:
        if seg.no_speech_prob is not None and seg.no_speech_prob > NO_SPEECH_PROB_THRESHOLD:
            continue
        if seg.avg_logprob is not None and seg.avg_logprob < AVG_LOGPROB_THRESHOLD:
            continue
        kept.append(seg.text)
        speech_duration += seg.end - seg.start
 
    text = " ".join(kept).strip()
    if not text or speech_duration < MIN_SPEECH_DURATION or _looks_like_hallucination(text):
        return None
 
    print(f"You said: {text}")
    return text
 
 
LLM_OPTIONS = {
    "temperature": 0.6,
    "top_p": 0.9,
    "top_k": 40,
    "repeat_penalty": 1.4,
    "repeat_last_n": 48,
    "num_predict": 100,
}
 
 
def ask_llm(history):
    try:
        response = ollama.chat(model=OLLAMA_MODEL, messages=history, think=False, options=LLM_OPTIONS)
    except ollama.ResponseError as e:
        if e.status_code == 404:
            ensure_model_pulled(OLLAMA_MODEL)
            return ask_llm(history)
        raise
    return response["message"]["content"].strip()
 
 
def ask_llm_in_background(history, result):
    """Run ask_llm() on a worker thread and stash the outcome in `result`
    (a plain dict) so the main thread can pick it up once ready — this is
    what lets the thinking gesture play concurrently with the actual LLM
    call instead of after it."""
    try:
        result["reply"] = ask_llm(history)
    except Exception as e:
        result["error"] = e
 
 
def trim_history(history):
    if len(history) > MAX_HISTORY_MESSAGES:
        history = [history[0]] + history[-(MAX_HISTORY_MESSAGES - 1):]
    return history
 
 
def is_repeat_question(current_text, previous_text):
    if not previous_text:
        return False
    a, b = current_text.strip().lower(), previous_text.strip().lower()
    return difflib.SequenceMatcher(None, a, b).ratio() >= REPEAT_SIMILARITY_THRESHOLD
 
 
def build_system_message(persona_key):
    persona = PERSONAS[persona_key]
    return {"role": "system", "content": persona["base_prompt"] + SAFETY_SUFFIX}
 
 
def get_persona_gestures(persona_key):
    """Merge a persona's gesture overrides on top of DEFAULT_GESTURES, so
    any persona that's added later without a "gestures" block still gets
    sane, working values instead of a KeyError. Also clamps the configured
    angles to the joint's real hardware range."""
    gestures = dict(DEFAULT_GESTURES)
    gestures.update(PERSONAS[persona_key].get("gestures", {}))
    gestures["neck_thinking_pause"] = max(gestures["neck_thinking_pause"], THINKING_GESTURE_MIN_PAUSE)
    gestures["neck_thinking_amplitude"] = max(0, min(NECK_YAW_LIMITS[1], gestures["neck_thinking_amplitude"]))
    gestures["arm_forward_angle"] = clamp_angle(gestures["arm_joint"], gestures["arm_forward_angle"])
    return gestures
 
 
def find_persona_by_fragment(fragment):
    fragment = fragment.strip(" .!?,")
    if not fragment:
        return None
    for key, persona in PERSONAS.items():
        for alias in [key] + persona["aliases"]:
            if alias in fragment or fragment in alias:
                return key
    return None
 
 
def try_match_switch_command(text_lower):
    for trigger in SWITCH_TRIGGERS:
        idx = text_lower.find(trigger)
        if idx != -1:
            match = find_persona_by_fragment(text_lower[idx + len(trigger):])
            if match:
                return match
    words = text_lower.split()
    if any(word in text_lower for word in SWITCH_INTENT_WORDS) or len(words) <= SHORT_UTTERANCE_WORD_LIMIT:
        for key, persona in PERSONAS.items():
            for alias in [key] + persona["aliases"]:
                if alias in text_lower:
                    return key
    return None
 
 
def is_list_command(text_lower):
    return any(trigger in text_lower for trigger in LIST_TRIGGERS)
 
 
def speak_persona_list(piper_voice, pyaudio_instance):
    names = ", ".join(p["display_name"] for p in PERSONAS.values())
    speak(piper_voice, pyaudio_instance, f"I can be: {names}. Just say something like 'become a pirate'.")
 
 
# ============================================================================
# ROBOT GESTURES
# ----------------------------------------------------------------------
# Hard rule enforced everywhere below: every call touches exactly one
# joint (shoulder OR elbow, never both) on exactly one arm (never left
# and right together). Neck movement is always single-joint by nature.
# ============================================================================
def set_single_joint(bot, side, joint, angle):
    """Move exactly one joint on exactly one arm via the SDK's
    bot.servo.set_single_servo(joint_name, angle) — the only call in the
    SDK that can move a shoulder or an elbow in isolation. Never use
    bot.move_left_arm()/move_right_arm() for gestures — those move two
    joints (shoulder + elbow) in a single command."""
    if not GESTURES_ENABLED or bot is None:
        return
    joint_name = SERVO_JOINT_NAMES.get((side, joint))
    if joint_name is None:
        print(f"⚠️ No servo joint_name mapped for ({side}, {joint}) — skipping.")
        return
    try:
        bot.servo.set_single_servo(joint_name, clamp_angle(joint, angle))
    except Exception as e:
        print(f"⚠️ Couldn't move {joint_name}: {e}")
 
 
def reset_persona_arm(bot, gestures):
    """Return the persona's one gesturing joint to rest. Only ever touches
    that single joint/arm — the other arm and the other joint on this arm
    are never commanded, so they simply stay wherever they already are."""
    rest_angle = ARM_REST_ANGLES.get(gestures["arm_joint"], 0)
    set_single_joint(bot, gestures["arm_side"], gestures["arm_joint"], rest_angle)
 
 
def center_head(bot):
    """Center the neck — done right before BonicBot starts speaking."""
    if not GESTURES_ENABLED or bot is None:
        return
    try:
        bot.look_center()
    except Exception as e:
        print(f"⚠️ Couldn't center head: {e}")
 
 
def push_arm_forward(bot, gestures):
    """Move the persona's single designated joint into its 'explaining'
    pose and hold it there while BonicBot talks — e.g. buddy raises its
    right shoulder, teacher bends its left elbow. Called synchronously,
    blocking, right before speak() starts."""
    set_single_joint(bot, gestures["arm_side"], gestures["arm_joint"], gestures["arm_forward_angle"])
 
 
def thinking_gesture_loop(bot, stop_event, style, pause, amplitude):
    """Play a persona-specific neck animation until stop_event is set —
    plays while the LLM generates in the background. Neck-only, via
    bot.set_neck(yaw), so pace and amplitude are tunable per persona.
    Always leaves the neck centered on exit, even on error.
 
    style="bounce": a full swing between -amplitude and +amplitude
    (energetic personas). style="tilt": a smaller lean to one side and
    back to center, not a full swing (deliberate/serious personas) —
    same neck API, calmer motion."""
    if not GESTURES_ENABLED or bot is None:
        return
    try:
        while not stop_event.is_set():
            if style == "bounce":
                bot.set_neck(-amplitude)
                if stop_event.wait(pause):
                    break
                bot.set_neck(amplitude)
                if stop_event.wait(pause):
                    break
            else:  # "tilt" or any unrecognized style falls back to this calmer version
                bot.set_neck(-amplitude)
                if stop_event.wait(pause):
                    break
                bot.set_neck(0)
                if stop_event.wait(pause):
                    break
    except Exception as e:
        print(f"⚠️ Thinking gesture skipped: {e}")
    finally:
        try:
            bot.look_center()
        except Exception as e:
            print(f"⚠️ Couldn't re-center neck: {e}")
 
 
def play_greeting_gesture(bot, gestures):
    """One-off gesture played right before a persona's greeting line —
    this is what makes switching persona feel like meeting someone new,
    not just hearing a different voice.
 
    style="arm_bob": two quick moves of the persona's single joint up and
    back to rest — same one joint, same one arm as its talking pose, never
    the other joint or the other arm.
    style="neck_sweep": one slow look left, then right, then back to
    center — neck only, no arm movement at all (more restrained)."""
    if not GESTURES_ENABLED or bot is None:
        return
    style = gestures["greeting_style"]
    try:
        if style == "arm_bob":
            side, joint = gestures["arm_side"], gestures["arm_joint"]
            angle = gestures["arm_forward_angle"]
            rest_angle = ARM_REST_ANGLES.get(joint, 0)
            for _ in range(2):
                set_single_joint(bot, side, joint, angle)
                time.sleep(0.3)
                set_single_joint(bot, side, joint, rest_angle)
                time.sleep(0.3)
        elif style == "neck_sweep":
            bot.look_left()
            time.sleep(0.6)
            bot.look_right()
            time.sleep(0.6)
            bot.look_center()
    except Exception as e:
        print(f"⚠️ Greeting gesture skipped: {e}")
 
 
def main():
    ensure_ollama_installed()
    ensure_ollama_running()
    ensure_model_pulled(OLLAMA_MODEL)
 
    pyaudio_instance = pyaudio.PyAudio()
    voice_cache = {}
    current_persona = DEFAULT_PERSONA
    piper_voice = load_persona_voice(current_persona, voice_cache)
    history = [build_system_message(current_persona)]
    last_user_text = None
 
    bot = BonicBot(host= HOST) if GESTURES_ENABLED else None
    if bot is not None:
        reset_persona_arm(bot, get_persona_gestures(current_persona))  # known starting position
        center_head(bot)
 
    print("Loading Whisper model...")
    whisper_model = WhisperModel(WHISPER_MODEL_SIZE, device="cpu", compute_type="int8")
 
    recognizer = sr.Recognizer()
    recognizer.pause_threshold = PAUSE_THRESHOLD
    recognizer.non_speaking_duration = NON_SPEAKING_DURATION
    recognizer.dynamic_energy_threshold = True
    mic = sr.Microphone(sample_rate=16000)
 
    print("Calibrating microphone, please stay quiet...")
    with mic as source:
        recognizer.adjust_for_ambient_noise(source, duration=1.0)
    if recognizer.energy_threshold > ENERGY_THRESHOLD_CEILING:
        recognizer.energy_threshold = ENERGY_THRESHOLD_CEILING
 
    play_greeting_gesture(bot, get_persona_gestures(current_persona))
    speak(piper_voice, pyaudio_instance, PERSONAS[current_persona]["greeting"])
 
    try:
        while True:
            user_text = listen(recognizer, mic, whisper_model)
            if user_text is None:
                speak(piper_voice, pyaudio_instance, "Sorry, I didn't catch that. Could you say it again?")
                continue
 
            text_lower = re.sub(r"[^\w\s]", "", user_text.strip().lower())
 
            if text_lower in EXIT_WORDS:
                speak(piper_voice, pyaudio_instance, "Goodbye! It was nice talking with you.")
                break
 
            if is_list_command(text_lower):
                speak_persona_list(piper_voice, pyaudio_instance)
                continue
 
            matched_persona = try_match_switch_command(text_lower)
            if matched_persona:
                if matched_persona == current_persona:
                    speak(piper_voice, pyaudio_instance,
                          f"I'm already your {PERSONAS[current_persona]['display_name']}!")
                else:
                    reset_persona_arm(bot, get_persona_gestures(current_persona))  # release the OLD joint
                    current_persona = matched_persona
                    piper_voice = load_persona_voice(current_persona, voice_cache)
                    history = [build_system_message(current_persona)]
                    last_user_text = None
                    play_greeting_gesture(bot, get_persona_gestures(current_persona))
                    speak(piper_voice, pyaudio_instance, PERSONAS[current_persona]["greeting"])
                continue
 
            if is_repeat_question(user_text, last_user_text):
                turn_content = (
                    f"{user_text}\n\n[The user just asked this again — your previous answer was "
                    "likely wrong, unclear, or not what they needed. Do not repeat it. If you're "
                    "not confident of the correct answer, say so honestly instead of guessing "
                    "again, while staying in character.]"
                )
            else:
                turn_content = user_text
 
            history.append({"role": "user", "content": turn_content})
            last_user_text = user_text
 
            gestures = get_persona_gestures(current_persona)
 
            # Play the persona's thinking gesture and run the LLM call at
            # the same time, so the wait isn't dead air on the robot side.
            stop_thinking = threading.Event()
            gesture_thread = threading.Thread(
                target=thinking_gesture_loop,
                args=(
                    bot, stop_thinking,
                    gestures["neck_thinking_style"],
                    gestures["neck_thinking_pause"],
                    gestures["neck_thinking_amplitude"],
                ),
                daemon=True,
            )
            gesture_thread.start()
 
            llm_result = {}
            llm_thread = threading.Thread(
                target=ask_llm_in_background, args=(history, llm_result), daemon=True
            )
            llm_thread.start()
 
            llm_thread.join()
            stop_thinking.set()
            gesture_thread.join()
 
            if "error" in llm_result:
                raise llm_result["error"]
            reply = llm_result["reply"]
 
            history.append({"role": "assistant", "content": reply})
            history = trim_history(history)
 
            center_head(bot)
            push_arm_forward(bot, gestures)
 
            speak(piper_voice, pyaudio_instance, reply)
 
            reset_persona_arm(bot, gestures)
 
    except KeyboardInterrupt:
        pass
    finally:
        pyaudio_instance.terminate()
 
 
if __name__ == "__main__":
    main()

Loading Whisper model...
Calibrating microphone, please stay quiet...
BonicBot: Heyyy, it's your buddy BonicBot! Ready to figure stuff out together — what's up?

🎤 Listening... (say 'goodbye' to quit, or 'who can you be?')
You said: what is a neural network
BonicBot: A neural network's basically a stack of math that learns patterns from examples — no worries, you'll get it!

🎤 Listening... (say 'goodbye' to quit, or 'who can you be?')
You said: switch to teacher
Downloading voice 'en_US-lessac-medium'...
BonicBot: Attention. This is Professor BonicBot. State your question clearly, and I shall answer precisely.

🎤 Listening... (say 'goodbye' to quit, or 'who can you be?')
You said: goodbye
BonicBot: Goodbye! It was nice talking with you.

def build_system_message(persona_key):
    persona = PERSONAS[persona_key]
    return {"role": "system", "content": persona["base_prompt"] + SAFETY_SUFFIX}

def find_persona_by_fragment(fragment):
    for key, persona in PERSONAS.items():
        for alias in [key] + persona["aliases"]:
            if alias in fragment or fragment in alias:
                return key
    return None

"wizard": {
    "display_name": "Wise Old Wizard",
    "aliases": ["wizard", "sorcerer", "old wizard", "mage"],
    "base_prompt": (
        "You are BonicBot playing a wise old wizard who speaks in gentle "
        "riddles and metaphors about magic, but always reveals the true, "
        "accurate answer by the end of your reply."
    ),
    "greeting": (
        "Ahh, a seeker of knowledge approaches. I am BonicBot, wizard of "
        "circuits and spells alike. Speak, and I shall divine an answer."
    ),
},