Spaces:

Jacid23
/

Lyon_chatbox

Running

App Files Files Community

Lyon_chatbox / src /lyon_chatbox /cascade /CASCADE_CODEBASE.md

Jacid23

Rebrand app as Lyon Chatbox

dfedf76 23 days ago

preview code

Raw

History Blame Contribute Delete

43 kB

Cascade Mode - Developer Documentation

This document provides detailed technical documentation for the cascade conversation mode in the Lyon Chatbox.

Overview

Cascade Mode implements a traditional ASR → LLM → TTS conversation pipeline for the Reachy Mini robot. It processes user speech through automatic speech recognition, generates AI responses via configurable LLM providers, and synthesizes speech output with synchronized head animations.

Key Design Principles

Provider abstraction - Swap ASR/LLM/TTS implementations via configuration
Latency optimization - Pre-warmed audio playback threads, parallel TTS generation
Clean separation - Handler (business logic) vs UI (presentation & I/O)

Current Scope

Gradio UI with continuous VAD recording
Console mode with VAD-based speech detection
Test file mode for automated end-to-end testing (--test-file)
Optional streaming ASR support
Head wobbler animation synchronized with audio playback

File Structure

cascade/
├── __init__.py                        # Package init (no public exports)
├── entry.py                           # Entry point from main.py
├── handler.py                         # Core orchestrator (lifecycle, ASR routing, state)
├── provider_factory.py                # Provider initialization (ASR/LLM/TTS factory functions)
├── pipeline.py                        # LLM response processing & tool execution
├── speech_output.py                   # SpeechOutput protocol + Gradio/Console implementations
├── turn_result.py                     # TurnResult + TurnItem dataclasses
├── config.py                          # Configuration loader (cascade.yaml)
├── timing.py                          # Latency tracking & profiling
├── vad.py                             # Silero VAD + VADStateMachine (shared by console & Gradio)
├── console.py                         # Console mode with VAD (CascadeLocalStream)
├── autotest_stream.py                     # Test file mode (automated TTS→ASR→LLM→TTS testing)
│
├── ui/                                # Gradio interface components
│   ├── __init__.py                    # Exports CascadeGradioUI
│   ├── audio_playback.py              # Pre-warmed audio output system
│   ├── audio_recording.py             # VAD-based continuous recording
│   └── gradio_app.py                  # Main Gradio interface
│
├── asr/                               # Automatic Speech Recognition
│   ├── __init__.py                    # Provider exports
│   ├── base.py                        # ASRProvider abstract base
│   ├── base_streaming.py              # StreamingASRProvider abstract base
│   ├── audio_utils.py                 # Shared WAV parsing & resampling (librosa)
│   ├── progressive_base.py            # ProgressiveASRBase — shared sliding window logic + DecodeResult/SentenceSegment
│   ├── whisper_openai.py              # OpenAI Whisper implementation
│   ├── parakeet_mlx_progressive.py    # Parakeet MLX progressive (Apple Silicon, inherits ProgressiveASRBase)
│   ├── parakeet_nemo_progressive.py   # Parakeet NeMo progressive (CUDA, inherits ProgressiveASRBase)
│   ├── deepgram.py                    # Deepgram streaming implementation
│   ├── nemotron.py                    # Nemotron ASR implementation
│   └── openai_realtime_asr.py         # OpenAI Realtime streaming implementation
│
├── llm/                               # Large Language Models
│   ├── __init__.py                    # Provider exports
│   ├── base.py                        # LLMProvider + LLMChunk abstractions
│   ├── openai.py                      # OpenAI GPT implementation
│   └── gemini.py                      # Google Gemini implementation
│
├── tts/                               # Text-to-Speech
│   ├── __init__.py                    # Provider exports
│   ├── base.py                        # TTSProvider abstract base
│   ├── utils.py                       # Shared TTS utilities (silence trimming)
│   ├── openai.py                      # OpenAI TTS implementation
│   ├── kokoro.py                      # Kokoro local TTS
│   └── elevenlabs.py                  # ElevenLabs API implementation
│
└── transcript_analysis/               # Real-time reactions to user speech
    ├── __init__.py                    # Package exports (EntityAnalyzer optional)
    ├── base.py                        # Data classes + TranscriptAnalyzer ABC
    ├── loader.py                      # YAML loader, callback importer
    ├── manager.py                     # Orchestrator, deduplication, dispatch
    ├── keyword_analyzer.py            # Keyword/glob matching
    └── entity_analyzer.py             # NER via GLiNER (optional dependency)

Core Components

CascadeHandler (`handler.py`)

The central orchestrator that manages conversation state, ASR routing, and lifecycle. Provider initialization is in provider_factory.py; LLM/tool pipeline logic is in pipeline.py.

Responsibilities:

Manage conversation history (OpenAI message format)
Route audio to ASR (manual and streaming paths)
Transcript analysis callbacks (fire-and-forget)
Cost tracking
Run async event loop in background thread

Key Attributes:

deps: ToolDependencies            # Tool dependencies (robot, vision, movement, etc.)
speech_output: SpeechOutput|None  # TTS playback backend (set by console or Gradio frontend)
asr: ASRProvider                  # Current ASR implementation
llm: LLMProvider                  # Current LLM implementation
tts: TTSProvider                  # Current TTS implementation
conversation_history: List        # OpenAI format: [{"role": "...", "content": ...}]
processing_lock: asyncio.Lock     # Prevent concurrent audio processing
tool_specs: List[Dict]            # Tools in Chat Completions format
is_streaming_asr: bool            # Whether current ASR provider supports streaming
transcript_manager                # TranscriptAnalysisManager | NoOpTranscriptManager
_captured_frames: list[bytes]     # Side-channel storage for see_image_through_camera JPEG frames
_current_turn_items: list[TurnItem]  # Per-turn accumulator for displayable items
_turn_results: list[TurnResult]   # Completed turns (private; use turn_results property)
cumulative_cost: float            # Running cost total across all turns

Key Properties:

turn_results -> list[TurnResult]
    # Public read-only accessor for completed turns (used by UI poller)

Key Methods:

async process_audio_manual(audio_bytes) -> TurnResult
    # Entry point for push-to-talk recording

async process_audio_streaming_start() -> None
    # Initialize streaming ASR session

async process_audio_streaming_chunk(chunk) -> Optional[str]
    # Send chunk to streaming ASR, get partial transcript

async process_audio_streaming_end() -> TurnResult
    # Finalize streaming ASR, run LLM pipeline

def start() -> None
    # Start background event loop (Gradio mode only)

def stop() -> None
    # Stop background event loop

def clear_state() -> None
    # Reset conversation history, captured frames, and turn results

async _run_pipeline_after_transcription(transcript) -> TurnResult
    # Shared post-ASR pipeline (validate → history → LLM → TTS → result)
    # Called by both manual and streaming paths

Provider Factory (`provider_factory.py`)

Pure factory functions for constructing ASR/LLM/TTS providers and transcript analysis from cascade.yaml config. No runtime state — called once during CascadeHandler.__init__.

Key Functions:

init_asr_provider() -> ASRProvider
init_llm_provider() -> LLMProvider
init_tts_provider() -> TTSProvider
init_transcript_analysis(deps) -> TranscriptAnalysisManager | NoOpTranscriptManager

Pipeline (`pipeline.py`)

Module-level async functions for LLM response processing and tool execution. Mutates conversation history, turn items, and captured frames in-place via PipelineContext.

PipelineContext:

@dataclass
class PipelineContext:
    llm: LLMProvider
    tts: TTSProvider
    speech_output: SpeechOutput | None
    conversation_history: list[dict[str, Any]]
    tool_specs: list[dict[str, Any]]
    current_turn_items: list[TurnItem]
    captured_frames: list[bytes]
    deps: ToolDependencies
    aggregate_cost_fn: Callable

Built once in handler._run_pipeline_after_transcription() and threaded through all pipeline calls.

Key Functions:

async process_llm_response(ctx: PipelineContext) -> None
    # Stream LLM, collect text/tool calls, dispatch to execute_tool_calls
    # Auto-injects a synthetic speak tool call if the LLM returns text without
    # using the speak tool (fallback for models that skip the tool)

async execute_tool_calls(tool_calls, ctx: PipelineContext) -> None
    # Execute individual tools, handle see_image_through_camera/speak specially:
    #   speak → calls speech_output.speak() for TTS synthesis + playback
    #   see_image_through_camera → stores JPEG in captured_frames, replaces b64 in history, re-calls LLM for analysis

TurnResult (`turn_result.py`)

Structured result returned by the handler after each conversation turn. Decouples handler output from UI rendering — the UI never parses conversation_history directly.

@dataclass
class TurnItem:
    kind: str         # "speak" | "image" | "tool" | "assistant"
    text: str         # For speak/assistant items
    image_jpeg: bytes # For image items (raw JPEG bytes)
    tool_name: str    # For tool items
    tool_content: str # For tool items (JSON string)

@dataclass
class TurnResult:
    transcript: str           # User's speech transcript
    items: list[TurnItem]     # Ordered displayable items
    cost: float               # ASR + LLM cost for this turn (not TTS)

    speak_text: str           # (property) All speak items joined with ". "
    has_speak: bool           # (property) Whether any speak item exists

How items are populated:

speak — from pipeline.execute_tool_calls() when speak tool is called
image — from pipeline.execute_tool_calls() when see_image_through_camera tool returns JPEG
tool — from pipeline.execute_tool_calls() for other tools (movements, etc.)
assistant — from pipeline.process_llm_response() when LLM returns text + tool calls but no speak

The handler stores completed turns in _turn_results (exposed as handler.turn_results property), used by the UI's continuous-mode poller.

UI Components (`ui/`)

The Gradio interface is split into focused modules:

AudioPlaybackSystem (`ui/audio_playback.py`)

Pre-warmed audio output system for low-latency TTS playback.

Responsibilities:

Detect playback backend (sounddevice vs robot.media)
Manage persistent playback and wobbler threads
Queue-based communication for zero-startup-cost playback

Key Interface:

playback = AudioPlaybackSystem(robot, head_wobbler, shutdown_event)
playback.put_audio(chunk)      # Queue audio for playback
playback.put_wobbler(chunk)    # Queue wobbler data
playback.signal_end_of_turn()  # Signal end of speech
playback.close()               # Shutdown threads

Recording Classes (`ui/audio_recording.py`)

ContinuousState: Enum for Gradio VAD lifecycle (IDLE → LISTENING → RECORDING → PROCESSING). IDLE is Gradio-specific; LISTENING/RECORDING/PROCESSING map to VADState values.

StreamingASRCallbacks: Dataclass for injecting ASR callbacks without coupling to handler.

ContinuousVADRecorder: VAD-based continuous recording, backed by VADStateMachine.

recorder = ContinuousVADRecorder(
    sample_rate, streaming_callbacks, on_speech_captured
)
recorder.start()   # Start VAD loop (creates VADStateMachine)
recorder.stop()    # Stop VAD loop
recorder.state     # Current ContinuousState (maps from VADStateMachine.state)

CascadeGradioUI (`ui/gradio_app.py`)

Main orchestrator that ties everything together.

Responsibilities:

Build Gradio interface (chatbot, buttons, status)
Coordinate audio pipeline (ASR → LLM → TTS)
Coordinate playback and VAD recording subsystems
Handle Gradio events and lifecycle

Key Attributes:

handler: CascadeHandler        # Reference to cascade handler
robot: ReachyMini | None       # Robot instance
playback: AudioPlaybackSystem  # Audio output
shutdown_event: threading.Event # Coordinated shutdown

Key Methods:

create_interface() -> gr.Blocks     # Build Gradio UI
launch(**kwargs) -> None            # Start server
close() -> None                     # Shutdown all subsystems

SpeechOutput (`speech_output.py`)

Protocol and implementations for TTS synthesis + playback. The handler calls speech_output.speak(text) inside execute_tool_calls when the speak tool fires — this is where TTS actually happens.

Protocol:

class SpeechOutput(Protocol):
    async def speak(text: str) -> None

Implementations:

Class	Used by	Behavior
`ConsoleSpeechOutput`	`CascadeLocalStream`, `CascadeTestStream`	Streams TTS chunks to a playback callback with rate limiting; drives head wobbler per chunk
`GradioSpeechOutput`	`CascadeGradioUI`	Splits text into sentences via `split_into_sentences()`, generates TTS in parallel with gate-event ordering, queues to `AudioPlaybackSystem`

split_into_sentences(text, min_length=8) — Splits on .!?,;—, keeps punctuation attached, merges short segments under min_length characters.

VAD State Machine (`vad.py`)

VADStateMachine extracts the shared pre-roll → speech-start → speech-end logic used by both console and Gradio modes. Callers feed audio chunks via process_chunk() and react to returned VADEvents.

vad_sm = VADStateMachine(vad)
event = vad_sm.process_chunk(audio_chunk)

if event == VADEvent.SPEECH_STARTED:
    # Pre-roll + current chunk available in vad_sm.speech_chunks
elif event == VADEvent.SPEECH_ENDED:
    # All speech frames in vad_sm.speech_chunks
    vad_sm.finish_processing()  # Reset to LISTENING
elif vad_sm.state == VADState.RECORDING:
    # Mid-recording: stream current chunk

States: LISTENING → RECORDING → PROCESSING → (finish_processing) → LISTENING

Console Mode (`console.py`)

VAD-based console interface for the cascade pipeline, used when neither --gradio nor --test-file is specified. Records from system mic, detects speech via VADStateMachine, and plays responses through the robot speaker.

Key Classes:

CascadeLocalStream — Stream manager that runs two concurrent async loops:
- _record_loop() — reads mic frames, processes through VADStateMachine
- _play_loop() — pulls audio from playback queue, resamples to output rate, pushes to robot.media

stream = CascadeLocalStream(handler, robot)
stream.launch()   # Blocking: runs asyncio event loop
stream.close()    # Stop media, cancel tasks

Wires ConsoleSpeechOutput into handler.speech_output at init time so the pipeline plays audio through the robot speaker.

Provider Abstractions

ASR Providers (`asr/`)

Base class (base.py):

class ASRProvider(ABC):
    async transcribe(audio_bytes, language) -> str

Streaming base (base_streaming.py):

class StreamingASRProvider(ASRProvider):
    async start_stream() -> None
    async send_audio_chunk(audio_chunk: bytes) -> None
    async get_partial_transcript() -> Optional[str]
    async end_stream() -> str                  # Final transcript

Implementations:

Provider	Type	Description
`WhisperOpenAIASR`	Batch	OpenAI Whisper API
`ParakeetMLXProgressiveASR`	Streaming	Local progressive with sliding window via mlx-audio (Apple Silicon)
`ParakeetNeMoProgressiveASR`	Streaming	Local progressive with sliding window via NeMo (CUDA)
`DeepgramASR`	Streaming	Deepgram Nova via WebSocket
`NemotronASR`	Streaming	NVIDIA Nemotron ASR
`OpenAIRealtimeASR`	Streaming	OpenAI Realtime API via WebSocket

Progressive ASR Base (progressive_base.py):

Both Parakeet providers inherit from ProgressiveASRBase, which contains the sentence-aware sliding window logic (~150 lines). Subclasses only implement three methods:

_decode(audio_np) -> DecodeResult — run inference, return text + sentence segments
_decode_full(audio_np) -> str — full-context decode for final transcription
_warmup() — model warmup (e.g. transcribe silence)

DecodeResult and SentenceSegment are shared dataclasses defined in progressive_base.py.

LLM Providers (`llm/`)

Base class (base.py):

@dataclass
class LLMChunk:
    type: str  # "text_delta" | "tool_call" | "done"
    content: Optional[str]      # For text_delta
    tool_call: Optional[Dict]   # For tool_call

class LLMProvider(ABC):
    async generate(messages, tools, temperature) -> AsyncIterator[LLMChunk]
    def parse_tool_call(tool_call) -> (call_id, tool_name, args_dict)
        # Default implementation handles OpenAI-style tool call format
        # Subclasses can override if needed

Implementations:

Provider	Models
`OpenAILLM`	GPT-4o-mini, GPT-5.2-chat (any OpenAI chat model)
`GeminiLLM`	Gemini 2.5 Flash Lite (any Gemini model)

TTS Providers (`tts/`)

Base class (base.py):

class TTSProvider(ABC):
    @property
    def sample_rate(self) -> int:  # Default 24000; override for non-24kHz providers
    async synthesize(text, voice=None) -> AsyncIterator[bytes]

Implementations:

Provider	Description
`OpenAITTS`	OpenAI TTS API (24kHz PCM int16)
`KokoroTTS`	Local TTS via Kokoro model
`ElevenLabsTTS`	ElevenLabs API

Data Flow

VAD-Triggered Pipeline

VAD detects speech end
  │
  ├─ Concatenate captured frames to WAV
  │
  └─> handler.process_audio_manual(wav_bytes)
      │
      ├─ PHASE 1: ASR
      │   └─ asr.transcribe(audio_bytes) → transcript
      │       └─ Add to conversation_history as user message
      │
      ├─ PHASE 2: LLM
      │   └─ llm.generate(history, tools) → response + tool_calls
      │       └─ Add assistant message to history
      │       └─ If text-only (no tool calls): auto-inject speak tool call
      │
      ├─ PHASE 3: Tool Execution + TTS
      │   └─ For each tool_call:
      │       ├─ Execute tool, add result to history
      │       └─ Special handling:
      │           • speak: call speech_output.speak(message)
      │             └─ TTS synthesis + playback happens HERE
      │           • see_image_through_camera: store JPEG, replace b64 in history, re-call LLM for analysis
      │
      └─> Return TurnResult to UI
          (transcript, items=[speak/image/tool/assistant], cost)

UI uses TurnResult.items for display only:
  │
  ├─ speak items → show in chatbot
  ├─ image items → decode JPEG, display
  └─ tool items → display with metadata

Streaming ASR Flow

User clicks START (streaming provider)
  │
  ├─ handler.process_audio_streaming_start()
  │   └─ Open WebSocket/stream connection
  │
During recording:
  │
  ├─ For each audio chunk:
  │   ├─ handler.process_audio_streaming_chunk(chunk)
  │   │   └─ Get partial transcript
  │   └─ Update UI with partial text
  │
User clicks STOP
  │
  └─ handler.process_audio_streaming_end()
      ├─ Close stream
      ├─ Get final transcript
      └─ Run LLM pipeline (same as batch mode)

Configuration

Config File: `cascade.yaml`

Each section has a provider: key selecting the active provider and a providers: dict defining all available providers. Each provider entry contains metadata keys (module, class, streaming, location, requires, hardware, description, import_check, install_extra) plus provider-specific settings.

Hardware Tags

Tag	Validation	Used by
`apple_silicon`	Hard error if not arm64 + Darwin	parakeet_mlx_progressive
`cuda`	Hard error if `torch.cuda.is_available()` is False	parakeet_nemo_progressive, nemotron
`null`	No check	kokoro, cloud providers

Dependency Checks

Local providers can declare import_check (module to import) and install_extra (uv extra name). At config load time, if the import fails, a RuntimeError is raised with install instructions.

asr:
  provider: parakeet_mlx_progressive  # Selected provider name
  providers:
    whisper_openai:
      module: whisper_openai
      class: WhisperOpenAIASR
      streaming: false
      requires: [OPENAI_API_KEY]
      model: whisper-1
    deepgram:
      module: deepgram
      class: DeepgramASR
      streaming: true
      requires: [DEEPGRAM_API_KEY]
      model: nova-2
    # ... other providers (parakeet_mlx_progressive, nemotron, openai_realtime_asr)

llm:
  provider: gemini-2.5-flash-lite
  temperature: 1.0
  providers:
    gpt-4o-mini:
      module: openai
      class: OpenAILLM
      requires: [OPENAI_API_KEY]
      model: gpt-4o-mini
    gemini-2.5-flash-lite:
      module: gemini
      class: GeminiLLM
      requires: [GEMINI_API_KEY]
      model: gemini-2.5-flash-lite

tts:
  provider: kokoro
  trim_silence: true
  providers:
    tts_openai:
      module: openai
      class: OpenAITTS
      requires: [OPENAI_API_KEY]
      voice: alloy
    kokoro:
      module: kokoro
      class: KokoroTTS
      hardware: null
      import_check: kokoro
      install_extra: cascade_kokoro
      voice: am_adam
    elevenlabs:
      module: elevenlabs
      class: ElevenLabsTTS
      requires: [ELEVENLABS_API_KEY]
      voice_id: "..."
      model: eleven_flash_v2_5

Environment Variables

OPENAI_API_KEY      # Required for OpenAI ASR/LLM/TTS
GEMINI_API_KEY      # Required for Gemini LLM
DEEPGRAM_API_KEY    # Required for Deepgram streaming ASR
ELEVENLABS_API_KEY  # Required for ElevenLabs TTS

Config Loading (`config.py`)

Lazy singleton via get_config() (created on first call, not on import):

from cascade.config import get_config
config = get_config()

# Key attributes:
config.asr_provider       # str — selected ASR provider name
config.llm_provider       # str — selected LLM provider name
config.tts_provider       # str — selected TTS provider name
config.asr_providers      # dict — all ASR provider definitions
config.llm_providers      # dict — all LLM provider definitions
config.tts_providers      # dict — all TTS provider definitions
config.llm_temperature    # float — LLM temperature (default 1.0)
config.tts_trim_silence   # bool — trim silence from TTS output
config.gliner_model       # str — GLiNER model for entity recognition
config.OPENAI_API_KEY     # str|None — from environment
config.DEEPGRAM_API_KEY   # str|None
config.GEMINI_API_KEY     # str|None
config.ELEVENLABS_API_KEY # str|None

# Helper methods:
config.get_asr_settings()    # Provider settings (excludes metadata)
config.is_asr_streaming()    # Whether selected ASR supports streaming
config.get_llm_settings()
config.get_tts_settings()

set_config(cfg) is available for test overrides.

Audio Playback System

Architecture

Pre-warmed persistent threads created at UI startup:

┌─────────────────────────┐
│ Playback Thread         │  ← Runs forever, blocks on queue
│ (sounddevice or robot)  │
└───────────▲─────────────┘
            │
       audio_queue
       (maxsize=100)
            │
┌───────────┴─────────────┐
│ GradioSpeechOutput      │  ← Enqueues TTS chunks (from speech_output.py)
│ .speak()                │
└───────────┬─────────────┘
            │
       wobbler_queue
       (maxsize=100)
            │
            ▼
┌─────────────────────────┐
│ Wobbler Thread          │  ← Runs forever, blocks on queue
│ (head animation)        │
└─────────────────────────┘

Playback Backends

sounddevice (laptop speakers):

sd.OutputStream(samplerate=24000, channels=1, dtype=int16)
Pre-initialized stream at startup
Direct chunk writes

robot.media (robot speakers):

robot.media.start_playing() called once
Convert int16 → float32, resample if needed
robot.media.push_audio_sample(audio_float)

Device Detection

The UI queries the default output device and checks for robot speaker keywords:

"respeaker", "xvf3800", "reachy"
Only uses robot.media if default matches a robot speaker

Module Dependencies

main.py
  │
  └─> entry.py (run_cascade_mode)
      │
      ├─> handler.py (CascadeHandler)
      │   │
      │   ├─> provider_factory.py (init_asr/llm/tts_provider)
      │   │   ├─> config.py
      │   │   ├─> asr/ providers
      │   │   ├─> llm/ providers
      │   │   └─> tts/ providers
      │   ├─> pipeline.py (process_llm_response, execute_tool_calls)
      │   │   └─> speech_output.py (SpeechOutput.speak() for TTS)
      │   └─> transcript_analysis/ (TranscriptAnalysisManager)
      │       ├─> loader.py (reads profiles/<name>/reactions.yaml)
      │       ├─> keyword_analyzer.py
      │       └─> entity_analyzer.py (optional, requires gliner)
      │
      ├─[--gradio]─> ui/gradio_app.py (CascadeGradioUI)
      │   ├─> ui/audio_playback.py (AudioPlaybackSystem)
      │   ├─> ui/audio_recording.py (Recorders)
      │   ├─> speech_output.py (GradioSpeechOutput → AudioPlaybackSystem)
      │   └─> handler (reference)
      │
      ├─[default]─> console.py (CascadeLocalStream)
      │   ├─> speech_output.py (ConsoleSpeechOutput → robot.media)
      │   ├─> vad.py (SileroVAD)
      │   └─> handler (reference)
      │
      └─[--test-file]─> autotest_stream.py (CascadeTestStream)
          ├─> speech_output.py (ConsoleSpeechOutput → sounddevice)
          └─> handler (reference)

Key relationships:

entry.py creates handler + one stream manager, wires speech_output into handler
Handler owns conversation state, provider instances, and transcript analysis
Pipeline calls handler.speech_output.speak() during tool execution — TTS happens inside the pipeline
Each stream manager provides its own SpeechOutput implementation:
- Gradio: GradioSpeechOutput (parallel sentence synthesis → pre-warmed playback threads)
- Console: ConsoleSpeechOutput (rate-limited streaming → robot speaker)
- Test: ConsoleSpeechOutput (rate-limited streaming → sounddevice callback)
AudioPlaybackSystem handles pre-warmed playback threads (Gradio mode only)
Recording classes encapsulate VAD-based continuous recording (Gradio mode only)
Transcript analysis runs in parallel with the main ASR → LLM → TTS pipeline

Entry Point & Integration

Entry Function (`entry.py`)

run_cascade_mode() selects one of three stream managers based on CLI flags:

Priority	Condition	Stream Manager	Source file	Handler lifecycle
1	`--test-file`	`CascadeTestStream`	`autotest_stream.py`	Synchronous (no `handler.start()`)
2	`--gradio`	`CascadeGradioUI.create_interface()`	`ui/gradio_app.py`	Background event loop (`handler.start()` / `handler.stop()`)
3	(default)	`CascadeLocalStream`	`console.py`	Synchronous (no `handler.start()`)

All three share the same shutdown sequence: stream_manager.close() → stop services → robot.media.close() → robot.client.disconnect() → os._exit(0).

Command Line Usage

lyon-chatbox --gradio --head-tracker yolo

Latency Tracking (`timing.py`)

LatencyTracker

Profiles end-to-end latency from user action to first audio playback.

Key Events (two flows):

Button flow (Gradio push-to-talk):

user_stop_click → recording_ready → transcribing_start → asr_complete → llm_start → llm_complete → tts_start → tts_first_chunk_ready → audio_playback_started

VAD flow (console mode, continuous mode, test file):

vad_speech_end → recording_captured → transcribing_start → asr_complete → llm_start → llm_complete → tts_start → tts_first_chunk_ready → audio_playback_started

print_summary() auto-detects which flow was used and displays the appropriate stages.

Usage:

from cascade.timing import tracker

tracker.reset("vad_speech_end")  # or "pipeline_start" for button flow
tracker.mark("event_name", {"metadata": "value"})
tracker.get_duration("start_event", "end_event")  # -> ms or None
tracker.print_summary()

Transcript Analysis (Live Reactions)

The transcript analysis system triggers reactive robot behaviors (sounds, movements, emotions) in real time as the user speaks, independently of the LLM pipeline. Reactions fire based on keywords or named entities detected in the ASR transcript.

Architecture

ASR transcript (partial or final)
  │
  └─> TranscriptAnalysisManager
      │
      ├─ KeywordAnalyzer.analyze(text)  ──→ {reaction_name: [matched_words]}
      │
      ├─ EntityAnalyzer.analyze(text)   ──→ [EntityMatch(text, label, confidence)]
      │
      ├─ Deduplication (per-turn)
      │
      └─ Dispatch → reaction.callback(deps, match, **params)

The handler calls the manager at three points:

_on_transcript_partial(text) — on each streaming ASR partial (debounced at 400ms)
_on_transcript_final(text) — when the final transcript is ready (fire-and-forget, parallel with LLM)
_on_turn_complete() — resets all deduplication state for the next turn

Reaction Configuration (`reactions.yaml`)

Each profile can define a reactions.yaml file. The loader (loader.py) reads it and imports callbacks from the profile's Python modules.

# Simple keyword trigger — fires if any word matches
- name: music_excitement
  callback: excited_about_music
  trigger:
    words: [music, guitar, piano, drum, violin]

# Entity trigger with repeatable — fires once per unique entity
- name: food_reaction
  callback: react_to_food_entity
  trigger:
    entities: [food]
  repeatable: true

# Boolean AND trigger — all sub-groups must match
- name: groovy_dance
  callback: do_groovy_dance
  trigger:
    all:
      - words: [danc*]
      - words: [groov*]

# With extra params passed as kwargs to the callback
- name: turn_left
  callback: turn_to_direction
  trigger:
    all:
      - words: [turn*]
      - words: [left]
  params:
    direction: left

Fields:

Field	Required	Description
`name`	yes	Unique reaction identifier
`callback`	yes	Python module name in the profile directory (module must contain a function with the same name)
`trigger`	yes	What activates the reaction (see Trigger Types below)
`params`	no	Extra kwargs passed to the callback
`repeatable`	no	If `true`, can fire multiple times per turn (default: `false`)

Trigger Types

`words` — Keyword matching

The word list supports three forms, mixable in a single list:

Form	Example	Matching strategy
Plain word	`guitar`	Substring match on the full transcript text. Matches "guitar", "guitars", "guitarist".
Glob pattern	`music*`	`fnmatch` against individual whitespace-split tokens. Matches "music", "musical", "musician". Supports `*` and `?`.
Multi-word phrase	`"grand piano"`	Substring match (use YAML quotes). Matches any occurrence of "grand piano" in the text.

Any one match in the list is enough to trigger the reaction (OR logic).

`entities` — Named Entity Recognition

Uses GLiNER (optional dependency) to detect entities by semantic label (e.g. food, person, location). The model is configurable in cascade.yaml:

transcript_analysis:
  gliner_model: "urchade/gliner_small-v2.1"

Entity analysis runs in a thread executor since GLiNER inference is CPU-bound.

`all` — Boolean AND

A list of sub-triggers that must all match for the reaction to fire. Internally, each sub-trigger is registered as a synthetic keyword entry (reaction_name__all_0, reaction_name__all_1, etc.) and merged back after analysis.

Deduplication

By default, each reaction fires at most once per turn (per conversation exchange). The manager tracks fired reactions in triggered_reactions: set[str] and skips duplicates.

Repeatable reactions (repeatable: true) bypass this gate. For entity-triggered repeatable reactions, deduplication is per unique entity text: "pizza" triggers once, "zucchini" triggers once, but "pizza" again is skipped. This is tracked via _triggered_entity_keys: set[tuple[str, str]].

All deduplication state resets when _on_turn_complete() is called.

Callback Signature

Every callback must be an async function with this signature:

async def my_callback(
    deps: ToolDependencies,
    match: TriggerMatch,
    **kwargs,    # receives params from YAML
) -> None:

TriggerMatch contains what matched:

match.words: list[str] — matched keywords (for word triggers)
match.entities: list[EntityMatch] — matched entities, each with .text, .label, .confidence

Callbacks are imported from profiles/<profile_name>/<callback_name>.py — the module must export a function with the same name as the module.

Data Classes (`base.py`)

@dataclass
class ReactionConfig:
    name: str
    callback: Callable[..., Awaitable[None]]
    trigger: TriggerConfig
    params: dict[str, Any] = field(default_factory=dict)
    repeatable: bool = False

@dataclass
class TriggerConfig:
    words: list[str] = field(default_factory=list)
    entities: list[str] = field(default_factory=list)
    all: list[TriggerConfig] = field(default_factory=list)

@dataclass
class TriggerMatch:
    words: list[str] = field(default_factory=list)
    entities: list[EntityMatch] = field(default_factory=list)

@dataclass
class EntityMatch:
    text: str          # e.g. "pizza"
    label: str         # e.g. "food"
    confidence: float  # 0.0 – 1.0

Design Decisions

Why TTS is driven by a SpeechOutput protocol, not hardcoded:

speak tool calls speech_output.speak() inside the pipeline — TTS happens as part of tool execution
Each stream manager injects its own implementation:
- Gradio: GradioSpeechOutput — parallel sentence synthesis, pre-warmed playback threads
- Console/Test: ConsoleSpeechOutput — rate-limited streaming with playback callback
Handler and pipeline stay decoupled from audio I/O details

Why conversation history lives in Handler:

Handler needs full context for LLM generation
Tool results must be added for multi-turn reasoning
Camera tool adds images to conversation for analysis
UI reads history for display only

Troubleshooting & Known Issues

MLX Thread Affinity (Apple Silicon)

Note: MLX-based providers (Parakeet) call MLX synchronously inside async methods, intentionally blocking the event loop for ~10-50ms per chunk. Do not wrap these calls in asyncio.to_thread() — MLX has thread affinity requirements and will produce empty transcriptions if inference runs in a different thread than model loading. The blocking is negligible at this scale. If a future model makes inference significantly slower (>100ms), consider a dedicated single-thread executor instead.

OpenAI Realtime ASR - "Streaming" Misconception

Problem: Partial transcripts don't appear during speech - they all arrive at once after speech ends.

Root Cause: OpenAI's "streaming transcription" means text streams out quickly after audio is committed, NOT that you get real-time partials while speaking.

How it actually works:

Audio chunks sent → buffer on server
Server VAD detects silence → commits buffer
THEN transcription starts and deltas stream rapidly (~200ms for full text)

Additional issue - Connection latency:

WebSocket connection takes ~800-1000ms to establish after speech starts
Audio recorded during this time overflows and is lost
Server VAD may not track speech properly due to this discontinuity

Configuration options (cascade.yaml):

openai_realtime_asr:
  use_server_vad: true   # Real-time partials after each silence detection
  use_server_vad: false  # All transcription at end (manual commit)

With use_server_vad: true:

Server VAD (500ms silence) and local Silero VAD (700ms) can coexist
Server streams partials after each detected pause
Local VAD controls when end_stream() is called

Comparison with Deepgram:

Deepgram also streams partials after audio is processed, not during speech
Both exhibit similar "batch of partials at end" behavior
True real-time mid-speech partials would require periodic audio commits (causing fragmented transcripts)

Potential fix (not implemented): Pre-warm WebSocket connection before speech starts, keeping it in standby mode to eliminate connection latency.

Test File Mode (`autotest_stream.py`)

Automated end-to-end testing of the cascade pipeline without human interaction. Reads text utterances from a file, synthesizes them to audio via TTS, and feeds the audio through the full pipeline (TTS→ASR→LLM→TTS→robot).

Usage

# Uses cascade pipeline (default)
lyon-chatbox --test-file scripts/test_utterances.txt --no-camera

Test File Format

Plain text, one utterance per line. # comments and blank lines are ignored:

# Greetings
Hello, what is your name?

# Movement commands
Can you look to the left?

Architecture

CascadeTestStream is a stream manager (like CascadeLocalStream or CascadeGradioUI) that replaces mic/VAD input with synthetic audio.

Text file → Input TTS (af_heart voice) → PCM audio
                                            │
                    ┌───────────────────────┤
                    │                       │
                    ▼                       ▼
            Speaker playback        ASR (streaming or batch)
            (sounddevice callback)          │
                                            ▼
                                    LLM → Tool calls
                                            │
                                            ▼
                                    Output TTS (robot voice)
                                            │
                                            ▼
                                    Speaker playback
                                    (same sounddevice stream)

Key components:

Input TTS — Separate TTSProvider instance using a distinct voice (af_heart) so user and robot are distinguishable
Callback-based playback — sd.OutputStream in callback mode pulls from a bytearray buffer at a steady rate. Both user and robot audio share this buffer, avoiding the choppy playback that blocking write() calls cause in an async context
Two processing paths:
- Streaming ASR (_process_streaming): resamples PCM to 16kHz, feeds 32ms chunks at real-time pace via process_audio_streaming_start/chunk/end. This triggers partial transcripts and transcript analysis reactions during "speech", matching the live VAD flow
- Batch ASR (_process_manual): plays user audio, then sends full WAV via process_audio_manual

What This Tests

Full ASR→LLM→TTS pipeline with real provider calls
Tool calling (speak, movements, emotions)
Transcript analysis / live reactions (streaming path only)
Head wobbler animation synchronized with response audio
Movement manager integration
Cost tracking across turns
Latency tracking (uses vad_speech_end reference for correct summary)
Conversation history accumulation across turns

What This Does NOT Test

Since CascadeTestStream is its own stream manager, it bypasses several components that the other modes use:

VAD speech detection — Utterances are pre-defined text, not detected from audio. Silero VAD is never invoked
Microphone input / robot.media recording — No start_recording() or get_audio_sample() calls. Audio comes from TTS, not hardware
Gradio UI — No web interface, no push-to-talk buttons, no chatbot display, no AudioPlaybackSystem
Robot speaker output — Response audio plays through computer speakers (sounddevice), not robot.media.push_audio_sample()
ContinuousVADRecorder — This recording class is not used
Barge-in / interruption handling — Utterances are sequential with fixed delays; there is no overlap between user speech and robot response
Audio resampling for robot hardware — The _play_loop resampling path in CascadeLocalStream (TTS rate → robot output rate) is not exercised
Camera / vision pipeline — Typically run with --no-camera; describe_camera_image tool calls would fail without it

Latency Tracking

The tracker is reset after input TTS playback finishes (not when utterance generation starts), so the summary reflects only the pipeline latency. It uses vad_speech_end as the reference event so print_summary() recognizes the VAD flow and shows all stages:

1. Recording Capture
2. ASR Processing
3. LLM Generation
4. TTS time to first audio
   ↳ Audio system delay
TOTAL PERCEIVED LATENCY: Speech End → First Audio

Future Extensions

Streaming TTS integration
Multi-language support with per-provider language codes
Visual/audio cues for tool execution feedback
Conversation reset without app restart
Multi-modal prompts with vision in system prompt

Cascade Mode - Developer Documentation

Overview

Key Design Principles

Current Scope

File Structure

Core Components

CascadeHandler (handler.py)

Provider Factory (provider_factory.py)

Pipeline (pipeline.py)

TurnResult (turn_result.py)

UI Components (ui/)

AudioPlaybackSystem (ui/audio_playback.py)

Recording Classes (ui/audio_recording.py)

CascadeGradioUI (ui/gradio_app.py)

SpeechOutput (speech_output.py)

VAD State Machine (vad.py)

Console Mode (console.py)

Provider Abstractions

ASR Providers (asr/)

LLM Providers (llm/)

TTS Providers (tts/)

Data Flow

VAD-Triggered Pipeline

Streaming ASR Flow

Configuration

Config File: cascade.yaml

Hardware Tags

Dependency Checks

Environment Variables

Config Loading (config.py)

Audio Playback System

Architecture

Playback Backends

Device Detection

Module Dependencies

Entry Point & Integration

Entry Function (entry.py)

Command Line Usage

Latency Tracking (timing.py)

LatencyTracker

Transcript Analysis (Live Reactions)

Architecture

Reaction Configuration (reactions.yaml)

Trigger Types

words — Keyword matching

entities — Named Entity Recognition

all — Boolean AND

Deduplication

Callback Signature

Data Classes (base.py)

Design Decisions

Troubleshooting & Known Issues

MLX Thread Affinity (Apple Silicon)

OpenAI Realtime ASR - "Streaming" Misconception

Test File Mode (autotest_stream.py)

Usage

Test File Format

Architecture

What This Tests

What This Does NOT Test

Latency Tracking

Future Extensions

CascadeHandler (`handler.py`)

Provider Factory (`provider_factory.py`)

Pipeline (`pipeline.py`)

TurnResult (`turn_result.py`)

UI Components (`ui/`)

AudioPlaybackSystem (`ui/audio_playback.py`)

Recording Classes (`ui/audio_recording.py`)

CascadeGradioUI (`ui/gradio_app.py`)

SpeechOutput (`speech_output.py`)

VAD State Machine (`vad.py`)

Console Mode (`console.py`)

ASR Providers (`asr/`)

LLM Providers (`llm/`)

TTS Providers (`tts/`)

Config File: `cascade.yaml`

Config Loading (`config.py`)

Entry Function (`entry.py`)

Latency Tracking (`timing.py`)

Reaction Configuration (`reactions.yaml`)

`words` — Keyword matching

`entities` — Named Entity Recognition

`all` — Boolean AND

Data Classes (`base.py`)

Test File Mode (`autotest_stream.py`)