diff --git a/scripts/analyze_chops.py b/scripts/analyze_chops.py new file mode 100644 index 00000000..66e7eaac --- /dev/null +++ b/scripts/analyze_chops.py @@ -0,0 +1,367 @@ +"""Detect choppy-audio artifacts (dropped/spliced chunks) in EVA recordings. + +A "chop" is a missing or spliced audio chunk. Because EVA streams fixed-size +chunks (20ms = 160 µ-law bytes, or 250ms), a dropped chunk leaves a specific, +machine-detectable fingerprint: a run of digital-zero samples that severs +continuous speech, with a HARD CUT on BOTH edges — full-amplitude speech +jumping straight to zero and back. A natural pause is the opposite: speech +tapers to ~0 before the silence, so its edge amplitude is tiny. Requiring both +edges to be hard cuts is what separates chops from pauses (validated against +labeled recordings: 0 false positives on a clean file, all real chops caught). +Chops are bucketed "short" (< 150ms, ~one dropped chunk) vs "long" (>= 150ms). + +Two modes: + + characterize + Given a WAV and labeled chop timestamps, print the discriminating + features inside each labeled window vs. the file's baseline, so we can + SEE what separates chops from normal speech/pauses and lock thresholds. + + scan + Run the calibrated detector over every audio_mixed.wav under a run + directory and emit a CSV of candidate chops sorted by confidence. + +Label file format (JSON) for characterize: + { + "audio_file": "output//records//audio_mixed.wav", + "chops": [ + {"t": 12.34}, # a point in the chop + {"start": 30.10, "end": 30.35} # or an explicit span (seconds) + ] + } + +Usage: + python scripts/analyze_chops.py characterize labels1.json labels2.json + python scripts/analyze_chops.py scan output/ --out chops.csv +""" + +from __future__ import annotations + +import argparse +import csv +import json +import sys +from dataclasses import dataclass, field +from pathlib import Path + +import numpy as np +import soundfile as sf + +# --- Framing / feature parameters ------------------------------------------- +FRAME_MS = 10.0 # analysis frame length (baseline stats only) +HOP_MS = 5.0 # analysis hop (baseline stats only) + +# --- Detector thresholds (calibrated against labeled EVA recordings) --------- +# A chop is a dropped/spliced chunk: a run of digital-zero samples that severs +# continuous speech, leaving a HARD CUT on BOTH edges (full-amplitude speech +# jumping straight to zero and back). A natural pause differs on exactly this +# point — speech tapers to ~0 before the silence, so its "cut" amplitude is +# tiny. Requiring both edges to be hard is what separates chops from pauses +# (validated: 0 false positives on a clean recording, catches all real chops). +SILENCE_EPS = 1e-4 # |sample| below this counts as digital silence (float32 -1..1) +MIN_ZERO_MS = 20.0 # ignore sub-chunk blips (one 20ms chunk is the smallest drop) +EDGE_WIN_MS = 5.0 # window just outside the silence used to measure cut amplitude +EDGE_STEP_MIN = 0.05 # BOTH edges' amplitude must exceed this (hard splice) +LONG_CHOP_MS = 150.0 # >= this is a "long" chop; below is "short" + + +@dataclass +class Chop: + start: float + end: float + kind: str # "gap" | "click" + confidence: float + detail: dict = field(default_factory=dict) + + @property + def duration(self) -> float: + return self.end - self.start + + +def load_mono(path: Path) -> tuple[np.ndarray, int]: + """Load a WAV as float32 mono in [-1, 1]. Stereo is averaged across channels.""" + data, sr = sf.read(str(path), dtype="float32", always_2d=True) + mono = data.mean(axis=1) + return mono, sr + + +def longest_zero_run_ms(audio: np.ndarray, sr: int, lo: int, hi: int) -> float: + """Longest contiguous digital-silence run (in ms) within sample range [lo, hi).""" + seg = np.abs(audio[lo:hi]) < SILENCE_EPS + if not seg.any(): + return 0.0 + best = run = 0 + for v in seg: + run = run + 1 if v else 0 + best = max(best, run) + return best * 1000.0 / sr + + +def detect_chops(audio: np.ndarray, sr: int) -> list[Chop]: + """Find dropped/spliced chunks: digital-zero runs with a hard cut on both edges. + + A chop severs continuous speech, so the sample just before the silence and + the sample just after are both at speech amplitude (a splice). Natural + pauses taper to ~0 before the silence, so at least one edge is tiny — those + are rejected by requiring MIN of the two edge amplitudes to exceed the + threshold. + """ + silent = np.abs(audio) < SILENCE_EPS + w = max(1, int(sr * EDGE_WIN_MS / 1000.0)) + min_run = int(sr * MIN_ZERO_MS / 1000.0) + chops: list[Chop] = [] + n = len(audio) + i = 0 + while i < n: + if not silent[i]: + i += 1 + continue + j = i + while j < n and silent[j]: + j += 1 + if (j - i) >= min_run: + pre = float(np.max(np.abs(audio[max(0, i - w) : i]))) if i > 0 else 0.0 + post = float(np.max(np.abs(audio[j : j + w]))) if j < n else 0.0 + edge = min(pre, post) + if edge >= EDGE_STEP_MIN: + dur_ms = (j - i) * 1000.0 / sr + chops.append( + Chop( + start=i / sr, + end=j / sr, + kind="long" if dur_ms >= LONG_CHOP_MS else "short", + confidence=round(min(1.0, edge / 0.25), 3), + detail={"dur_ms": round(dur_ms, 1), "pre": round(pre, 3), "post": round(post, 3)}, + ) + ) + i = j + return chops + + +def detect_all(path: Path) -> list[Chop]: + audio, sr = load_mono(path) + return sorted(detect_chops(audio, sr), key=lambda c: c.start) + + +# --------------------------------------------------------------------------- +# characterize +# --------------------------------------------------------------------------- +def _label_spans(chops: list[dict]) -> list[tuple[float, float]]: + spans = [] + for c in chops: + if "start" in c and "end" in c: + spans.append((float(c["start"]), float(c["end"]))) + else: + t = float(c["t"]) + spans.append((t - 0.4, t + 0.4)) # ±400ms window (labels may be slightly misaligned) + return spans + + +def _window_stats(audio: np.ndarray, sr: int, a: float, b: float) -> tuple[float, float, float]: + """Return (min frame RMS, longest zero-run ms, max sample-diff) over [a, b] seconds.""" + lo, hi = max(0, int(a * sr)), min(len(audio), int(b * sr)) + seg = audio[lo:hi] + if seg.size == 0: + return 0.0, 0.0, 0.0 + fr = max(1, int(sr * FRAME_MS / 1000)) + hp = max(1, int(sr * HOP_MS / 1000)) + rmss = [np.sqrt(np.mean(seg[k : k + fr] ** 2)) for k in range(0, max(1, seg.size - fr), hp)] + min_rms = float(min(rmss)) if rmss else 0.0 + max_diff = float(np.max(np.abs(np.diff(seg)))) if seg.size > 1 else 0.0 + return min_rms, longest_zero_run_ms(audio, sr, lo, hi), max_diff + + +def characterize(label_files: list[Path]) -> None: + print(f"{'file/window':<38} {'min_rms':>9} {'zero_ms':>9} {'max_diff':>9} {'detected':>18}") + print("-" * 90) + for lf in label_files: + spec = json.loads(lf.read_text()) + wav = Path(spec["audio_file"]) + if not wav.is_absolute(): + wav = (lf.parent / wav).resolve() + audio, sr = load_mono(wav) + detected = detect_all(wav) + + print(f"[{wav.name}] labeled chops:") + for a, b in _label_spans(spec.get("chops", [])): + mr, zr, md = _window_stats(audio, sr, a, b) + near = [d for d in detected if d.start <= b + 0.1 and d.end >= a - 0.1] + tag = ",".join(f"{d.kind}({d.confidence})" for d in near) or "MISS" + print(f" {a:6.2f}-{b:6.2f}s{'':<20} {mr:9.4f} {zr:9.1f} {md:9.4f} {tag:>18}") + + # Baseline: sample random speech windows NOT overlapping any label. + spans = _label_spans(spec.get("chops", [])) + rng = np.random.default_rng(0) + dur = len(audio) / sr + shown = 0 + for _ in range(200): + a = float(rng.uniform(0, max(0.001, dur - 0.3))) + b = a + 0.3 + if any(a < s2 and b > s1 for s1, s2 in spans): + continue + mr, zr, md = _window_stats(audio, sr, a, b) + if mr > 0.02: # only report active-speech baselines (min RMS above the noise floor) + print(f" baseline {a:6.2f}-{b:6.2f}s{'':<11} {mr:9.4f} {zr:9.1f} {md:9.4f} {'':>18}") + shown += 1 + if shown >= 5: + break + print() + + +# --------------------------------------------------------------------------- +# scan +# --------------------------------------------------------------------------- +def scan(run_dir: Path, out_csv: Path, min_conf: float) -> None: + wavs = sorted(run_dir.rglob("audio_mixed.wav")) + if not wavs: + print(f"No audio_mixed.wav found under {run_dir}", file=sys.stderr) + return + rows = [] + for wav in wavs: + # e.g. records///audio_mixed.wav -> "/" + record = f"{wav.parent.parent.name}/{wav.parent.name}" + try: + chops = [c for c in detect_all(wav) if c.confidence >= min_conf] + except Exception as e: # noqa: BLE001 + print(f" ERROR {wav}: {e}", file=sys.stderr) + continue + for c in chops: + rows.append( + { + "record": record, + "file": str(wav), + "start_s": round(c.start, 3), + "end_s": round(c.end, 3), + "duration_ms": round(c.duration * 1000, 1), + "kind": c.kind, + "confidence": c.confidence, + "detail": json.dumps(c.detail), + } + ) + print(f"{record}: {len(chops)} candidate chop(s)") + rows.sort(key=lambda r: r["confidence"], reverse=True) + with out_csv.open("w", newline="") as f: + w = csv.DictWriter(f, fieldnames=list(rows[0].keys()) if rows else ["record"]) + w.writeheader() + w.writerows(rows) + print(f"\n{len(rows)} candidate chops across {len(wavs)} files -> {out_csv}") + + +def detect_overlap( + user_path: Path, asst_path: Path, speech_rms: float = 0.02, hop_s: float = 0.02, min_run_s: float = 0.20 +) -> dict: + """Measure cross-channel overlap (both user and assistant speaking at once). + + Overlap is present in normal conversation (interruptions), so the useful + signal for a *drift/misalignment* bug is whether overlap grows late in the + conversation vs early — a drift accumulates over turns, natural overlap does + not. Returns totals plus an early(first 40%)/late(last 40%) split. + """ + u, sr = load_mono(user_path) + a, _ = load_mono(asst_path) + n = min(len(u), len(a)) + u, a = u[:n], a[:n] + h = max(1, int(sr * hop_s)) + nf = n // h + both = np.array( + [ + (np.sqrt(np.mean(u[i * h : (i + 1) * h] ** 2)) > speech_rms) + and (np.sqrt(np.mean(a[i * h : (i + 1) * h] ** 2)) > speech_rms) + for i in range(nf) + ] + ) + + def runs_secs(mask: np.ndarray) -> float: + total = 0.0 + i = 0 + m = len(mask) + while i < m: + if mask[i]: + j = i + while j < m and mask[j]: + j += 1 + if (j - i) * hop_s >= min_run_s: + total += (j - i) * hop_s + i = j + else: + i += 1 + return total + + q = max(1, nf // 5) + return { + "total_s": round(runs_secs(both), 2), + "early_s": round(runs_secs(both[: 2 * q]), 2), + "late_s": round(runs_secs(both[3 * q :]), 2), + } + + +def scan_overlap(run_dir: Path, out_csv: Path) -> None: + """Report per-record chops (user channel) and user/assistant overlap.""" + users = sorted(run_dir.rglob("audio_user.wav")) + rows = [] + for u in users: + a = u.parent / "audio_assistant.wav" + if not a.exists(): + continue + record = f"{u.parent.parent.name}/{u.parent.name}" + try: + chops = detect_all(u) + ov = detect_overlap(u, a) + except Exception as e: # noqa: BLE001 + print(f" ERROR {u}: {e}", file=sys.stderr) + continue + # Separate audible LONG chops (>=150ms) from benign 20ms single-frame + # micro-gaps, which are a pre-existing baseline artifact and otherwise + # dominate (and distort) the raw count. + chops_long = sum(c.kind == "long" for c in chops) + chops_short = sum(c.kind == "short" for c in chops) + rows.append( + { + "record": record, + "chops_long": chops_long, + "chops_short": chops_short, + **{f"overlap_{k}": v for k, v in ov.items()}, + } + ) + with out_csv.open("w", newline="") as f: + w = csv.DictWriter(f, fieldnames=list(rows[0].keys()) if rows else ["record"]) + w.writeheader() + w.writerows(rows) + tcl = sum(r["chops_long"] for r in rows) + tcs = sum(r["chops_short"] for r in rows) + to = sum(r["overlap_total_s"] for r in rows) + te = sum(r["overlap_early_s"] for r in rows) + tl = sum(r["overlap_late_s"] for r in rows) + print( + f"{len(rows)} records: LONG chops={tcl} (audible) short/20ms={tcs} (benign) " + f"overlap total={to:.1f}s (early={te:.1f}s late={tl:.1f}s) -> {out_csv}" + ) + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + sub = ap.add_subparsers(dest="cmd", required=True) + + c = sub.add_parser("characterize", help="Inspect features in labeled chop windows vs baseline") + c.add_argument("labels", nargs="+", type=Path, help="Label JSON file(s)") + + s = sub.add_parser("scan", help="Detect chops across all audio_mixed.wav in a run dir") + s.add_argument("run_dir", type=Path) + s.add_argument("--out", type=Path, default=Path("chops.csv")) + s.add_argument("--min-conf", type=float, default=0.0) + + o = sub.add_parser("overlap", help="Report chops + user/assistant overlap (early/late) per record") + o.add_argument("run_dir", type=Path) + o.add_argument("--out", type=Path, default=Path("overlap.csv")) + + args = ap.parse_args() + if args.cmd == "characterize": + characterize(args.labels) + elif args.cmd == "scan": + scan(args.run_dir, args.out, args.min_conf) + elif args.cmd == "overlap": + scan_overlap(args.run_dir, args.out) + + +if __name__ == "__main__": + main() diff --git a/scripts/run_text_only.py b/scripts/run_text_only.py index f08d439e..ff6a5290 100644 --- a/scripts/run_text_only.py +++ b/scripts/run_text_only.py @@ -49,6 +49,7 @@ from eva.models.record import EvaluationRecord from eva.models.results import ConversationResult, MetricScore, RecordMetrics from eva.utils import router +from eva.utils.bootstrap import run_seed from eva.utils.culture import resolve_scenario_db, resolve_user_goal from eva.utils.hash_utils import get_dict_hash from eva.utils.log_processing import ( @@ -948,7 +949,9 @@ async def _run_one(oid: str) -> dict[str, Any]: if all_record_metrics: metric_names = requested_metrics + ["text_response_latency"] - per_metric = MetricsRunner._build_per_metric_aggregates(all_record_metrics, metric_names, num_draws=num_trials) + per_metric = MetricsRunner._build_per_metric_aggregates( + all_record_metrics, metric_names, num_draws=num_trials, seed=run_seed(output_dir.name) + ) # Augment text_response_latency with raw latency stats in seconds if "text_response_latency" in per_metric: @@ -996,7 +999,9 @@ async def _run_one(oid: str) -> dict[str, Any]: sum(all_call_latencies) / len(all_call_latencies), 3 ) - overall_scores = compute_run_level_aggregates(all_record_metrics, num_trials, text_composites) + overall_scores = compute_run_level_aggregates( + all_record_metrics, num_trials, text_composites, run_seed(output_dir.name) + ) data_quality = MetricsRunner._build_data_quality(all_record_metrics, per_metric) metrics_summary_data: dict[str, Any] = { diff --git a/src/eva/assistant/audio_buffer.py b/src/eva/assistant/audio_buffer.py new file mode 100644 index 00000000..e6b28486 --- /dev/null +++ b/src/eva/assistant/audio_buffer.py @@ -0,0 +1,62 @@ +"""Audio buffer processor tuned for EVA's continuous bot-to-bot streams. + +pipecat 1.x's ``AudioBufferProcessor`` inserts wall-clock silence into a +recording buffer whenever a frame is *processed* more than 200ms after the +previous one (``_fill_buffer_silence_gap``). That behavior assumes a gap in +processing time means a real silent period (a muted microphone, an idle pause). + +That assumption does not hold for EVA. The user simulator sends a *continuous* +20ms stream — real caller audio plus its own silence frames — so the audio is +never actually gapped. When the assistant's event loop stalls under high +concurrency, user frames simply arrive late and bunched; pipecat then sees a +>200ms wall-clock gap and fabricates silence *between* frames whose audio is +all present. The result is a hard-cut silence run in the middle of speech — the +"chop" heard in ``audio_user.wav`` — even though the conversation (and the +real-time STT stream) was unaffected. This regressed when pipecat went +0.0.104 -> 1.x. + +We keep the byte-count track synchronization (``_sync_buffer_to_position``, +which aligns the user and bot channels and predates 1.x) but disable the +wall-clock silence fabrication, restoring contiguous recording of the incoming +stream. +""" + +from __future__ import annotations + +from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor + +from eva.utils.logging import get_logger + +logger = get_logger(__name__) + + +class ContiguousAudioBufferProcessor(AudioBufferProcessor): + """AudioBufferProcessor that records frames contiguously. + + Disables pipecat's wall-clock ``_fill_buffer_silence_gap`` because EVA's + input stream is already continuous; any processing-time gap is event-loop + jitter, not a real silent period, so filling it fabricates the choppy + silence artifact. Track alignment still happens via ``_sync_buffer_to_position``. + + Emits a log line on init and periodically reports how much silence it + suppressed, so a run can confirm the override was actually active. + """ + + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self._suppressed_silence_bytes = 0 + logger.info("ContiguousAudioBufferProcessor active — pipecat wall-clock silence fill disabled") + + def _fill_buffer_silence_gap(self, buffer, last_update_time, now, frame_bytes) -> None: + # No-op: do not fabricate silence from processing-time gaps. Track what + # would have been inserted so the effect is measurable in the logs. + if last_update_time is None or self._sample_rate == 0: + return + gap = (now - last_update_time) - frame_bytes / (self._sample_rate * 2) + if gap > 0.2: + would_insert = int(gap * self._sample_rate * 2) + self._suppressed_silence_bytes += would_insert + logger.debug( + f"Suppressed {gap * 1000:.0f}ms wall-clock silence fill " + f"(cumulative {self._suppressed_silence_bytes / (self._sample_rate * 2):.1f}s)" + ) diff --git a/src/eva/assistant/elevenlabs_server.py b/src/eva/assistant/elevenlabs_server.py index eb5b6493..0a543440 100644 --- a/src/eva/assistant/elevenlabs_server.py +++ b/src/eva/assistant/elevenlabs_server.py @@ -246,6 +246,15 @@ async def _handle_session(self, websocket: WebSocket) -> None: # noqa: C901 # channel. Both tracks pad silence relative to this so the mixed output # keeps real wall-clock timing (set lazily; see _pad_buffer_to_walltime). _record_t0: float | None = None + # Pad the USER track to wall-clock only at the start of a user turn, then + # append contiguously within the turn (mirrors the assistant pacer's + # turn_start handling). Padding on EVERY frame — the previous behavior — + # injected silence whenever frames were processed late under concurrency, + # fabricating mid-utterance "chops" in audio_user.wav even though the + # audio and the conversation were fine. True inter-turn gaps (e.g. the + # assistant's speaking period, when the user sim sends no frames) are + # still filled by the one pad at the next user-turn start. + _pad_user_on_turn_start: bool = True # Queue of (mulaw_chunk, pcm16k_chunk, turn_start) items; the pacer drains # at real-time rate, sends the mulaw and records the PCM at playback time. @@ -355,7 +364,7 @@ async def _forward_user_audio() -> None: """Read Twilio WS messages, convert audio, send to ElevenLabs.""" nonlocal stream_sid, twilio_connected nonlocal _user_speech_start_ts, _user_speech_stop_ts - nonlocal _user_speaking, _in_model_turn, _record_t0 + nonlocal _user_speaking, _in_model_turn, _record_t0, _pad_user_on_turn_start try: while twilio_connected and self._running: try: @@ -380,6 +389,11 @@ async def _forward_user_audio() -> None: _user_speech_start_ts = msg.get("timestamp_ms") _user_speaking = True _in_model_turn = False + # New user turn: allow one wall-clock pad to place this + # utterance on the real timeline (fills the assistant's + # speaking gap). Cleared after the first frame so the + # rest of the turn records contiguously. + _pad_user_on_turn_start = True logger.info(f"User speech start: {_user_speech_start_ts}") elif event == "user_speech_stop": _user_speech_stop_ts = msg.get("timestamp_ms") @@ -395,11 +409,17 @@ async def _forward_user_audio() -> None: now = time.monotonic() if _record_t0 is None: _record_t0 = now - _pad_buffer_to_walltime( - self.user_audio_buffer, - now - _record_t0, - self._audio_sample_rate, - ) + # Pad to wall-clock only at the start of a user turn; + # within the turn append contiguously so processing-time + # jitter under concurrency cannot inject mid-utterance + # silence (the choppy-audio bug). See _pad_user_on_turn_start. + if _pad_user_on_turn_start: + _pad_buffer_to_walltime( + self.user_audio_buffer, + now - _record_t0, + self._audio_sample_rate, + ) + _pad_user_on_turn_start = False self.user_audio_buffer.extend(pcm_16k) # Feed raw 8 kHz mulaw to ElevenLabs — the agent diff --git a/src/eva/assistant/gemini_live_server.py b/src/eva/assistant/gemini_live_server.py index b07fff28..de69e32c 100644 --- a/src/eva/assistant/gemini_live_server.py +++ b/src/eva/assistant/gemini_live_server.py @@ -50,6 +50,12 @@ # so the user simulator's silence detection works correctly. MULAW_CHUNK_SIZE = 160 # bytes per chunk (20ms at 8kHz, 1 byte per sample) MULAW_CHUNK_DURATION_S = 0.02 # 20ms per chunk +# Don't pad the user track to align with the assistant when real user audio +# arrived within this window. The speaking-state flag can go stale under +# event-loop jitter, and padding then injects silence into an active user +# utterance (the choppy-audio bug). This guard only ever *skips* a pad, so it +# can never add silence or worsen alignment. +USER_ACTIVE_GUARD_S = 0.3 # --------------------------------------------------------------------------- @@ -176,6 +182,9 @@ def __init__( # Recording sample rate (Gemini outputs 24 kHz) self._audio_sample_rate = _RECORDING_SAMPLE_RATE + # Monotonic time of the last real user-audio frame appended; guards + # against padding the user track mid-utterance (see USER_ACTIVE_GUARD_S). + self._last_user_audio_mono: float = 0.0 # Gemini model name from s2s_params or default s2s_params = self.pipeline_config.s2s_params or {} @@ -399,6 +408,7 @@ async def _forward_user_audio() -> None: if not _in_model_turn: sync_buffer_to_position(self.assistant_audio_buffer, len(self.user_audio_buffer)) self.user_audio_buffer.extend(pcm_24k) + self._last_user_audio_mono = time.monotonic() # Send to Gemini await session.send_realtime_input( @@ -509,7 +519,13 @@ async def _process_gemini_events() -> None: if len(pcm_24k) < 6: continue - if not _user_speaking: + # Skip the pad if the user track is actively receiving + # audio (flag may be stale under jitter) — padding then + # would inject a mid-utterance chop. + user_recently_active = ( + time.monotonic() - self._last_user_audio_mono + ) <= USER_ACTIVE_GUARD_S + if not _user_speaking and not user_recently_active: sync_buffer_to_position( self.user_audio_buffer, len(self.assistant_audio_buffer) ) diff --git a/src/eva/assistant/openai_realtime_server.py b/src/eva/assistant/openai_realtime_server.py index 3d368900..27661537 100644 --- a/src/eva/assistant/openai_realtime_server.py +++ b/src/eva/assistant/openai_realtime_server.py @@ -38,6 +38,13 @@ # so the user simulator's silence detection works correctly. MULAW_CHUNK_SIZE = 160 # bytes per chunk (20ms at 8kHz, 1 byte per sample) MULAW_CHUNK_DURATION_S = 0.02 # 20ms per chunk +# Don't pad the user track to align with the assistant when real user audio +# arrived within this window. The speaking-state flag can go stale under +# event-loop jitter, and padding then injects silence into an active user +# utterance (the choppy-audio bug). This guard only ever *skips* a pad, so it +# can never add silence or worsen alignment — during genuine user silence the +# timestamp is old and padding proceeds normally. +USER_ACTIVE_GUARD_S = 0.3 def _wall_ms() -> str: @@ -82,6 +89,9 @@ def __init__(self, **kwargs: Any) -> None: super().__init__(**kwargs) self._audio_sample_rate = OPENAI_SAMPLE_RATE + # Monotonic time of the last real user-audio frame appended; used to + # guard against padding the user track mid-utterance (see USER_ACTIVE_GUARD_S). + self._last_user_audio_mono: float = 0.0 self._system_prompt: str = self._build_system_prompt() @@ -376,6 +386,7 @@ async def _forward_user_audio(self, websocket: WebSocket, conn: Any) -> None: sync_buffer_to_position(self.assistant_audio_buffer, sync_target) synced = len(self.assistant_audio_buffer) - asst_before self.user_audio_buffer.extend(pcm16_24k) + self._last_user_audio_mono = time.monotonic() self._user_frame_count += 1 if self._user_frame_count % 50 == 0: diff = len(self.user_audio_buffer) - len(self.assistant_audio_buffer) @@ -579,7 +590,10 @@ async def _on_audio_delta(self, event: Any, audio_output_queue: asyncio.Queue[by user_before = len(self.user_audio_buffer) synced = 0 - if not self._user_speaking: + # Skip the pad if the user track is actively receiving audio (flag may be + # stale under jitter) — padding then would inject a mid-utterance chop. + user_recently_active = (time.monotonic() - self._last_user_audio_mono) <= USER_ACTIVE_GUARD_S + if not self._user_speaking and not user_recently_active: sync_buffer_to_position(self.user_audio_buffer, len(self.assistant_audio_buffer)) synced = len(self.user_audio_buffer) - user_before self.assistant_audio_buffer.extend(pcm16_bytes) diff --git a/src/eva/assistant/pipecat_server.py b/src/eva/assistant/pipecat_server.py index e747a163..c758f62c 100644 --- a/src/eva/assistant/pipecat_server.py +++ b/src/eva/assistant/pipecat_server.py @@ -28,7 +28,6 @@ LLMUserAggregatorParams, UserTurnStoppedMessage, ) -from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor from pipecat.serializers.twilio import TwilioFrameSerializer from pipecat.services.cartesia.turns.stt import CartesiaTurnsSTTService from pipecat.transports.websocket.fastapi import ( @@ -40,6 +39,7 @@ from pipecat.utils.time import time_now_iso8601 from eva.assistant.agentic.audit_log import convert_to_epoch_ms, current_timestamp_ms +from eva.assistant.audio_buffer import ContiguousAudioBufferProcessor from eva.assistant.base_server import AbstractAssistantServer from eva.assistant.pipeline.agent_processor import BenchmarkAgentProcessor, UserAudioCollector, UserObserver from eva.assistant.pipeline.audio_llm_processor import ( @@ -402,7 +402,10 @@ async def on_user_transcription(text: str, timestamp: str, turn_id: int | None) # Create processors # Configure audio buffer with 1-second buffer size for event triggering - audiobuffer = AudioBufferProcessor( + # ContiguousAudioBufferProcessor disables pipecat 1.x's wall-clock + # silence fabrication, which under concurrency injects choppy silence + # into the recorded user track (see eva/assistant/audio_buffer.py). + audiobuffer = ContiguousAudioBufferProcessor( sample_rate=SAMPLE_RATE, num_channels=1, # Mono (mixed user + bot audio) buffer_size=SAMPLE_RATE * 2, # 1 second of 16-bit audio (2 bytes per sample)