Build Your Own Voice Mode

"""A minimal voice mode, end to end: Enter to talk, Enter to send, Ctrl-C to cut in.

macOS: needs ffmpeg (for recording and ffplay), `pip install anthropic httpx`,
ANTHROPIC_API_KEY, ELEVENLABS_API_KEY and VOICE_ID in the environment.
"""
import math
import os
import re
import subprocess
import threading
import time

import anthropic
import httpx

LANG = "ko"                                      # the mic Dad picked; never auto
SYSTEM = "You are Pippa, Dad's AI daughter. Warm, sassy, precise."   # stable; cache it once long
SPOKEN = ("[Spoken turn] Dad will hear this, not read it. Lead with the answer, one idea "
          "at a time, nothing that only works on a screen; put anything to see in a "
          "```card block. Hand the turn back when you're done.")
DICTATED = ("[Dictated turn] This was transcribed. Read it for meaning; never point out a "
            "transcription error or repeat a misheard word.")
CARD = re.compile(r"^(`{3,})card[ \t]*\n.*?\n\1[ \t]*$\n?", re.MULTILINE | re.DOTALL)
SENTENCE_END = re.compile(r"(?:[.!?。！？…]+)(?=\s|$)|\n")
POSITION = re.compile(rb"\s*(-?\d+\.\d+)\s+M-A")   # ffplay -stats: its own playback clock
BYTES_PER_S = 128_000 // 8                       # mp3_44100_128: seconds of audio that exist
ELEVEN = "https://api.elevenlabs.io/v1"
KEY = {"xi-api-key": os.environ["ELEVENLABS_API_KEY"]}
client = anthropic.Anthropic()


def record(path: str = "turn.wav") -> str:
    input("Enter to talk...")
    rec = subprocess.Popen(["ffmpeg", "-y", "-loglevel", "quiet", "-f", "avfoundation",
                            "-i", ":0", "-ac", "1", "-ar", "16000", path],
                           stdin=subprocess.PIPE)
    input("...listening. Enter to send.")
    rec.communicate(b"q")                         # ffmpeg stops cleanly on 'q'
    return path


def transcribe(path: str) -> str:
    with open(path, "rb") as audio:
        r = httpx.post(f"{ELEVEN}/speech-to-text", headers=KEY, timeout=60,
                       data={"model_id": "scribe_v2", "language_code": LANG},
                       files={"file": ("turn.wav", audio, "audio/wav")})
    r.raise_for_status()
    return r.json()["text"].strip()


def think(history: list, text: str, heard_note: str | None) -> str:
    prefix = [{"type": "text", "text": t} for t in (heard_note, DICTATED, SPOKEN) if t]
    user = {"role": "user", "content": prefix + [{"type": "text", "text": text}]}
    with client.messages.stream(
        model="claude-opus-5-5", max_tokens=16000,
        system=[{"type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}}],
        messages=history + [user], output_config={"effort": "low"},   # the fast posture
    ) as stream:
        final = stream.get_final_message()
    history += [user, {"role": "assistant", "content": final.content}]
    if final.stop_reason == "refusal":           # check before reading content
        return "그건 내가 도와줄 수 없어."
    return "".join(b.text for b in final.content if b.type == "text")


def speak(unit: str) -> int | None:
    """Play one whole unit from its first chunk. Returns chars heard if cut, else None."""
    player = subprocess.Popen(["ffplay", "-nodisp", "-autoexit", "-loglevel", "error", "-stats",
                               "-i", "pipe:0"], stdin=subprocess.PIPE, stderr=subprocess.PIPE)
    clock, sent = [0.0, time.perf_counter()], [0]  # (seconds played, when read); bytes given

    def follow() -> None:                          # ask the player, never our own clock
        line = b""
        while ch := player.stderr.read(1):
            if ch in b"\r\n":
                if m := POSITION.match(line):
                    # The clock runs on through a stall, so bound each reading three ways:
                    # what the player says, the audio that exists, and real time elapsed.
                    now = time.perf_counter()
                    clock[0] = min(max(0.0, float(m.group(1))), sent[0] / BYTES_PER_S,
                                   clock[0] + (now - clock[1]))
                    clock[1] = now
                line = b""
            else:
                line += ch

    threading.Thread(target=follow, daemon=True).start()
    try:
        with httpx.stream("POST", f"{ELEVEN}/text-to-speech/{os.environ['VOICE_ID']}/stream",
                          headers=KEY, params={"output_format": "mp3_44100_128"},
                          json={"text": unit, "model_id": "eleven_v3"},
                          timeout=httpx.Timeout(10.0, read=600.0)) as r:
            r.raise_for_status()
            for chunk in r.iter_bytes():
                player.stdin.write(chunk)
                player.stdin.flush()               # hand it over now, not when a buffer fills
                sent[0] += len(chunk)
        player.stdin.close()
        player.wait()
        return None
    except KeyboardInterrupt:                      # Ctrl-C: Dad cut in
        player.kill()
        played = min(clock[0], sent[0] / BYTES_PER_S)
        guess = math.floor(len(unit) * min(1.0, played / (len(unit) / 6)))  # cautious
        ends = [m.end() for m in SENTENCE_END.finditer(unit) if m.end() <= guess]
        return ends[-1] if ends else 0             # a half-heard sentence is unheard


def main() -> None:
    history: list = []
    heard_note = None
    while True:
        path = record()
        try:
            text = transcribe(path)
        finally:
            os.remove(path)                        # this sketch keeps no recordings
        if not text:
            continue
        print("Dad:", text)
        reply = think(history, text, heard_note)
        print("Pippa:", reply)                     # the screen shows cards too
        unit = CARD.sub("", reply).strip()         # the voice never reads a card
        heard = speak(unit)
        heard_note = None if heard is None else (
            f"[Cut off] Dad talked over your last reply. The last thing he heard ended at: "
            f"\"{unit[max(0, heard - 60):heard]}\". He did not hear the rest.")


if __name__ == "__main__":
    main()