first
This commit is contained in:
valenti
@@ -0,0 +1,147 @@
|
||||
//! Backchannel: brief acknowledgments emitted *while the user is speaking*.
|
||||
//!
|
||||
//! This is distinct from a filler (see [`TurnManager`](crate::orchestrator::TurnManager)),
|
||||
//! which is played once the user stops, at end-of-utterance. A backchannel is a
|
||||
//! short "mm-hmm" / "ajá" mixed in at a reduced gain to signal that the agent is
|
||||
//! listening, without taking the turn. It is emitted only while the agent itself
|
||||
//! is silent (otherwise the user is being interrupted) and never more than
|
||||
//! `max_per_turn` times.
|
||||
//!
|
||||
//! Every behavioural knob is configured from the environment
|
||||
//! ([`BackchannelSettings`]), so the feature can be tuned or disabled without a
|
||||
//! rebuild.
|
||||
|
||||
use crate::config::{env_bool, env_f32, env_u32, env_u64};
|
||||
use crate::orchestrator::FillerBank;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
/// Runtime configuration for the backchannel subsystem, read from `.env`.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct BackchannelSettings {
|
||||
/// Master switch (`BACKCHANNEL_ENABLED`).
|
||||
pub enabled: bool,
|
||||
/// Directory of `*.raw` PCM clips (`BACKCHANNEL_DIR`).
|
||||
pub dir: PathBuf,
|
||||
/// Mix gain applied to backchannel clips (`BACKCHANNEL_GAIN`). Kept low so
|
||||
/// the cue stays in the background.
|
||||
pub gain: f32,
|
||||
/// Continuous speech required before the first backchannel
|
||||
/// (`BACKCHANNEL_MIN_SPEECH_MS`).
|
||||
pub min_speech_ms: u64,
|
||||
/// Minimum spacing between two backchannels (`BACKCHANNEL_INTERVAL_MS`).
|
||||
pub interval_ms: u64,
|
||||
/// Maximum backchannels per user turn (`BACKCHANNEL_MAX_PER_TURN`).
|
||||
pub max_per_turn: u32,
|
||||
}
|
||||
|
||||
impl Default for BackchannelSettings {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
enabled: true,
|
||||
dir: PathBuf::from("assets/backchannel"),
|
||||
gain: 0.35,
|
||||
min_speech_ms: 1_500,
|
||||
interval_ms: 3_000,
|
||||
max_per_turn: 2,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl BackchannelSettings {
|
||||
/// Build the settings from environment variables, falling back to defaults.
|
||||
pub fn from_env() -> Self {
|
||||
let mut s = Self::default();
|
||||
if let Ok(p) = std::env::var("BACKCHANNEL_DIR") {
|
||||
if !p.is_empty() {
|
||||
s.dir = PathBuf::from(p);
|
||||
}
|
||||
}
|
||||
s.enabled = env_bool("BACKCHANNEL_ENABLED", s.enabled);
|
||||
s.gain = env_f32("BACKCHANNEL_GAIN", s.gain);
|
||||
s.min_speech_ms = env_u64("BACKCHANNEL_MIN_SPEECH_MS", s.min_speech_ms);
|
||||
s.interval_ms = env_u64("BACKCHANNEL_INTERVAL_MS", s.interval_ms);
|
||||
s.max_per_turn = env_u32("BACKCHANNEL_MAX_PER_TURN", s.max_per_turn);
|
||||
|
||||
s.gain = s.gain.clamp(0.0, 1.0);
|
||||
// Never cue before the turn has meaningfully started.
|
||||
s.interval_ms = s.interval_ms.max(s.min_speech_ms);
|
||||
s
|
||||
}
|
||||
}
|
||||
|
||||
/// Per-session backchannel decision logic.
|
||||
///
|
||||
/// Fed by the VAD: [`on_speech_start`](Self::on_speech_start) when a user turn
|
||||
/// begins, [`on_speech_end`](Self::on_speech_end) when it closes, and
|
||||
/// [`maybe_emit`](Self::maybe_emit) on every frame while the user is speaking.
|
||||
pub struct BackchannelScheduler {
|
||||
bank: Arc<FillerBank>,
|
||||
settings: BackchannelSettings,
|
||||
/// Round-robin cursor over the clip bank.
|
||||
next_index: usize,
|
||||
/// When the current user turn started (only set while speaking).
|
||||
turn_start: Option<Instant>,
|
||||
/// When the last backchannel was emitted in this turn.
|
||||
last_emitted: Option<Instant>,
|
||||
/// Backchannels emitted in the current turn.
|
||||
emitted: u32,
|
||||
}
|
||||
|
||||
impl BackchannelScheduler {
|
||||
pub fn new(bank: Arc<FillerBank>, settings: BackchannelSettings) -> Self {
|
||||
Self {
|
||||
bank,
|
||||
settings,
|
||||
next_index: 0,
|
||||
turn_start: None,
|
||||
last_emitted: None,
|
||||
emitted: 0,
|
||||
}
|
||||
}
|
||||
|
||||
/// Begin a new user turn (VAD speech start).
|
||||
pub fn on_speech_start(&mut self) {
|
||||
self.turn_start = Some(Instant::now());
|
||||
self.last_emitted = None;
|
||||
self.emitted = 0;
|
||||
}
|
||||
|
||||
/// Close the current user turn (VAD speech end).
|
||||
pub fn on_speech_end(&mut self) {
|
||||
self.turn_start = None;
|
||||
self.last_emitted = None;
|
||||
self.emitted = 0;
|
||||
}
|
||||
|
||||
/// Emit a backchannel into `mixer` if the policy allows it right now.
|
||||
///
|
||||
/// Must be called while the user is speaking and the agent is silent.
|
||||
/// Returns the clip name when one was emitted.
|
||||
pub fn maybe_emit(&mut self, mixer: &crate::audio::AudioMixer) -> Option<String> {
|
||||
if !self.settings.enabled {
|
||||
return None;
|
||||
}
|
||||
let start = self.turn_start?;
|
||||
if self.emitted >= self.settings.max_per_turn {
|
||||
return None;
|
||||
}
|
||||
if start.elapsed() < Duration::from_millis(self.settings.min_speech_ms) {
|
||||
return None;
|
||||
}
|
||||
if let Some(last) = self.last_emitted {
|
||||
if last.elapsed() < Duration::from_millis(self.settings.interval_ms) {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
|
||||
let (name, samples) = self.bank.nth(self.next_index)?;
|
||||
let name = name.to_string();
|
||||
mixer.backchannel_queue.push(samples);
|
||||
self.next_index = self.next_index.wrapping_add(1);
|
||||
self.last_emitted = Some(Instant::now());
|
||||
self.emitted += 1;
|
||||
Some(name)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
//! Turn orchestration: EOU handling, filler injection, TTS dispatch, barge-in.
|
||||
|
||||
pub mod backchannel;
|
||||
pub mod turn_manager;
|
||||
|
||||
pub use backchannel::{BackchannelScheduler, BackchannelSettings};
|
||||
pub use turn_manager::{spawn_orchestrator, FillerBank, TurnManager};
|
||||
@@ -0,0 +1,722 @@
|
||||
//! Turn orchestration: reacts to EOU events from the STT worker, schedules the
|
||||
//! latency filler, runs the reasoning LLM, fires the TTS request, and handles
|
||||
//! barge-in flush.
|
||||
//!
|
||||
//! Two pipeline shapes are supported:
|
||||
//! - **With LLM** (`LLM_ENABLED=true`): transcript → LLM → sentence chunks →
|
||||
//! TTS. The assistant text is also forwarded to the client so a chat box can
|
||||
//! display it.
|
||||
//! - **Without LLM**: transcript → TTS (the original pass-through behaviour).
|
||||
|
||||
use crate::audio::{AudioMixer, SampleRingBuffer};
|
||||
use crate::config::Config;
|
||||
use crate::llm::{ChatMessage, LlmClient};
|
||||
use crate::stt::{Tokenizer, Transcript};
|
||||
use crate::tts::TtsEngine;
|
||||
use serde_json::json;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
/// Capacity (samples) of the per-request staging buffer used by the TTS
|
||||
/// pipeline — 30 s at the mixer sample rate. Large enough to hold a whole
|
||||
/// chunk even when a later request finishes synthesizing while an earlier one
|
||||
/// is still playing out.
|
||||
const TTS_SLOT_CAPACITY_SAMPLES: usize = crate::config::OUTPUT_SAMPLE_RATE as usize * 30;
|
||||
|
||||
/// One in-flight synthesis request in the TTS pipeline.
|
||||
struct TtsSlot {
|
||||
/// Staging buffer the request streams its PCM into. The relay forwards it
|
||||
/// to the mixer queue in submission order.
|
||||
buffer: SampleRingBuffer,
|
||||
/// Task running the synthesis; finished means the chunk is fully staged.
|
||||
handle: tokio::task::JoinHandle<()>,
|
||||
}
|
||||
|
||||
/// Move every sample currently staged in `src` into `dst`, respecting `dst`'s
|
||||
/// free space so already-queued (i.e. currently playing) audio is never
|
||||
/// dropped. Returns the number of samples moved.
|
||||
fn pump(src: &SampleRingBuffer, dst: &SampleRingBuffer) -> usize {
|
||||
let mut tmp = [0i16; 8192];
|
||||
let mut moved = 0;
|
||||
loop {
|
||||
let want = dst
|
||||
.remaining_capacity()
|
||||
.min(src.len())
|
||||
.min(tmp.len());
|
||||
if want == 0 {
|
||||
break;
|
||||
}
|
||||
let n = src.pop_into(&mut tmp[..want]);
|
||||
if n == 0 {
|
||||
break;
|
||||
}
|
||||
dst.push(&tmp[..n]);
|
||||
moved += n;
|
||||
}
|
||||
moved
|
||||
}
|
||||
|
||||
/// Preloaded PCM samples keyed by name (e.g. "uhm"). Used for the filler bank
|
||||
/// injected at end-of-utterance (played round-robin) and, from a different
|
||||
/// directory, for the backchannel clips played while the user speaks.
|
||||
pub struct FillerBank {
|
||||
samples: HashMap<String, Vec<i16>>,
|
||||
/// Names in a stable order, for round-robin selection.
|
||||
names: Vec<String>,
|
||||
}
|
||||
|
||||
impl FillerBank {
|
||||
/// Load every `*.raw` file in `dir` as raw 16-bit LE mono PCM. `label` is
|
||||
/// only used in log messages (e.g. "filler", "backchannel").
|
||||
pub fn load(dir: &std::path::Path, label: &str) -> Self {
|
||||
let mut samples = HashMap::new();
|
||||
if let Ok(entries) = std::fs::read_dir(dir) {
|
||||
for entry in entries.flatten() {
|
||||
let path = entry.path();
|
||||
if path.extension().and_then(|e| e.to_str()) != Some("raw") {
|
||||
continue;
|
||||
}
|
||||
let name = path
|
||||
.file_stem()
|
||||
.and_then(|n| n.to_str())
|
||||
.unwrap_or("unknown")
|
||||
.to_string();
|
||||
match std::fs::read(&path) {
|
||||
Ok(bytes) => {
|
||||
let pcm = bytes
|
||||
.chunks_exact(2)
|
||||
.map(|c| i16::from_le_bytes([c[0], c[1]]))
|
||||
.collect::<Vec<i16>>();
|
||||
info!(label, name, samples = pcm.len(), "loaded {label} clip");
|
||||
samples.insert(name, pcm);
|
||||
}
|
||||
Err(e) => warn!(label, path = %path.display(), error = %e, "failed to read {label} clip"),
|
||||
}
|
||||
}
|
||||
if samples.is_empty() {
|
||||
warn!(label, dir = %dir.display(), "no {label} clips found");
|
||||
}
|
||||
} else {
|
||||
warn!(label, dir = %dir.display(), "{label} directory not found");
|
||||
}
|
||||
|
||||
let mut names: Vec<String> = samples.keys().cloned().collect();
|
||||
names.sort();
|
||||
Self { samples, names }
|
||||
}
|
||||
|
||||
/// Get the samples for a clip by name. Kept for reference/debugging.
|
||||
#[allow(dead_code)]
|
||||
pub fn get(&self, name: &str) -> Option<&Vec<i16>> {
|
||||
self.samples.get(name)
|
||||
}
|
||||
|
||||
/// Clip at position `index` (wrapping), with its name. Used for the
|
||||
/// round-robin filler and backchannel selection.
|
||||
pub fn nth(&self, index: usize) -> Option<(&str, &Vec<i16>)> {
|
||||
if self.names.is_empty() {
|
||||
return None;
|
||||
}
|
||||
let name = &self.names[index % self.names.len()];
|
||||
self.samples.get(name).map(|s| (name.as_str(), s))
|
||||
}
|
||||
|
||||
/// Whether the bank holds no clips.
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.names.is_empty()
|
||||
}
|
||||
}
|
||||
|
||||
/// Orchestrates a single conversation turn.
|
||||
pub struct TurnManager {
|
||||
mixer: AudioMixer,
|
||||
tts: Arc<dyn TtsEngine>,
|
||||
fillers: Arc<FillerBank>,
|
||||
/// Optional tokenizer to decode STT token IDs into text. If `None`, a
|
||||
/// placeholder string is sent to the TTS (useful for testing the pipeline
|
||||
/// without a real tokenizer).
|
||||
tokenizer: Option<Tokenizer>,
|
||||
/// Cancellation token for the currently running TTS request.
|
||||
cancel: Arc<parking_lot::Mutex<Option<CancellationToken>>>,
|
||||
/// Round-robin cursor over the filler bank (end-of-utterance fillers).
|
||||
filler_cursor: AtomicUsize,
|
||||
/// Reasoning LLM client. `None` => direct STT → TTS pipeline.
|
||||
llm: Option<Arc<LlmClient>>,
|
||||
/// Server configuration (LLM prompt / streaming / history window).
|
||||
config: Arc<Config>,
|
||||
/// Rolling conversation history (user + assistant turns).
|
||||
history: parking_lot::Mutex<Vec<ChatMessage>>,
|
||||
/// Outbound text-event channel (JSON strings) consumed by the session
|
||||
/// sender task and forwarded to the browser (chat box).
|
||||
out_tx: Option<mpsc::UnboundedSender<String>>,
|
||||
}
|
||||
|
||||
impl TurnManager {
|
||||
pub fn new(
|
||||
mixer: AudioMixer,
|
||||
tts: Arc<dyn TtsEngine>,
|
||||
fillers: Arc<FillerBank>,
|
||||
tokenizer: Option<Tokenizer>,
|
||||
llm: Option<Arc<LlmClient>>,
|
||||
config: Arc<Config>,
|
||||
out_tx: Option<mpsc::UnboundedSender<String>>,
|
||||
) -> Self {
|
||||
Self {
|
||||
mixer,
|
||||
tts,
|
||||
fillers,
|
||||
tokenizer,
|
||||
cancel: Arc::new(parking_lot::Mutex::new(None)),
|
||||
filler_cursor: AtomicUsize::new(0),
|
||||
llm,
|
||||
config,
|
||||
history: parking_lot::Mutex::new(Vec::new()),
|
||||
out_tx,
|
||||
}
|
||||
}
|
||||
|
||||
/// Pick the next filler clip round-robin from the bank.
|
||||
fn next_filler(&self) -> Option<(String, Vec<i16>)> {
|
||||
let i = self.filler_cursor.fetch_add(1, Ordering::Relaxed);
|
||||
self.fillers.nth(i).map(|(name, s)| (name.to_string(), s.clone()))
|
||||
}
|
||||
|
||||
/// Schedule the single end-of-turn filler acknowledgment.
|
||||
///
|
||||
/// The filler is a latency mask: it buys time while the reply is being
|
||||
/// generated, so it is only useful when the response audio is *late*. It is
|
||||
/// therefore not played unconditionally — if the reply starts playing
|
||||
/// within `FILLER_DELAY_MS`, the filler is dropped instead of speaking over
|
||||
/// the beginning of the answer. At most one clip is emitted per turn and a
|
||||
/// barge-in (`cancel`) drops the pending one. Disabled entirely by
|
||||
/// `FILLER_ENABLED=false`.
|
||||
fn schedule_filler(&self, cancel: &CancellationToken) {
|
||||
if !self.config.filler_enabled {
|
||||
return;
|
||||
}
|
||||
let Some((name, samples)) = self.next_filler() else {
|
||||
debug!("filler enabled but the bank is empty — skipping");
|
||||
return;
|
||||
};
|
||||
|
||||
let delay_ms = self.config.filler_delay_ms;
|
||||
let mixer = self.mixer.clone();
|
||||
let cancel = cancel.clone();
|
||||
let samples_len = samples.len();
|
||||
tokio::spawn(async move {
|
||||
tokio::select! {
|
||||
_ = tokio::time::sleep(Duration::from_millis(delay_ms)) => {}
|
||||
// Barge-in / new turn: never play a stale acknowledgment.
|
||||
_ = cancel.cancelled() => return,
|
||||
}
|
||||
if mixer.tts_playback_started() {
|
||||
debug!(
|
||||
filler = %name,
|
||||
delay_ms, "response audio already playing — filler skipped"
|
||||
);
|
||||
return;
|
||||
}
|
||||
mixer.filler_queue.push(&samples);
|
||||
debug!(filler = %name, samples = samples_len, delay_ms, "pushed filler (delayed)");
|
||||
});
|
||||
}
|
||||
|
||||
/// Send a JSON event to the client, if a channel is attached.
|
||||
fn emit(&self, value: serde_json::Value) {
|
||||
if let Some(tx) = &self.out_tx {
|
||||
let _ = tx.send(value.to_string());
|
||||
}
|
||||
}
|
||||
|
||||
/// React to an EOU / transcript event from the STT worker.
|
||||
pub async fn on_eou(&self, transcript: &Transcript) {
|
||||
info!(
|
||||
tokens = transcript.token_ids.len(),
|
||||
duration_ms = transcript.duration_ms,
|
||||
text = %transcript.text,
|
||||
"EOU detected"
|
||||
);
|
||||
|
||||
// The Parakeet pipeline decodes text internally; fall back to the
|
||||
// external tokenizer only if no text was produced.
|
||||
let text = if !transcript.text.trim().is_empty() {
|
||||
transcript.text.trim().to_string()
|
||||
} else {
|
||||
match &self.tokenizer {
|
||||
Some(tok) => tok.decode(&transcript.token_ids),
|
||||
None => format!("<transcript {} tokens>", transcript.token_ids.len()),
|
||||
}
|
||||
};
|
||||
if text.is_empty() {
|
||||
debug!("empty transcript after decode, skipping turn");
|
||||
return;
|
||||
}
|
||||
self.handle_user_turn(text).await;
|
||||
}
|
||||
|
||||
/// Handle a user turn typed in the chat box (no STT involved).
|
||||
pub async fn on_user_text(&self, text: &str) {
|
||||
let text = text.trim();
|
||||
if text.is_empty() {
|
||||
return;
|
||||
}
|
||||
self.handle_user_turn(text.to_string()).await;
|
||||
}
|
||||
|
||||
/// Shared entry point for every user turn, whether spoken or typed.
|
||||
///
|
||||
/// 1. Forward the recognized turn to the client chat box.
|
||||
/// 2. Cancel any in-flight reply (barge-in within the turn).
|
||||
/// 3. Run the LLM (if configured) or send the text straight to the TTS.
|
||||
///
|
||||
/// The filler is not queued here: it is scheduled once the response is
|
||||
/// dispatched, so it only plays if the audio is late (see
|
||||
/// [`Self::schedule_filler`]).
|
||||
async fn handle_user_turn(&self, text: String) {
|
||||
self.emit(json!({ "type": "user", "text": text }));
|
||||
|
||||
self.cancel_current_tts();
|
||||
|
||||
match &self.llm {
|
||||
Some(llm) => self.run_llm_turn(llm.clone(), text).await,
|
||||
None => self.dispatch_tts(&text),
|
||||
}
|
||||
}
|
||||
|
||||
/// Run one full LLM turn: stream the reply, chunk it into sentences and
|
||||
/// synthesize each sentence as soon as it is complete.
|
||||
async fn run_llm_turn(&self, llm: Arc<LlmClient>, user_text: String) {
|
||||
// Build the message list from the rolling history.
|
||||
let messages = {
|
||||
let mut hist = self.history.lock();
|
||||
hist.push(ChatMessage::user(&user_text));
|
||||
let max = self.config.llm_history_messages.max(2);
|
||||
if hist.len() > max {
|
||||
let drop = hist.len() - max;
|
||||
hist.drain(..drop);
|
||||
}
|
||||
let mut msgs = Vec::with_capacity(hist.len() + 1);
|
||||
let system = self.config.llm_system_prompt.trim();
|
||||
if !system.is_empty() {
|
||||
msgs.push(ChatMessage::system(system));
|
||||
}
|
||||
msgs.extend(hist.iter().cloned());
|
||||
msgs
|
||||
};
|
||||
|
||||
let cancel = CancellationToken::new();
|
||||
*self.cancel.lock() = Some(cancel.clone());
|
||||
self.mixer.mark_stream_start();
|
||||
self.schedule_filler(&cancel);
|
||||
debug!(
|
||||
endpoint = llm.endpoint(),
|
||||
model = llm.model(),
|
||||
messages = messages.len(),
|
||||
stream = self.config.llm_stream,
|
||||
"dispatching LLM turn"
|
||||
);
|
||||
|
||||
// TTS sink: a *pipelined* synthesizer. Each chunk is requested as soon
|
||||
// as the LLM produces it instead of waiting for the previous chunk's
|
||||
// stream to end, so the per-request start-up latency overlaps the
|
||||
// previous chunk's playback. Requests write into their own staging
|
||||
// buffer; a relay forwards those buffers to the mixer queue strictly in
|
||||
// submission order, which keeps playback in order and gapless.
|
||||
let (sentence_tx, mut sentence_rx) = mpsc::unbounded_channel::<String>();
|
||||
let tts = self.tts.clone();
|
||||
let tts_queue = self.mixer.tts_queue.clone();
|
||||
let cancel_tts = cancel.clone();
|
||||
let mixer_for_tts = self.mixer.clone();
|
||||
let pipeline_depth = self.config.tts_pipeline_depth;
|
||||
let tts_task = tokio::spawn(async move {
|
||||
let (slot_tx, mut slot_rx) = mpsc::unbounded_channel::<TtsSlot>();
|
||||
let permits = Arc::new(tokio::sync::Semaphore::new(pipeline_depth));
|
||||
|
||||
// Collector: start one synthesis task per chunk immediately, so its
|
||||
// start-up latency runs concurrently with the chunks already
|
||||
// playing. `permits` caps how many requests are in flight at once.
|
||||
let collector = {
|
||||
let tts = tts.clone();
|
||||
let cancel = cancel_tts.clone();
|
||||
let permits = permits.clone();
|
||||
tokio::spawn(async move {
|
||||
while let Some(sentence) = sentence_rx.recv().await {
|
||||
if cancel.is_cancelled() {
|
||||
break;
|
||||
}
|
||||
let buffer = SampleRingBuffer::new(TTS_SLOT_CAPACITY_SAMPLES);
|
||||
let buf_task = buffer.clone();
|
||||
let permits = permits.clone();
|
||||
let cancel_task = cancel.clone();
|
||||
let tts_task_engine = tts.clone();
|
||||
let handle = tokio::spawn(async move {
|
||||
let _permit = tokio::select! {
|
||||
p = permits.acquire_owned() => match p {
|
||||
Ok(p) => p,
|
||||
Err(_) => return,
|
||||
},
|
||||
_ = cancel_task.cancelled() => return,
|
||||
};
|
||||
if cancel_task.is_cancelled() {
|
||||
return;
|
||||
}
|
||||
if let Err(e) =
|
||||
tts_task_engine.synthesize(&sentence, &buf_task, &cancel_task).await
|
||||
{
|
||||
error!(error = %e, "TTS synthesis failed (pipelined)");
|
||||
}
|
||||
});
|
||||
if slot_tx.send(TtsSlot { buffer, handle }).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
})
|
||||
};
|
||||
|
||||
// Relay: forward each chunk's staged audio to the mixer queue in
|
||||
// order, waiting for the current chunk to finish before advancing.
|
||||
while let Some(slot) = slot_rx.recv().await {
|
||||
let started = std::time::Instant::now();
|
||||
debug!(
|
||||
staged = slot.buffer.len(),
|
||||
finished = slot.handle.is_finished(),
|
||||
"TTS pipeline: relaying chunk to mixer"
|
||||
);
|
||||
let mut relayed = 0usize;
|
||||
loop {
|
||||
relayed += pump(&slot.buffer, &tts_queue);
|
||||
if cancel_tts.is_cancelled() {
|
||||
break;
|
||||
}
|
||||
if slot.handle.is_finished() && slot.buffer.is_empty() {
|
||||
debug!(
|
||||
relayed,
|
||||
elapsed_ms = started.elapsed().as_millis(),
|
||||
queue_len = tts_queue.len(),
|
||||
"TTS pipeline chunk fully relayed to the mixer"
|
||||
);
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(2)).await;
|
||||
}
|
||||
if cancel_tts.is_cancelled() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
let _ = collector.await;
|
||||
if !cancel_tts.is_cancelled() {
|
||||
mixer_for_tts.mark_stream_end();
|
||||
}
|
||||
});
|
||||
|
||||
// LLM producer: streams clean text deltas into a channel.
|
||||
let (delta_tx, mut delta_rx) = mpsc::unbounded_channel::<String>();
|
||||
let stream_mode = self.config.llm_stream;
|
||||
let llm_cancel = cancel.clone();
|
||||
let llm_task = tokio::spawn(async move {
|
||||
if stream_mode {
|
||||
llm.chat_stream(&messages, &llm_cancel, delta_tx).await
|
||||
} else {
|
||||
let res = llm.chat(&messages, &llm_cancel).await;
|
||||
if let Ok(text) = &res {
|
||||
let _ = delta_tx.send(text.clone());
|
||||
}
|
||||
res
|
||||
}
|
||||
});
|
||||
|
||||
// Consume deltas: emit `assistant_delta` events and feed the splitter,
|
||||
// so the TTS starts as soon as the first few words are available.
|
||||
let mut splitter = SentenceSplitter::new(
|
||||
self.config.llm_stream_first_chunk_chars,
|
||||
self.config.llm_stream_chunk_chars,
|
||||
);
|
||||
let mut full = String::new();
|
||||
while let Some(delta) = delta_rx.recv().await {
|
||||
if cancel.is_cancelled() {
|
||||
break;
|
||||
}
|
||||
self.emit(json!({ "type": "assistant_delta", "text": delta }));
|
||||
full.push_str(&delta);
|
||||
for sentence in splitter.push(&delta) {
|
||||
if sentence_tx.send(sentence).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
if !cancel.is_cancelled() {
|
||||
if let Some(rest) = splitter.finish() {
|
||||
let _ = sentence_tx.send(rest);
|
||||
}
|
||||
}
|
||||
drop(sentence_tx);
|
||||
let _ = tts_task.await;
|
||||
|
||||
match llm_task.await {
|
||||
Ok(Ok(_)) => {}
|
||||
Ok(Err(e)) => {
|
||||
error!(error = %e, "LLM request failed");
|
||||
self.emit(json!({ "type": "error", "text": e.to_string() }));
|
||||
}
|
||||
Err(e) => {
|
||||
error!(error = %e, "LLM task join failed");
|
||||
self.emit(json!({ "type": "error", "text": format!("LLM task failed: {e}") }));
|
||||
}
|
||||
}
|
||||
|
||||
if !cancel.is_cancelled() && !full.trim().is_empty() {
|
||||
self.history.lock().push(ChatMessage::assistant(full.clone()));
|
||||
self.emit(json!({ "type": "assistant_end" }));
|
||||
}
|
||||
|
||||
let mut guard = self.cancel.lock();
|
||||
if guard.as_ref().map(|c| c.is_cancelled()).unwrap_or(false) {
|
||||
*guard = None;
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle an explicit client-side interrupt (barge-in).
|
||||
/// Flushes all queued audio and cancels the in-flight LLM/TTS request.
|
||||
pub fn interrupt(&self) {
|
||||
info!("interrupt received, flushing audio");
|
||||
self.cancel_current_tts();
|
||||
self.mixer.flush();
|
||||
}
|
||||
|
||||
/// Whether the agent is currently speaking (TTS in flight or queued audio).
|
||||
/// Used by the VAD-driven barge-in to avoid flushing an idle mixer.
|
||||
pub fn is_speaking(&self) -> bool {
|
||||
self.mixer.is_active()
|
||||
}
|
||||
|
||||
/// Update the voice-cloning reference at runtime. Pass `None` for
|
||||
/// `audio_path` to clear the reference.
|
||||
pub fn set_reference(&self, audio_path: Option<String>, text: Option<String>) {
|
||||
self.tts.set_reference(audio_path, text);
|
||||
}
|
||||
|
||||
/// Synthesize `text` directly with the TTS engine, bypassing the LLM.
|
||||
/// Used by the playground's "Speak" box to test the TTS pipeline.
|
||||
pub async fn speak(&self, text: &str) {
|
||||
let text = text.trim();
|
||||
if text.is_empty() {
|
||||
return;
|
||||
}
|
||||
info!(text_len = text.len(), "speak request (direct TTS)");
|
||||
|
||||
// Cancel any in-flight TTS and clear the mixer queues so the new
|
||||
// utterance starts cleanly (no overlap with the previous one).
|
||||
self.cancel_current_tts();
|
||||
self.mixer.flush();
|
||||
|
||||
self.dispatch_tts(text);
|
||||
}
|
||||
|
||||
/// Fire one TTS request for `text` into the mixer's TTS queue.
|
||||
fn dispatch_tts(&self, text: &str) {
|
||||
let cancel = CancellationToken::new();
|
||||
*self.cancel.lock() = Some(cancel.clone());
|
||||
|
||||
let tts = self.tts.clone();
|
||||
let queue = self.mixer.tts_queue.clone();
|
||||
let cancel_holder = self.cancel.clone();
|
||||
let mixer = self.mixer.clone();
|
||||
let text_owned = text.to_string();
|
||||
self.mixer.mark_stream_start();
|
||||
self.schedule_filler(&cancel);
|
||||
debug!(
|
||||
engine = tts.name(),
|
||||
sample_rate = tts.sample_rate(),
|
||||
text_len = text.len(),
|
||||
"dispatching TTS synthesis"
|
||||
);
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = tts.synthesize(&text_owned, &queue, &cancel).await {
|
||||
error!(error = %e, "TTS synthesis failed");
|
||||
}
|
||||
// Signal completion (unless cancelled by a barge-in) so the mixer
|
||||
// drains short utterances instead of waiting for the pre-buffer.
|
||||
if !cancel.is_cancelled() {
|
||||
mixer.mark_stream_end();
|
||||
debug!(queue_len = queue.len(), "TTS stream end signalled to mixer");
|
||||
}
|
||||
let mut guard = cancel_holder.lock();
|
||||
if guard.as_ref().map(|c| c.is_cancelled()).unwrap_or(false) {
|
||||
*guard = None;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
fn cancel_current_tts(&self) {
|
||||
if let Some(token) = self.cancel.lock().take() {
|
||||
token.cancel();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn an orchestrator loop that consumes transcripts from `rx` and drives
|
||||
/// the turn manager.
|
||||
pub fn spawn_orchestrator(
|
||||
manager: Arc<TurnManager>,
|
||||
mut rx: mpsc::UnboundedReceiver<Transcript>,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
while let Some(transcript) = rx.recv().await {
|
||||
manager.on_eou(&transcript).await;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Splits a streaming reply into speech-sized chunks so the TTS can start
|
||||
/// speaking while the model is still generating the rest.
|
||||
///
|
||||
/// A chunk always ends on `.`, `!`, `?`, `;` or a newline. On top of that, a
|
||||
/// word-boundary flush keeps latency bounded when the model runs on without
|
||||
/// punctuation: the *first* chunk is cut once [`first_chunk_chars`] characters
|
||||
/// are buffered (kept small, so the agent starts on the first few words), and
|
||||
/// every following chunk at [`chunk_chars`] (larger, for better prosody). A
|
||||
/// threshold of `0` disables that flush.
|
||||
///
|
||||
/// [`first_chunk_chars`]: SentenceSplitter::first_chunk_chars
|
||||
/// [`chunk_chars`]: SentenceSplitter::chunk_chars
|
||||
struct SentenceSplitter {
|
||||
buf: String,
|
||||
/// Bytes buffered before the first chunk is flushed on a word boundary.
|
||||
first_chunk_chars: usize,
|
||||
/// Same, for every chunk after the first.
|
||||
chunk_chars: usize,
|
||||
/// How many chunks were handed out so far (drives the threshold above).
|
||||
emitted: usize,
|
||||
}
|
||||
|
||||
impl SentenceSplitter {
|
||||
fn new(first_chunk_chars: usize, chunk_chars: usize) -> Self {
|
||||
Self {
|
||||
buf: String::new(),
|
||||
first_chunk_chars,
|
||||
chunk_chars,
|
||||
emitted: 0,
|
||||
}
|
||||
}
|
||||
|
||||
/// Feed a delta and return every complete chunk it produced.
|
||||
fn push(&mut self, delta: &str) -> Vec<String> {
|
||||
self.buf.push_str(delta);
|
||||
let mut out = Vec::new();
|
||||
|
||||
let mut start = 0usize;
|
||||
for (i, b) in self.buf.bytes().enumerate() {
|
||||
// ASCII terminators only: safe to slice at `i + 1`.
|
||||
if matches!(b, b'.' | b'!' | b'?' | b';' | b'\n') {
|
||||
let segment = self.buf[start..=i].trim();
|
||||
if !segment.is_empty() {
|
||||
out.push(segment.to_string());
|
||||
self.emitted += 1;
|
||||
}
|
||||
start = i + 1;
|
||||
}
|
||||
}
|
||||
if start > 0 {
|
||||
self.buf.drain(..start);
|
||||
}
|
||||
|
||||
// Word-boundary flush: cut at the first whitespace at or past the
|
||||
// threshold, so the chunk holds at least that many bytes.
|
||||
let threshold = if self.emitted == 0 {
|
||||
self.first_chunk_chars
|
||||
} else {
|
||||
self.chunk_chars
|
||||
};
|
||||
if threshold > 0 && self.buf.len() >= threshold {
|
||||
let safe = floor_char_boundary(&self.buf, threshold);
|
||||
if let Some(rel) = self.buf[safe..].find(' ') {
|
||||
let end = safe + rel;
|
||||
let segment = self.buf[..end].trim().to_string();
|
||||
if !segment.is_empty() {
|
||||
out.push(segment);
|
||||
self.emitted += 1;
|
||||
}
|
||||
self.buf.drain(..end);
|
||||
}
|
||||
}
|
||||
|
||||
out
|
||||
}
|
||||
|
||||
/// Flush whatever is left once the stream has ended.
|
||||
fn finish(&mut self) -> Option<String> {
|
||||
let rest = self.buf.trim().to_string();
|
||||
self.buf.clear();
|
||||
if rest.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(rest)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Largest char boundary `<= i`.
|
||||
fn floor_char_boundary(s: &str, mut i: usize) -> usize {
|
||||
if i >= s.len() {
|
||||
return s.len();
|
||||
}
|
||||
while i > 0 && !s.is_char_boundary(i) {
|
||||
i -= 1;
|
||||
}
|
||||
i
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn splits_on_sentence_terminators() {
|
||||
let mut s = SentenceSplitter::new(20, 120);
|
||||
assert_eq!(
|
||||
s.push("Hola. ¿Cómo estás? Bien"),
|
||||
vec!["Hola.".to_string(), "¿Cómo estás?".to_string()]
|
||||
);
|
||||
assert_eq!(s.finish(), Some("Bien".to_string()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn splits_across_deltas() {
|
||||
let mut s = SentenceSplitter::new(20, 120);
|
||||
assert!(s.push("Hola").is_empty());
|
||||
assert_eq!(s.push(" mundo!"), vec!["Hola mundo!".to_string()]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn first_chunk_flushes_early_on_word_boundary() {
|
||||
let mut s = SentenceSplitter::new(20, 120);
|
||||
let out = s.push("Ciao come stai oggi tutto bene");
|
||||
assert_eq!(out.len(), 1);
|
||||
assert!(out[0].len() >= 20, "chunk too short: {:?}", out[0]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn later_chunks_use_the_larger_threshold() {
|
||||
let mut s = SentenceSplitter::new(20, 120);
|
||||
// First chunk is small and punctuation-free.
|
||||
let first = s.push("Ciao come stai oggi tutto bene");
|
||||
assert_eq!(first.len(), 1);
|
||||
// The next run must reach the larger threshold before flushing.
|
||||
let second = s.push(&"palabra ".repeat(20)); // 160 bytes
|
||||
assert_eq!(second.len(), 1);
|
||||
assert!(second[0].len() >= 120, "chunk too short: {}", second[0].len());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn zero_threshold_disables_word_boundary_flush() {
|
||||
let mut s = SentenceSplitter::new(0, 0);
|
||||
assert!(s.push(&"palabra ".repeat(40)).is_empty());
|
||||
assert!(!s.finish().unwrap_or_default().is_empty());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user