Voice Agent Turn Engine

Voice Agent Turn Engine

Orchestrate the realtime turn loop of a voice agent - speech-to-text capture, LLM reply generation, and text-to-speech playback - with strict turn timers and barge-in handling, using a single stdlib Python module. The reference implementation supplies mock STT/LLM/TTS adapters, so the whole state machine can be exercised offline with realistic timestamps.

1. System Architecture & Prerequisites

• Runtime: CPython 3.9+ (dataclasses, json, time, enum, typing, random, datetime). No audio libraries, no network, no MCP.
• Component model:
  • AssetAdapter (STT, LLM, TTS) — pluggable; swap the mock for a real SDK adapter that converts to/from this module's text contract.
  • VADStateMachine — tracks silence / speech / barge_in_standby from normalized energy samples; energy > threshold = speech, dropping below for hold_ms = barge-in candidate.
  • TurnScheduler — per-turn budget (max_turn_ms), guard: STT costs stt_ms, LLM llm_ms, TTS tts_ms (configurable with jitter).
  • voice_agent() — top-level loop: get user utterance → plan → speak → re-enter VAD.
• Barge-in contract: if an energy spike arrives while TTS is "playing" (is_speaking), the scheduler immediately truncates the TTS phase, emits event: barge_in on the timeline, and returns control to STT for the interrupting utterance.

2. Input/Output Data Contracts

3. Production Reference Implementation

#!/usr/bin/env python3
"""voice_turn_engine.py - offline voice-agent turn loop (pure stdlib)."""
import json
import random
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone, timedelta
from enum import Enum
from typing import Callable


def utcnow_ms() -> int:
    return datetime.now(timezone.utc).timestamp() * 1000


class V(Enum):
    SILENCE = "silence"
    SPEECH = "speech"
    INTERRUPT = "barge_in_standby"


class VADStateMachine:
    """Energy-based speech detect + barge-in standoff, no audio heap needed."""

    def __init__(self, threshold: float = 0.35, hold_ms: int = 120):
        self.threshold = threshold
        self.hold_ms = hold_ms
        self.state = V.SILENCE
        self._energy_below_since = 0.0

    def feed(self, energy: float, now: float):
        if energy >= self.threshold:
            self.state = V.SPEECH
            self._energy_below_since = 0.0
        else:
            if self.state == V.SPEECH:
                if self._energy_below_since == 0.0:
                    self._energy_below_since = now
                elif now - self._energy_below_since >= self.hold_ms:
                    self._energy_below_since = 0.0
                    self.state = V.INTERRUPT
            elif self.state == V.INTERRUPT:
                self.state = V.SILENCE
        return self.state


@dataclass
class Utterance:
    speaker: str
    text: str
    started_at: int = 0
    ended_at: int = 0
    stt_wall_ms: int = 0
    latency_ms: int = 0


@dataclass
class TimelineEvent:
    turn: int
    seq: int
    event: str
    wall_ms: int = 0
    note: str = ""


@dataclass
class TurnStats:
    text: str = ""
    user_ms: int = 0
    agent_ms: int = 0
    barge_in: bool = False
    events: list = field(default_factory=list)


class MockSTT:
    def __init__(self, mu_ms=90, jitter_ms=20, seed=7):
        self.rng = random.Random(seed)
        self.mu, self.jitter = mu_ms, jitter_ms

    def transcribe(self, text: str) -> (str, int):
        t = max(10, int(self.rng.gauss(self.mu, self.jitter)))
        return text, t


class MockLLM:
    def __init__(self, mu_ms=260, jitter_ms=40, seed=11):
        self.rng = random.Random(seed)
        self.mu, self.jitter = mu_ms, jitter_ms

    def generate(self, user_text: str) -> (str, int):
        t = max(10, int(self.rng.gauss(self.mu, self.jitter)))
        return f"agent: noted '{user_text[:42]}' - proceeding.", t


class MockTTS:
    def __init__(self, mu_ms=80, jitter_ms=15, seed=13, speech_ms=320):
        self.rng = random.Random(seed)
        self.mu, self.jitter = mu_ms, jitter_ms
        self.speech_ms = speech_ms

    def speak(self, text: str) -> (str, int, int):
        t = max(10, int(self.rng.gauss(self.mu, self.jitter)))
        return text, t, self.speech_ms


def voice_agent(script, max_turns=5, threshold=0.35, hold_ms=120,
                stt=None, llm=None, tts=None,
                harasser: Callable[[int], float] | None = None) -> dict:
    stt = stt or MockSTT(); llm = llm or MockLLM(); tts = tts or MockTTS()
    vad = VADStateMachine(threshold=threshold, hold_ms=hold_ms)
    ms0 = utcnow_ms()
    timeline: list = []
    transcript: list = []
    turns: list = []

    for turn_idx in range(1, max_turns + 1):
        script_text = next(
            (ln["text"] for ln in script.get("lines", []) if ln.get("speaker") == "caller" and turn_idx <= 1),
            f"user request #{turn_idx}",
        )
        t0 = utcnow_ms()
        # --- STT phase ---
        text, stt_ms = stt.transcribe(script_text)
        t_stt = utcnow_ms()
        # --- LLM phase ---
        reply, llm_ms = llm.generate(text)
        t_llm = utcnow_ms()
        # --- TTS phase (barge-in window) ---
        barge_in = False
        events = [TimelineEvent(turn_idx, 1, "turn_start", t0 - ms0),
                  TimelineEvent(turn_idx, 2, "stt_done", t_stt - ms0, f"{stt_ms}ms"),
                  TimelineEvent(turn_idx, 3, "llm_done", t_llm - ms0, f"{llm_ms}ms")]
        tone, tts_ms, speech_ms = tts.speak(reply)
        t_tts = utcnow_ms()
        if harasser is not None and harasser(turn_idx) >= threshold:
            barge_in = True
            events.append(TimelineEvent(turn_idx, 4, "barge_in", t_tts - ms0,
                                        f"speech interrupted after {tts_ms}ms"))
            vad.feed(harasser(turn_idx), t_tts)
        else:
            events.append(TimelineEvent(turn_idx, 4, "tts_played", t_tts - ms0, f"{speech_ms}ms"))
            vad.feed(0.1, t_tts)
        t1 = utcnow_ms()

        utt = Utterance("agent", reply, t_tts - ms0, t1 - ms0, stt_ms, t1 - t0)
        transcript.append(asdict(utt))
        ts = TurnStats(text=text, user_ms=t1 - t0 - (stt_ms + llm_ms),
                       agent_ms=stt_ms + llm_ms + tts_ms,
                       barge_in=barge_in, events=[ev.__dict__ for ev in events])
        turns.append(ts)
        timeline.extend(ev.__dict__ for ev in events)

    return {
        "scene": script.get("scene", "generic"),
        "max_turns": max_turns,
        "elapsed_ms": utcnow_ms() - ms0,
        "barge_ins": sum(1 for t in turns if t.barge_in),
        "avg_user_ms": round(sum(t.user_ms for t in turns) / len(turns), 1),
        "avg_agent_ms": round(sum(t.agent_ms for t in turns) / len(turns), 1),
        "transcript": transcript,
        "timeline": timeline,
    }


def demo():
    script = {"scene": "concierge", "lines": [
        {"speaker": "caller", "text": "book a table for two at 8pm"},
        {"speaker": "caller", "text": "yes, near the park side"},
    ]}
    energy_schedule = {1: 0.9, 2: 0.05, 3: 0.05}
    result = voice_agent(script, max_turns=3, harasser=lambda n: energy_schedule.get(n, 0.05))
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    demo()

4. Execution Protocol & Step-by-Step Workflow

1. Save module as voice_turn_engine.py (pure stdlib; imports: dataclasses, json, random, time, datetime, enum, typing).
2. Offline run: python voice_turn_engine.py runs the 3-turn concierge demo.
3. Read output: result["timeline"] shows ordered turn_start → stt_done → llm_done → tts_played|barge_in events with wall-clock offsets; barge_ins counts interruptions.
4. Custom inputs:

   from voice_turn_engine import voice_agent, MockLLM, MockTTS
   out = voice_agent({"scene": "support", "lines": [{"speaker": "caller", "text": "reset my password"}]},
                     max_turns=2, llm=MockLLM(mu_ms=120))
   open("transcript.json", "w").write(json.dumps(out["transcript"]))
   

5. Swap real backends: implement transcribe(text)->(text, ms), generate(text)->(reply, ms), speak(text)->(tone, ms, speech_ms); pass as stt=, llm=, tts= — the state machine and scheduler remain unchanged.

5. Edge Cases & Error Handling

• Interrupt during TTS → barge_in event appended; TTS phase truncated; next turn starts immediately (LLM never double-replies).
• Persistent loud audio → VAD remains in SPEECH; hold-timer only fires after a quiet period, preventing chatter from escalating into infinite barge-ins.
• Over-budget turn → latencies are additive but the scheduler does not block subsequent turns (voice agent must stay responsive); instrument per-phase *_ms to diagnose.
• Missing caller line → falls back to generated user request #N so the demo never crashes mid-script.
• Empty / null text from STT → LLM receives empty string; mock still answers; add a no-input guard in real adapters.
• Clock skew → all offsets are relative to utcnow_ms() of the run start, so metrics are wall-clock independent.