"""Classify each annotation independently without end-signal state. Past ASR segments locate the current user speech block; continuations separated by <=3 s keep the block start. No label value or earlier model prediction is used to construct inputs. For missing ASR use past Silero VAD starts only. """ import csv import hashlib import importlib.metadata import json from collections import Counter from pathlib import Path import numpy as np import soundfile as sf from loguru import logger from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 HERE = Path(__file__).resolve().parent ROOT = HERE.parents[1] OUT = HERE / "results/work-data-can-reply" def read(path): with path.open(encoding="utf-8-sig", newline="") as f: return list(csv.DictReader(f)) def write(path, rows): with path.open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) def metrics(rows): c = Counter((int(r["can_reply_now"]), r["prediction"]) for r in rows if r["can_reply_now"] in ("0", "1") and r["prediction"] is not None) tn, fp, fn, tp = c[0, 0], c[0, 1], c[1, 0], c[1, 1] total = tn + fp + fn + tp ratio = lambda n, d: n/d if d else None return dict(n=total, tn=tn, fp=fp, fn=fn, tp=tp, accuracy=ratio(tn+tp, total), false_reply_rate=ratio(fp, fp+tn), false_hold_rate=ratio(fn, fn+tp), recall=ratio(tp, tp+fn), precision=ratio(tp, tp+fp), f1=ratio(2*tp, 2*tp+fp+fn)) def block_origin(history, decision): # Use only segments that have begun by this decision. Full ASR end times # are only used for PREVIOUS segments that have already ended. users = sorted((r for r in history if r["role"] == "user" and int(r["start_ms"]) < decision), key=lambda r: int(r["start_ms"])) if not users: return None index = len(users)-1 while index > 0: previous_end = int(users[index-1]["end_ms"]) start = int(users[index]["start_ms"]) if previous_end > decision or start-previous_end > 3000: break # A complete non-overlapping robot question between user segments # establishes a new observed conversational block. if any(r["role"] == "agent" and previous_end <= int(r["start_ms"]) and int(r["end_ms"]) <= start for r in history): break index -= 1 return max(0, int(users[index]["start_ms"])-500) def main(): logger.remove() labels_path = ROOT / "labels/EOT标注/工作数据/eot_labels.csv" history_path = ROOT / "labels/EOT标注/工作数据/asr_history.csv" labels, history = read(labels_path), read(history_path) events = read(HERE / "results/work-data-pipecat/events.csv") analyzer = LocalSmartTurnAnalyzerV3( smart_turn_model_path=str(HERE / "smart-turn-v3.2-cpu.onnx"), cpu_count=1) audio = {} rows = [] for label in labels: call = label["call_id"] if call not in audio: pcm, sr = sf.read(ROOT / "data/韩泰机器人抢话/音频" / f"{call}.mp3", dtype="int16", always_2d=True) audio[call] = (pcm[:, 0], sr) pcm, sr = audio[call] decision = int(label["decision_ms"]) origin = block_origin([r for r in history if r["call_id"] == call], decision) source = "asr_user_block" issue = "" if origin is None: # VAD events are independent of Smart Turn acceptance/rejection. # Use only a start observed BEFORE the annotation, no EOT events. starts = [float(e["event_ms"]) for e in events if e["call_id"] == call and e["event"] == "vad_start" and float(e["event_ms"]) <= decision] if not starts: raise ValueError(f"No causal user/VAD onset for {call} {decision}") origin = max(0, starts[-1] + 32 - 200 - 500) source = "past_vad_missing_asr" issue = "missing_earlier_user_asr_segment" start = max(round(origin*sr/1000), decision*sr//1000 - sr*8) end = decision*sr//1000 assert 0 <= start < end <= len(pcm) samples = pcm[start:end].astype(np.float32)/32768.0 analyzer.set_sample_rate(sr) result = analyzer._predict_endpoint(samples) r = dict(label) r.update(probability=result["probability"], prediction=result["prediction"], correct=int(result["prediction"] == int(label["can_reply_now"])), error_type=("false_reply" if result["prediction"] else "false_hold") if result["prediction"] != int(label["can_reply_now"]) else "correct", audio_start_ms=start*1000/sr, audio_end_ms=decision, real_audio_ms=(end-start)*1000/sr, model_input_ms=8000, leading_zero_ms=8000-(end-start)*1000/sr, user_block_origin_ms=origin, input_source=source, alignment_issue=issue, audio_channel=0, label_status="draft" if label["annotator"] == "xlsx_draft" else "unconfirmed_manual") rows.append(r) OUT.mkdir(parents=True, exist_ok=True) write(OUT / "predictions.csv", rows) write(OUT / "errors.csv", [r for r in rows if not r["correct"]]) thresholds = [] for threshold in (0.5, 0.7, 0.8, 0.9, 0.95, 0.98, 0.99): recomputed = [dict(r, prediction=int(r["probability"] > threshold)) for r in rows] thresholds.append(dict(threshold=threshold, **metrics(recomputed))) write(OUT / "thresholds.csv", thresholds) summary = dict(total_points=len(rows), metrics=metrics(rows), grouped={field: {value: metrics([r for r in rows if r[field] == value]) for value in sorted({r[field] for r in rows})} for field in ("label_status", "call_id", "reason", "input_source")}, config=dict(pipecat_version=importlib.metadata.version("pipecat-ai"), model="smart-turn-v3.2-cpu.onnx", threshold=0.5, pre_speech_ms=500, max_duration_secs=8, continuation_gap_ms=3000, selected_audio_channel=0, model_sha256=hashlib.sha256((HERE/"smart-turn-v3.2-cpu.onnx").read_bytes()).hexdigest(), label_sha256=hashlib.sha256(labels_path.read_bytes()).hexdigest(), history_sha256=hashlib.sha256(history_path.read_bytes()).hexdigest()), method="Each exact decision_ms is classified directly by the model. Earlier model scores, EOT signals, VAD stop gates, silence timeout forced replies and ASR return gates never determine prediction or clear this input.", threshold_diagnostics=thresholds, audio_origin="Past ASR user segments locate current speech block. Consecutive user segments within 3 s share the first start unless a complete non-overlapping robot question separates them. Keep 500 ms before that start and then last 8 s. Intra-segment pauses are not resolved by coarse ASR timing. Two missing-ASR points use only past Silero VAD starts; full-audio model events are not used.", limitations=["Exploratory scoring: 29 xlsx_draft and 15 rows missing annotator/confidence; no explicit verified status.", "Coarse ASR speech block origin approximates the discussed cache rule; this is point classification, not a complete online Pipecat strategy replay.", "cb3238dc 13540 text is 辽宁葫芦岛 but its corresponding user ASR starts at 16600; review alignment. e304f740 11531 surname utterance is absent from ASR history.", "User channel 0 can contain robot echo/overlap; no future ASR-based masking.", "Selected-point false-reply rate is not interval-level interruption rate."]) (OUT/"summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2)+"\n") print(json.dumps(summary["metrics"], ensure_ascii=False, indent=2)) for r in rows: if not r["correct"]: print(r["call_id"][:8], r["decision_ms"], r["can_reply_now"], round(r["probability"], 5), r["audio_start_ms"], r["user_text_at_decision"]) if __name__ == "__main__": main()