"""Causal, audio-only Pipecat replay and non-mutating annotation-point probes. The real Pipecat VAD and turn analyzer run on a media clock. Frames are 32 ms. At annotated times an additional diagnostic model query does not alter replay. No labels, transcripts or future audio are given to either neural model. """ import argparse import csv import hashlib import importlib.metadata import inspect import json import time from collections import Counter from pathlib import Path from types import SimpleNamespace import numpy as np import soundfile as sf from loguru import logger import pipecat.audio.turn.smart_turn.base_smart_turn as base_module import pipecat.audio.vad.silero as silero_module from pipecat.audio.turn.base_turn_analyzer import EndOfTurnState from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams, VADState HERE = Path(__file__).resolve().parent ROOT = HERE.parents[1] def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def read_csv(path): with path.open(encoding="utf-8-sig", newline="") as handle: return list(csv.DictReader(handle)) def write_csv(path, rows): if not rows: return with path.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) def metrics(rows, prediction): selected = [r for r in rows if r["input_valid"] and r[prediction] in (0, 1) and r["can_reply_now"] in ("0", "1")] counts = Counter((int(r["can_reply_now"]), r[prediction]) for r in selected) tn, fp, fn, tp = (counts[(0, 0)], counts[(0, 1)], counts[(1, 0)], counts[(1, 1)]) ratio = lambda n, d: n / d if d else None return dict(n=len(selected), tn=tn, fp=fp, fn=fn, tp=tp, accuracy=ratio(tp+tn, len(selected)), false_reply_rate=ratio(fp, tn+fp), false_hold_rate=ratio(fn, tp+fn), recall=ratio(tp, tp+fn), precision=ratio(tp, tp+fp), f1=ratio(2*tp, 2*tp+fp+fn)) class MediaClock: seconds = 0.0 def time(self): return 1000.0 + self.seconds def monotonic(self): # Non-zero origin matches BaseSmartTurn's 0 sentinel. return 1000.0 + self.seconds def source_audio_start(analyzer): """Return the exact selected buffer-frame start before max-duration trim.""" target = analyzer._speech_start_time - ( analyzer.params.pre_speech_ms / 1000 + analyzer._vad_start_secs) times = [t for t, _ in analyzer._audio_buffer] if not times: return None # BaseSmartTurn's chunks are timestamped at arrival/end, not frame start. index = next((i for i, t in enumerate(times) if t >= target), 0) t, chunk = analyzer._audio_buffer[index] return max(0.0, t - 1000 - len(chunk) / analyzer.sample_rate) def replay(call_id, labels, history, audio_path, clock, events, channel): audio, sr = sf.read(audio_path, dtype="int16", always_2d=True) if sr not in (8000, 16000) or channel >= audio.shape[1]: raise ValueError(f"Unsupported audio format: {audio_path}") pcm = audio[:, channel] vad = SileroVADAnalyzer(sample_rate=sr, params=VADParams()) vad.set_sample_rate(sr) analyzer = LocalSmartTurnAnalyzerV3( smart_turn_model_path=str(HERE / "smart-turn-v3.2-cpu.onnx"), cpu_count=1, params=SmartTurnParams()) analyzer.set_sample_rate(sr) frame_samples = sr * 32 // 1000 speaking = False completed = False last_origin = None last_stop_ms = None last_action = None previous_vad_start = None pointer = 0 results = [] users = [r for r in history if r["role"] == "user"] agents = [r for r in history if r["role"] == "agent"] def event(kind, result=None, origin=None): return dict(call_id=call_id, event_ms=round(clock.seconds*1000, 3), event=kind, probability=result.probability if result else None, prediction=int(result.is_complete) if result else None, audio_origin_ms=round(origin*1000, 3) if origin is not None else None) for label in sorted(labels, key=lambda r: int(r["decision_ms"])): decision = int(label["decision_ms"]) cutoff = decision * sr // 1000 if cutoff > len(pcm): raise ValueError(f"Decision beyond audio: {call_id} {decision}") # Only complete transport frames observed by this checkpoint are fed. while pointer + frame_samples <= cutoff: end = pointer + frame_samples clock.seconds = end / sr chunk = pcm[pointer:end].tobytes() state = vad._run_analyzer(chunk) new_start = state == VADState.SPEAKING and not speaking new_stop = state == VADState.QUIET and speaking # The aggregator queues VAD events: raw audio reaches the turn # strategy with its old speech flag, followed by the VAD event. origin = source_audio_start(analyzer) if analyzer.speech_triggered else None appended = analyzer.append_audio(chunk, speaking) if analyzer.speech_triggered: last_origin = source_audio_start(analyzer) if appended == EndOfTurnState.COMPLETE: completed = True last_action = event("silence_timeout", origin=origin) events.append(last_action) if new_start: speaking = True completed = False last_action = None previous_vad_start = clock.seconds * 1000 analyzer.update_vad_start_secs(vad.params.start_secs) events.append(event("vad_start")) elif new_stop: speaking = False last_stop_ms = clock.seconds * 1000 origin = source_audio_start(analyzer) state, result = analyzer._process_speech_segment(analyzer._audio_buffer) last_origin = origin events.append(event("vad_stop", result, origin)) if state == EndOfTurnState.COMPLETE: completed = True last_action = event("model_complete", result, origin) events.append(last_action) analyzer.clear() else: completed = False pointer = end # Non-mutating model probe at EXACT decision_ms, using the start # selected by the current cache; after an earlier completion, use the # last completed segment start and label that diagnostic explicitly. active = analyzer.speech_triggered origin = source_audio_start(analyzer) if active else last_origin source = "active_turn" if active else "previous_completed_turn" prob = None prediction = None start_sample = None real_ms = None if origin is not None: start_sample = max(round(origin*sr), cutoff - int(sr*8)) if start_sample < cutoff: sample = pcm[start_sample:cutoff].astype(np.float32) / 32768.0 output = analyzer._predict_endpoint(sample) prob, prediction = output["probability"], output["prediction"] real_ms = len(sample)*1000/sr assert start_sample <= cutoff and real_ms <= 8000 before_users = [r for r in users if int(r["start_ms"]) < decision] current_user = max(before_users, key=lambda r: int(r["start_ms"])) if before_users else None # Incomplete ASR history does not invalidate an observed audio/VAD # prediction; keep it as a separate alignment issue for review. input_valid = prediction is not None current_asr_speaking = any(int(r["start_ms"]) <= decision < int(r["end_ms"]) for r in users) current_agent_speaking = any(int(r["start_ms"]) < decision < int(r["end_ms"]) for r in agents) label_value = int(label["can_reply_now"]) if label["can_reply_now"] in ("0", "1") else None strategy_prediction = int(completed) result = dict(label) result.update(input_valid=int(input_valid), input_issue="" if input_valid else "no_vad_origin", history_issue="" if current_user is not None else "no_prior_user_asr_segment", source_audio_channel=channel, point_probability=prob, point_prediction=prediction, point_correct=int(prediction == label_value) if input_valid else None, point_audio_source=source, point_audio_start_ms=start_sample*1000/sr if start_sample is not None else None, point_audio_end_ms=decision, point_real_audio_ms=real_ms, model_input_ms=8000, leading_zero_ms=8000-real_ms if real_ms is not None else None, vad_speaking=int(speaking), vad_last_start_ms=previous_vad_start, vad_last_stop_ms=last_stop_ms, strategy_prediction=strategy_prediction, strategy_correct=int(strategy_prediction == label_value) if input_valid else None, strategy_last_action_ms=last_action["event_ms"] if last_action else None, strategy_last_action=last_action["event"] if last_action else None, strategy_probability=last_action["probability"] if last_action else None, asr_user_segment_crosses_point=int(current_asr_speaking), asr_agent_segment_crosses_point=int(current_agent_speaking), label_status="draft" if label["annotator"] == "xlsx_draft" else "unconfirmed_manual") results.append(result) vad._executor.shutdown(wait=True) return results, dict(call_id=call_id, sample_rate=sr, channels=audio.shape[1], selected_user_channel=channel, audio_sha256=digest(audio_path)) def main(): parser = argparse.ArgumentParser() parser.add_argument("--channel", type=int, default=0) parser.add_argument("--output", type=Path, default=HERE / "results" / "work-data-pipecat") args = parser.parse_args() labels_path = ROOT / "labels/EOT标注/工作数据/eot_labels.csv" history_path = ROOT / "labels/EOT标注/工作数据/asr_history.csv" labels, history = read_csv(labels_path), read_csv(history_path) keys = [(r["call_id"], r["decision_ms"]) for r in labels] assert len(set(keys)) == len(keys), "Duplicate annotation keys" clock = MediaClock() base_module.time = SimpleNamespace(monotonic=clock.monotonic, perf_counter=time.perf_counter) silero_module.time = SimpleNamespace(time=clock.time) logger.remove() logger.add(lambda message: None, level="ERROR") events, predictions, audio_inputs = [], [], [] for call_id in sorted({r["call_id"] for r in labels}): clock.seconds = 0 rows, meta = replay(call_id, [r for r in labels if r["call_id"] == call_id], [r for r in history if r["call_id"] == call_id], ROOT / "data/韩泰机器人抢话/音频" / f"{call_id}.mp3", clock, events, args.channel) predictions.extend(rows) audio_inputs.append(meta) print(f"{call_id}: {len(rows)} points; {sum(r['input_valid'] for r in rows)} eligible", flush=True) predictions.sort(key=lambda r: keys.index((r["call_id"], r["decision_ms"]))) args.output.mkdir(parents=True, exist_ok=True) write_csv(args.output / "predictions.csv", predictions) write_csv(args.output / "events.csv", events) errors = [r for r in predictions if r["input_valid"] and (not r["point_correct"] or not r["strategy_correct"])] write_csv(args.output / "errors.csv", errors) grouped = {} for field in ("label_status", "call_id", "reason", "point_audio_source"): grouped[field] = {value: { "point_model": metrics([r for r in predictions if r[field] == value], "point_prediction"), "audio_strategy": metrics([r for r in predictions if r[field] == value], "strategy_prediction")} for value in sorted({r[field] for r in predictions})} model_path = HERE / "smart-turn-v3.2-cpu.onnx" summary = dict( total_points=len(predictions), eligible_points=sum(r["input_valid"] for r in predictions), official_verified_points=0, point_model=metrics(predictions, "point_prediction"), audio_strategy=metrics(predictions, "strategy_prediction"), grouped=grouped, alignment_issue_points=sum(bool(r["history_issue"]) for r in predictions), history_aligned_subset={ "point_model": metrics([r for r in predictions if not r["history_issue"]], "point_prediction"), "audio_strategy": metrics([r for r in predictions if not r["history_issue"]], "strategy_prediction")}, config=dict(pipecat_version=importlib.metadata.version("pipecat-ai"), onnxruntime_version=importlib.metadata.version("onnxruntime"), vad_params=VADParams().model_dump(), smart_turn_params=SmartTurnParams().model_dump(), frame_ms=32, media_clock=True, cpu_count=1, threshold=0.5, model_sha256=digest(model_path), label_sha256=digest(labels_path), history_sha256=digest(history_path), audio_inputs=audio_inputs, replay_source_sha256={str(cls.__module__): digest(Path(inspect.getfile(cls))) for cls in (SileroVADAnalyzer, LocalSmartTurnAnalyzerV3, base_module.BaseSmartTurn)}), definitions={ "point_model": "Non-mutating diagnostic inference at exact decision_ms. Active cache origin is used; after replay completed a turn, last completed origin is used and explicitly flagged. These extra queries never change replay state.", "audio_strategy": "Whether audio-only EOT is latched at decision_ms: actual VAD stops trigger inference; resumed VAD speech clears the latch; incomplete preserves cache; complete clears cache; silence timeout forces completion. It does not measure whether an earlier wrong EOT happened before speech resumed.", "event_order": "Audio appended with previous speaking flag, then queued VAD state changes handled. 32ms frames; no future or partial frames fed to VAD before a checkpoint.", "silero_reset": "Pipecat periodic five-second Silero state resets use media clock, not processing wall time.", "asr": "No realtime ASR return/finalization logs; transcript wait gate, external 5s controller timeout and ASR fallback are not simulated. History only audits point validity and speaker overlap, never enters model input.", "labels": "29 xlsx_draft and 15 no annotator/confidence; exploratory results only, no confirmed-label production accuracy claim.", "channel": "User channel 0 selected by RMS contrasts on non-overlapping ASR user/agent segments; echo/overlap can remain.", "invalid_points": "No detected VAD origin/prediction: excluded from metrics. Missing prior ASR user segments are alignment issues, retained in exploratory audio metrics and separately reported.", "scope": "Selected point classification/state snapshots; no complete turn boundaries, interval-level interruption rates or end latency claims."}) (args.output / "summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2)+"\n") print(json.dumps({k: summary[k] for k in ("total_points", "eligible_points", "point_model", "audio_strategy")}, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()