File size: 5,688 Bytes
cc9dbda | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 | """Coherence Dial validation harness.
Per swept clip:
- Whisper-WER vs the ORIGINAL typed sentence (should RISE as dial level rises)
- Resemblyzer cosine vs the level-0 same-voice/same-seed reference clip
(should stay >= 0.85 across all levels — voice preserved)
Aggregate:
- mean WER per level (and per voice)
- mean cosine per level (and per voice)
- Spearman( level, mean_WER ) --- gate >= +0.80
- min cosine across levels --- gate >= 0.85
Verdict: PASS / PARTIAL / FAIL with a per-level breakdown JSON report.
Hallucination guard: when Whisper's avg_logprob falls below the threshold the transcription is
junk (model invented text) — we floor WER to 1.0 in that case so glossolalia doesn't get a
spuriously LOW WER because the model dreamed coherent words from noise.
"""
import argparse
import json
import math
import sys
from pathlib import Path
import numpy as np
def load_manifest(path: Path):
rows = json.loads(path.read_text())
if isinstance(rows, dict) and "clips" in rows:
rows = rows["clips"]
return rows
def whisper_wer(wav_path, ref_text, model, no_speech_threshold=0.8, logprob_threshold=-1.5):
import jiwer
out = model.transcribe(str(wav_path), no_speech_threshold=no_speech_threshold,
logprob_threshold=logprob_threshold, condition_on_previous_text=False,
language="en", fp16=False)
hyp = (out.get("text") or "").strip()
seg_avg_logprob = np.mean([s.get("avg_logprob", -10) for s in out.get("segments", [])]) if out.get("segments") else -10
seg_no_speech = np.mean([s.get("no_speech_prob", 1.0) for s in out.get("segments", [])]) if out.get("segments") else 1.0
if not hyp or seg_avg_logprob < logprob_threshold or seg_no_speech > no_speech_threshold:
return 1.0, hyp, float(seg_avg_logprob), float(seg_no_speech)
wer = jiwer.wer(ref_text.lower(), hyp.lower())
return float(min(wer, 1.0)), hyp, float(seg_avg_logprob), float(seg_no_speech)
def resemblyzer_embed(wav_path, encoder):
from resemblyzer import preprocess_wav
wav = preprocess_wav(str(wav_path))
return encoder.embed_utterance(wav)
def main():
p = argparse.ArgumentParser()
p.add_argument("--manifest", required=True, help="sweep manifest JSON (from sweep_dial.py)")
p.add_argument("--out", default="sweep/eval_report.json")
p.add_argument("--whisper", default="base.en")
p.add_argument("--spearman-gate", type=float, default=0.80)
p.add_argument("--cosine-gate", type=float, default=0.85)
args = p.parse_args()
rows = load_manifest(Path(args.manifest))
if not rows:
print("empty manifest", file=sys.stderr); sys.exit(1)
import whisper
from resemblyzer import VoiceEncoder
print(f"loading whisper:{args.whisper} + resemblyzer", file=sys.stderr)
asr = whisper.load_model(args.whisper)
enc = VoiceEncoder()
# index lv0 reference per (voice, seed, sentence)
ref_by = {}
for r in rows:
if r["level"] == 0:
ref_by[(r["voice"], r["seed"], r["sentence"])] = r["wav"]
per_clip = []
for i, r in enumerate(rows):
wav = r["wav"]; ref = r["sentence"]
wer, hyp, lp, nsp = whisper_wer(wav, ref, asr)
cos = 1.0
if r["level"] != 0:
ref_wav = ref_by.get((r["voice"], r["seed"], r["sentence"]))
if ref_wav and Path(ref_wav).exists():
e_clip = resemblyzer_embed(wav, enc)
e_ref = resemblyzer_embed(ref_wav, enc)
cos = float(np.dot(e_clip, e_ref) / ((np.linalg.norm(e_clip) * np.linalg.norm(e_ref)) + 1e-9))
per_clip.append({**r, "wer": wer, "hyp": hyp, "logprob": lp, "no_speech": nsp, "cos_vs_lv0": cos})
if (i + 1) % 20 == 0:
print(f" {i + 1}/{len(rows)}", file=sys.stderr)
# aggregate per level
levels = sorted({r["level"] for r in per_clip})
per_level = {}
for lv in levels:
sub = [r for r in per_clip if r["level"] == lv]
per_level[lv] = {
"n": len(sub),
"mean_wer": float(np.mean([r["wer"] for r in sub])),
"mean_cos_vs_lv0": float(np.mean([r["cos_vs_lv0"] for r in sub])),
}
xs = np.array(levels, dtype=float)
ys = np.array([per_level[lv]["mean_wer"] for lv in levels])
cs = np.array([per_level[lv]["mean_cos_vs_lv0"] for lv in levels])
from scipy.stats import spearmanr
sp = float(spearmanr(xs, ys).correlation) if len(xs) >= 3 else float("nan")
min_cos = float(cs[xs > 0].min()) if (xs > 0).any() else 1.0
verdict = ("PASS" if (sp >= args.spearman_gate and min_cos >= args.cosine_gate)
else "PARTIAL" if (sp >= 0.5 or min_cos >= 0.75) else "FAIL")
report = {
"manifest": args.manifest, "n_clips": len(per_clip),
"per_level": per_level, "spearman_wer_vs_level": sp,
"min_cos_vs_lv0_across_levels": min_cos,
"gates": {"spearman": args.spearman_gate, "cosine": args.cosine_gate},
"verdict": verdict,
}
Path(args.out).parent.mkdir(parents=True, exist_ok=True)
Path(args.out).write_text(json.dumps({**report, "per_clip": per_clip}, indent=2))
print("\n===== COHERENCE DIAL EVAL =====")
for lv in levels:
r = per_level[lv]
print(f" level {lv}: WER={r['mean_wer']:.3f} cos_vs_lv0={r['mean_cos_vs_lv0']:.3f} (n={r['n']})")
print(f"\nSpearman(WER, level) = {sp:+.3f} (gate >= +{args.spearman_gate:.2f})")
print(f"min cos vs lv0 = {min_cos:.3f} (gate >= {args.cosine_gate:.2f})")
print(f"VERDICT: {verdict}")
print(f"full report -> {args.out}")
if __name__ == "__main__":
main()
|