Lesson 16: BonicBot Listens, Researches, and Talks Back

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

import difflib
import inspect
import json
import queue
import re
import shutil
import subprocess
import sys
import threading
import time
from pathlib import Path
from urllib.parse import quote
 
import numpy as np
import ollama
import pyaudio
import requests
import speech_recognition as sr
from faster_whisper import WhisperModel
from piper import PiperVoice, SynthesisConfig
from piper.download_voices import download_voice
 
 
# ============================================================================
# CONFIG
# ============================================================================
OLLAMA_MODEL = "qwen3:0.6b"   # supports native tool-calling + a think toggle, still light enough for CPU
 
PIPER_VOICE_NAME = "en_US-lessac-medium"
PIPER_VOICES_DIR = Path(__file__).resolve().parent / "piper_voices"
PIPER_LENGTH_SCALE = 0.8   # > 1 = slower speech, < 1 = faster (1.0 = voice's own normal pace)
PIPER_VOLUME = 1.0         # > 1 = louder, < 1 = quieter
 
WHISPER_MODEL_SIZE = "small.en"   # tiny.en (fastest) / base.en (good balance) / small.en (most accurate, slower)
 
SERVER_START_TIMEOUT = 15.0
SERVER_POLL_INTERVAL = 0.5
 
LISTEN_TIMEOUT = 60.0       # seconds to wait for the person to start speaking
PHRASE_TIME_LIMIT = 15.0    # max seconds for a single thing they say
MAX_HISTORY_MESSAGES = 13   # system message + ~6 back-and-forth exchanges
 
REPEAT_SIMILARITY_THRESHOLD = 0.85
 
LOOKUP_TIMEOUT = 6.0
 
NO_SPEECH_PROB_THRESHOLD = 0.6   # above this, the segment is probably not speech
AVG_LOGPROB_THRESHOLD = -1.0     # below this, the model itself was unsure
 
EXIT_WORDS = {"goodbye", "bye", "stop", "exit", "quit"}
 
# Concise, punchy intro message
INTRO_MESSAGE = "Hi, I'm BonicBot! Ask me about any person, place, or event, and I'll look it up. Say goodbye to quit."
 
SYSTEM_PROMPT = (
    "You are BonicBot, a friendly, patient voice assistant helping students "
    "learn about NLP and AI. Keep answers short -- one to three sentences, "
    "spoken out loud. Use simple words. "
    "For every turn, think it through and choose exactly ONE of these three "
    "actions:\n"
    "1. ANSWER DIRECTLY -- only for greetings, small talk, opinions, simple "
    "arithmetic, or questions about yourself. Never answer a real-world "
    "factual question this way -- your own memory of people, places, "
    "dates, organizations, and events is not reliable.\n"
    "2. USE THE lookup_topic TOOL -- for any question about a specific "
    "real-world person, place, date, organization, event, or thing, even "
    "if you think you already know the answer. Call it BEFORE answering, "
    "not after.\n"
    "3. ASK ONE SHORT CLARIFYING QUESTION -- if the input is incomplete, "
    "cut off mid-sentence, ambiguous, or you genuinely can't tell what's "
    "being asked. Ask specifically about what's missing (for example: "
    "'Which country's prime minister did you mean?'). Do not guess, and "
    "do not call the tool with a vague or incomplete query -- ask first, "
    "then use the tool once you know the real topic.\n"
    "Never respond with a dismissive or quiz-style line like 'None of the "
    "above' or a flat 'I don't know' -- one of the three actions above "
    "always applies. If the tool returns nothing useful, say so honestly "
    "and ask a clarifying question about what they meant, instead of "
    "guessing.\n"
    "Your input comes from speech-to-text and is sometimes messy -- filler "
    "words, or a stray unrelated fragment mixed in with the real question "
    "from background noise. Find the one clear question or statement "
    "inside it and act on that, ignoring unrelated fragments; if nothing "
    "clear survives that filtering, that's when you ask a clarifying "
    "question. When a tool result names a specific office or role (e.g. "
    "President vs. Prime Minister), use the tool's wording exactly -- "
    "don't swap it for a different title from memory. Never pretend to be "
    "a human, and never give unsafe, harmful, or inappropriate advice."
)
 
 
EMOJI_PATTERN = re.compile(
    "["
    "\U0001F300-\U0001FAFF"
    "\U00002600-\U000027BF"
    "\U0001F1E6-\U0001F1FF"
    "\U0001F900-\U0001F9FF"
    "\U00002190-\U000021FF"
    "\U00002B00-\U00002BFF"
    "]+",
    flags=re.UNICODE,
)
 
 
def clean_for_speech(text):
    """Strips markdown formatting and emoji before handing text to Piper —
    the model doesn't know it's writing for a voice, so it still reaches
    for **bold**, `code`, and emoji out of habit."""
    # Strip <think> blocks so BonicBot doesn't speak its internal monologue out loud
    text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL | re.IGNORECASE)
 
    text = EMOJI_PATTERN.sub("", text)
    text = re.sub(r"\*\*(.*?)\*\*", r"\1", text)   # **bold**
    text = re.sub(r"\*(.*?)\*", r"\1", text)       # *italic*
    text = re.sub(r"`(.*?)`", r"\1", text)         # `code`
    text = re.sub(r"#+\s*", "", text)              # headers
    text = re.sub(r"[*_~]", "", text)              # stray leftover symbols
    text = re.sub(r"\s+", " ", text).strip()
    return text
 
 
# ============================================================================
# EXTERNAL LIBRARY — encyclopedia lookup, not part of the BonicBot Bridge SDK
# ============================================================================
def lookup_topic(query):
    """Resolves `query` to the best-matching Wikipedia page via the search
    API, then fetches its summary. Resolving via search instead of treating
    `query` as an exact title matters because the model is tiny and often
    passes slightly-off or multi-word phrasings -- an exact-title lookup
    silently fails on those and feeds 'No information was found' back to
    the model for no good reason."""
    query = query.strip()
    if not query:
        return None
 
    try:
        search_resp = requests.get(
            "https://en.wikipedia.org/w/rest.php/v1/search/page",
            params={"q": query, "limit": 1},
            timeout=LOOKUP_TIMEOUT,
            headers={"User-Agent": "BonicBot-Lesson16"},
        )
        if search_resp.status_code != 200:
            return None
        results = search_resp.json().get("pages", [])
        if not results:
            return None
        title = results[0]["key"]
    except (requests.RequestException, KeyError, ValueError, IndexError):
        return None
 
    try:
        summary_resp = requests.get(
            f"https://en.wikipedia.org/api/rest_v1/page/summary/{quote(title)}",
            timeout=LOOKUP_TIMEOUT,
            headers={"User-Agent": "BonicBot-Lesson16"},
        )
        if summary_resp.status_code != 200:
            return None
        return summary_resp.json().get("extract")
    except requests.RequestException:
        return None
 
 
# The tool description Ollama uses to decide WHEN to call lookup_topic.
LOOKUP_TOOL = {
    "type": "function",
    "function": {
        "name": "lookup_topic",
        "description": (
            "Look up a short, accurate factual summary of a topic. Use "
            "this by default for any specific person, place, date, "
            "organization, event, or thing -- don't rely on your own "
            "memory for facts like these."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The topic to look up, e.g. 'Mount Everest'",
                }
            },
            "required": ["query"],
        },
    },
}
 
# Last-resort cleanup only: if the model still produces a flat dismissive
# line despite the system prompt telling it not to, rewrite it into an
# actual clarifying question instead of leaving a dead end. This does NOT
# force a tool call -- it's just turning a bad phrasing into a good one.
PUNT_PATTERN = re.compile(
    r"none of the above|not (?:applicable|available)|no (?:correct )?answer",
    re.IGNORECASE,
)
 
 
# ============================================================================
# OLLAMA SETUP — all handled from Python, no manual `ollama` commands
# ============================================================================
def ensure_ollama_installed():
    if shutil.which("ollama") is None:
        print("⚠️ The Ollama app isn't installed on this computer yet.")
        print("This is a one-time setup step, done once per computer:")
        print("  1. Go to https://ollama.com/download")
        print("  2. Download and run the installer for your OS")
        print("  3. Re-run this script afterwards")
        sys.exit(1)
 
 
def ensure_ollama_running():
    try:
        ollama.list()
        return
    except Exception:
        pass
    print("Ollama server isn't running yet — starting it now...")
    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)
    print("⚠️ Couldn't confirm the server started in time — continuing anyway.")
 
 
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):
        print(f"✅ Model '{model_name}' is already downloaded.")
        return
    print(f"Model '{model_name}' not found locally — downloading now "
          f"(one-time, a few hundred MB)...")
    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
    print(f"✅ Model '{model_name}' ready.")
 
 
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():
        print(f"✅ Voice '{voice_name}' is already downloaded.")
        return model_path, config_path
    print(f"Voice '{voice_name}' not found locally — downloading now "
          f"(one-time, roughly 60MB)...")
    try:
        download_voice(voice_name, voices_dir)
    except Exception as e:
        print(f"⚠️ Couldn't download the Piper voice: {e}")
        print("Check your internet connection and the voice name, then try again.")
        sys.exit(1)
    print(f"✅ Voice '{voice_name}' ready.")
    return model_path, config_path
 
 
# ============================================================================
# BACKGROUND STREAMING TEXT-TO-SPEECH (Sentence Queue & Interrupts)
# ============================================================================
tts_queue = queue.Queue()
interrupt_event = threading.Event()
 
def tts_worker(piper_voice, pyaudio_instance):
    """Monitors the tts_queue and plays synthesized audio sentence-by-sentence.
    Allows for immediate audio interruption if interrupt_event is set."""
    while True:
        item = tts_queue.get()
        if item is None:
            break
        sentence, syn_config = item
        
        interrupt_event.clear()
        stream = None
        try:
            for chunk in piper_voice.synthesize(sentence, syn_config=syn_config):
                if interrupt_event.is_set():
                    break
                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)
        except Exception:
            pass
        finally:
            if stream is not None:
                try:
                    stream.stop_stream()
                    stream.close()
                except Exception:
                    pass
        tts_queue.task_done()
 
 
def stop_tts():
    """Instantly stops any ongoing synthesis and purges the speech queue."""
    interrupt_event.set()
    while not tts_queue.empty():
        try:
            tts_queue.get_nowait()
            tts_queue.task_done()
        except queue.Empty:
            break
 
 
def wait_for_tts():
    """Blocks the main thread loop until all queued sentences have finished 
    being spoken out loud by the background TTS worker thread."""
    tts_queue.join()
 
 
def speak_async(text, syn_config):
    """Splits a full block of text into sentences and queues them asynchronously."""
    sentences = re.split(r'(?<=[.!?])\s+', text)
    for s in sentences:
        s = s.strip()
        if s:
            cleaned = clean_for_speech(s)
            if cleaned:
                tts_queue.put((cleaned, syn_config))
 
 
def print_and_speak(text, syn_config):
    """Prints and queues hardcoded platform messages (like intro and goodbye)."""
    print(f"BonicBot: {text}")
    speak_async(text, syn_config)
 
 
# ============================================================================
# SPEECH-TO-TEXT — faster-whisper, fully offline
# ============================================================================
_TRANSCRIBE_PARAMS = set(inspect.signature(WhisperModel.transcribe).parameters)
 
def _build_transcribe_kwargs():
    kwargs = {"language": "en", "beam_size": 5}
    if "temperature" in _TRANSCRIBE_PARAMS:
        kwargs["temperature"] = 0.0   # deterministic, less prone to hallucinating text
    if "condition_on_previous_text" in _TRANSCRIBE_PARAMS:
        kwargs["condition_on_previous_text"] = False   # stops hallucination loops feeding on themselves
    if "vad_filter" in _TRANSCRIBE_PARAMS:
        kwargs["vad_filter"] = True   # skip silent/non-speech stretches entirely
        if "vad_parameters" in _TRANSCRIBE_PARAMS:
            kwargs["vad_parameters"] = dict(min_silence_duration_ms=500)
    return kwargs
 
 
def listen(recognizer, mic, whisper_model):
    """Records one utterance from the microphone and transcribes it locally."""
    with mic as source:
        print("\n🎤 Listening... (say 'goodbye' to quit)")
        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, **_build_transcribe_kwargs())
 
    kept = []
    for seg in segments:
        no_speech_prob = getattr(seg, "no_speech_prob", 0.0)
        avg_logprob = getattr(seg, "avg_logprob", 0.0)
        if no_speech_prob > NO_SPEECH_PROB_THRESHOLD or avg_logprob < AVG_LOGPROB_THRESHOLD:
            print(f"  (dropped low-confidence audio: {seg.text.strip()!r})")
            continue
        kept.append(seg.text)
 
    text = " ".join(kept).strip()
    if not text:
        return None
    print(f"You said: {text}")
    return text
 
 
# ============================================================================
# LOCAL LLM WITH NATIVE TOOL-CALLING (via Ollama)
# ============================================================================
def _chat(history, syn_config):
    """Thin wrapper around ollama.chat. Streams the response live to the 
    terminal, dynamically parsing out sentences as they complete to queue
    them synchronously into the background speech pipeline."""
    try:
        stream = ollama.chat(
            model=OLLAMA_MODEL, messages=history, tools=[LOOKUP_TOOL],
            think=True, options={"temperature": 0},
            stream=True
        )
    except ollama.ResponseError as e:
        if e.status_code == 404:
            ensure_model_pulled(OLLAMA_MODEL)
            stream = ollama.chat(
                model=OLLAMA_MODEL, messages=history, tools=[LOOKUP_TOOL],
                think=True, options={"temperature": 0},
                stream=True
            )
        else:
            raise
 
    assembled_content = ""
    final_tool_calls = None
 
    # Instantly cut off any leftover audio from prior responses
    stop_tts()
 
    # Print in gray so the entire generation (thinking block included) is distinct
    sys.stdout.write("\n  \033[90m[Generating...] ")
    sys.stdout.flush()
 
    buffer = ""
    speaking_buffer = ""
    parse_ptr = 0
    in_think = False
 
    for chunk in stream:
        msg = chunk.get("message", {})
        content_chunk = msg.get("content", "")
        
        if content_chunk:
            assembled_content += content_chunk
            
            # Print the chunk to console immediately
            sys.stdout.write(content_chunk)
            sys.stdout.flush()
            
            # Feed characters into speech pipeline buffer
            buffer += content_chunk
            
            # Real-time state machine parsing out `<think>...</think>` boundaries 
            # to keep internal monologues strictly out of BonicBot's voice
            while parse_ptr < len(buffer):
                if not in_think:
                    think_start = buffer.lower().find("<think>", parse_ptr)
                    if think_start != -1:
                        speaking_buffer += buffer[parse_ptr:think_start]
                        parse_ptr = think_start + len("<think>")
                        in_think = True
                    else:
                        potential_start = buffer.lower().rfind("<", parse_ptr)
                        if potential_start != -1 and len(buffer) - potential_start < len("<think>"):
                            speaking_buffer += buffer[parse_ptr:potential_start]
                            parse_ptr = potential_start
                            break
                        else:
                            speaking_buffer += buffer[parse_ptr:]
                            parse_ptr = len(buffer)
                else:
                    think_end = buffer.lower().find("</think>", parse_ptr)
                    if think_end != -1:
                        parse_ptr = think_end + len("</think>")
                        in_think = False
                    else:
                        potential_end_start = buffer.lower().rfind("<", parse_ptr)
                        if potential_end_start != -1 and len(buffer) - potential_end_start < len("</think>"):
                            parse_ptr = potential_end_start
                            break
                        else:
                            parse_ptr = len(buffer)
            
            # Extract and queue completed sentences from the filtered speaking buffer
            while True:
                match = re.search(r'([.!?])(\s+|$)', speaking_buffer)
                if match:
                    end_idx = match.end()
                    sentence = speaking_buffer[:end_idx].strip()
                    speaking_buffer = speaking_buffer[end_idx:]
                    if sentence:
                        cleaned = clean_for_speech(sentence)
                        if cleaned:
                            tts_queue.put((cleaned, syn_config))
                else:
                    if '\n' in speaking_buffer:
                        parts = speaking_buffer.split('\n', 1)
                        sentence = parts[0].strip()
                        speaking_buffer = parts[1]
                        if sentence:
                            cleaned = clean_for_speech(sentence)
                            if cleaned:
                                tts_queue.put((cleaned, syn_config))
                    else:
                        break
 
        if msg.get("tool_calls"):
            final_tool_calls = msg["tool_calls"]
 
    # Flush any remaining unpunctuated text at the end of generation
    if speaking_buffer.strip():
        cleaned = clean_for_speech(speaking_buffer.strip())
        if cleaned:
            tts_queue.put((cleaned, syn_config))
 
    # Reset terminal color back to default
    sys.stdout.write("\033[0m\n")
    sys.stdout.flush()
 
    final_message = {"role": "assistant", "content": assembled_content}
    if final_tool_calls:
        final_message["tool_calls"] = final_tool_calls
 
    return {"message": final_message}
 
 
def _extract_query(call):
    args = call.get("function", {}).get("arguments", {})
    if isinstance(args, str):
        try:
            args = json.loads(args)
        except (json.JSONDecodeError, TypeError):
            args = {}
    return args.get("query", "") if isinstance(args, dict) else ""
 
 
def ask_llm(history, syn_config):
    """Sends the conversation to the local model with the lookup_topic tool
    available and thinking enabled."""
    response = _chat(history, syn_config)
    message = response["message"]
    tool_calls = message.get("tool_calls")
 
    if tool_calls:
        history.append({
            "role": "assistant",
            "content": message.get("content", ""),
            "tool_calls": tool_calls,
        })
        for call in tool_calls:
            query = _extract_query(call)
            print(f"  🔎 Looking up: {query}")
            result = lookup_topic(query)
            if result is None:
                print(f"  ⚠️ No summary found for '{query}'.")
                result = f"No information was found for '{query}'."
            history.append({"role": "tool", "content": result})
 
        response = _chat(history, syn_config)
        message = response["message"]
 
    final_text = message["content"].strip()
    if PUNT_PATTERN.search(final_text):
        final_text = "Sorry, could you tell me a bit more about who or what you mean?"
        stop_tts()
        speak_async(final_text, syn_config)
    return final_text
 
 
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
 
 
# ============================================================================
# MAIN LOOP — listen, think (decide: answer / tool / clarify), speak
# ============================================================================
def main():
    ensure_ollama_installed()
    ensure_ollama_running()
    ensure_model_pulled(OLLAMA_MODEL)
 
    model_path, config_path = ensure_piper_voice_ready(PIPER_VOICE_NAME, PIPER_VOICES_DIR)
    print("Loading Piper voice into memory...")
    piper_voice = PiperVoice.load(model_path, config_path=config_path)
    pyaudio_instance = pyaudio.PyAudio()
 
    # Start the daemon background TTS speaker thread
    t = threading.Thread(target=tts_worker, args=(piper_voice, pyaudio_instance), daemon=True)
    t.start()
 
    print("Loading Whisper model into memory...")
    whisper_model = WhisperModel(WHISPER_MODEL_SIZE, device="cpu", compute_type="int8")
 
    recognizer = sr.Recognizer()
    recognizer.pause_threshold = 1.5   
    recognizer.non_speaking_duration = 0.5
    recognizer.dynamic_energy_threshold = True
    mic = sr.Microphone(sample_rate=16000)   # 16kHz is what Whisper expects
 
    syn_config = SynthesisConfig(length_scale=PIPER_LENGTH_SCALE, volume=PIPER_VOLUME)
    history = [{"role": "system", "content": SYSTEM_PROMPT}]
    last_user_text = None
 
    print("Calibrating microphone for background noise, please stay quiet...")
    with mic as source:
        recognizer.adjust_for_ambient_noise(source, duration=2.0)
 
    try:
        # Speak the concise intro message
        print_and_speak(INTRO_MESSAGE, syn_config)
 
        while True:
            # Wait until BonicBot is completely finished speaking before opening the microphone.
            # This elegantly solves the feedback loop!
            wait_for_tts()
 
            user_text = listen(recognizer, mic, whisper_model)
 
            if user_text is None:
                continue
 
            # Halt any playing text immediately when the user finishes speaking a command
            stop_tts()
 
            if user_text.strip().lower() in EXIT_WORDS:
                print_and_speak("Goodbye! It was nice talking with you.", syn_config)
                # Wait until the goodbye message finishes speaking before exiting the process
                wait_for_tts()
                break
 
            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. Either try the lookup_topic tool with a more specific "
                    "query, or ask a clarifying question about what they meant.]"
                )
            else:
                turn_content = user_text
 
            history.append({"role": "user", "content": turn_content})
            last_user_text = user_text
 
            reply = ask_llm(history, syn_config)
            history.append({"role": "assistant", "content": reply})
            history = trim_history(history)
 
    except KeyboardInterrupt:
        print("\nShutting down...")
    finally:
        pyaudio_instance.terminate()
 
 
if __name__ == "__main__":
    main()

Calibrating microphone for background noise, please stay quiet...
BonicBot: Hello! I'm BonicBot. Ask me anything, and I'll look things up if I need to. Say goodbye to stop.

🎤 Listening... (say 'goodbye' to quit)
You said: what is the tallest mountain in the world
Thinking...
  🔎 Looking up: Mount Everest
BonicBot: The tallest mountain in the world is Mount Everest, standing over 8,800 meters tall!

🎤 Listening... (say 'goodbye' to quit)
You said: say hi
Thinking...
BonicBot: Hi there! Great to see you.

🎤 Listening... (say 'goodbye' to quit)
You said: goodbye
BonicBot: Goodbye! It was nice talking with you.

response = ollama.chat(model=OLLAMA_MODEL, messages=history, tools=[LOOKUP_TOOL])
message = response["message"]
if message.get("tool_calls"):
    ...