| """A/B eval: fishaudio/drama-3 (FP8, mounted) vs akhaliq/drama-3-int4 (hybrid int4). |
| Generates clips for a fixed prompt set on both, transcribes with faster-whisper large-v3, |
| computes WER vs prompt text and speaker similarity (resemblyzer) for clone prompts. |
| Uploads eval_results.json + clips to akhaliq/drama-3-int4/eval/.""" |
| import os, subprocess, sys, time, json, re |
|
|
| import requests |
|
|
| VENV_PY = "/root/venv/bin/python" |
| OUT = "/root/eval_out" |
| REF_URL = "https://huggingface.co/datasets/zhaochenyang20/seed-tts-eval-mini/resolve/main/en/prompt-wavs/common_voice_en_10119832.wav" |
| REF_TEXT = "We asked over twenty different people, and they all said it was his." |
|
|
| PROMPTS = [ |
| ("plain1", "Hello, how are you? The weather is lovely today, so let's take a long walk by the river.", False), |
| ("plain2", "Get the trust fund to the bank early.", True), |
| ("plain3", "The quick brown fox jumps over the lazy dog, and then runs into the forest.", False), |
| ("long1", "The quarterly report is due on Thursday, so please finish it by noon. After that, we will review the numbers together and decide whether the new pricing plan should launch next month or wait until spring.", False), |
| ("tags1", "[sighs] Fine. If we miss it, it's on you. [laughs] Well, maybe that's fair.", True), |
| ("zh1", "今天天气很好，我们一起去公园散步吧。", False), |
| ("ja1", "今日は天気がいいので、一緒に公園を散歩しましょう。", False), |
| ("clone1", "The meeting has been moved to three o'clock on Friday afternoon.", True), |
| ] |
|
|
|
|
| def sh(cmd, **kw): |
| print("+", " ".join(str(c) for c in cmd), flush=True) |
| subprocess.run(cmd, check=True, **kw) |
|
|
|
|
| def tail(path, n=12000): |
| try: |
| return open(path, errors="replace").read()[-n:] |
| except Exception as e: |
| return f"<no log: {e}>" |
|
|
|
|
| def install_stack(): |
| import io, tarfile, pathlib, glob as _glob |
| r = requests.get("https://github.com/sgl-project/sglang-omni/archive/refs/heads/main.tar.gz", timeout=300) |
| r.raise_for_status() |
| with tarfile.open(fileobj=io.BytesIO(r.content), mode="r:gz") as tf: |
| tf.extractall("/root") |
| if not os.path.exists("/root/sglang-omni"): |
| os.replace("/root/sglang-omni-main", "/root/sglang-omni") |
| sh(["uv", "venv", "--python", f"{sys.version_info.major}.{sys.version_info.minor}", "/root/venv"]) |
| try: |
| sh(["uv", "pip", "install", "--prerelease=allow", "--python", VENV_PY, |
| "-e", ".[fishaudio]"], cwd="/root/sglang-omni") |
| except subprocess.CalledProcessError: |
| with open("/root/overrides.txt", "w") as fh: |
| fh.write("protobuf>=6.31.1,<7.0.0\n") |
| sh(["uv", "pip", "install", "--prerelease=allow", "--python", VENV_PY, |
| "--override", "/root/overrides.txt", "-e", "/root/sglang-omni[fishaudio]"]) |
| sh(["uv", "pip", "install", "--python", VENV_PY, "ninja", "numpy", |
| "faster-whisper", "jiwer", "resemblyzer", "soundfile", "setuptools"]) |
| sh(["uv", "pip", "install", "--python", VENV_PY, "nvidia-cuda-nvcc"]) |
| _nvcc = _glob.glob("/root/venv/lib/python3.12/site-packages/nvidia/**/bin/nvcc", recursive=True) |
| NVCC_BIN = str(pathlib.Path(_nvcc[0]).parent) |
| CUDA_HOME = str(pathlib.Path(_nvcc[0]).parent.parent) |
| _libdirs = sorted({str(pathlib.Path(p).parent) for p in _glob.glob( |
| "/root/venv/lib/python3.12/site-packages/nvidia/**/libcudart.so*", recursive=True)}) |
| for d in _libdirs: |
| for so in pathlib.Path(d).glob("libcudart.so.*"): |
| link = pathlib.Path(d) / "libcudart.so" |
| if not link.exists(): |
| link.symlink_to(so.name) |
| os.makedirs("/root/stubs", exist_ok=True) |
| with open("/root/stubs/deep_ep.py", "w") as fh: |
| fh.write('"""Stub: deep_ep needs a CUDA toolkit this image lacks; unused on our path."""\n' |
| "class Buffer:\n pass\n\nclass Config:\n pass\n") |
| env = dict(os.environ, PYTHONPATH="/root/stubs", |
| SGLANG_OMNI_STARTUP_TIMEOUT="2700", |
| PATH=f"{NVCC_BIN}:/root/venv/bin:" + os.environ["PATH"], |
| CUDA_HOME=CUDA_HOME, |
| LIBRARY_PATH=":".join(_libdirs) + ":" + os.environ.get("LIBRARY_PATH", ""), |
| LD_LIBRARY_PATH=":".join(_libdirs) + ":" + os.environ.get("LD_LIBRARY_PATH", "")) |
| return env |
|
|
|
|
| def serve_and_generate(model_dir, tag, serve_env, ref, engine_frac): |
| log_path = f"/root/serve_{tag}.log" |
| proc = subprocess.Popen( |
| ["/root/venv/bin/sgl-omni", "serve", "--model-path", model_dir, "--port", "8000", |
| "--tts_engine.gpu_memory_fraction", engine_frac, "--vocoder.gpu_memory_fraction", "0.2"], |
| stdout=open(log_path, "w"), stderr=subprocess.STDOUT, env=serve_env) |
| deadline = time.time() + 2700 |
| while time.time() < deadline: |
| if proc.poll() is not None: |
| print(f"SERVER {tag} DIED:\n" + tail(log_path), flush=True) |
| sys.exit(1) |
| try: |
| if requests.get("http://localhost:8000/health", timeout=3).ok: |
| break |
| except Exception: |
| pass |
| time.sleep(5) |
| else: |
| print(f"SERVER {tag} NOT READY:\n" + tail(log_path), flush=True) |
| sys.exit(1) |
| print(f"SERVER {tag} UP", flush=True) |
| outdir = f"{OUT}/{tag}" |
| os.makedirs(outdir, exist_ok=True) |
| for name, text, use_ref in PROMPTS: |
| payload = {"input": text} |
| if use_ref: |
| payload["references"] = [ref] |
| r = requests.post("http://localhost:8000/v1/audio/speech", json=payload, timeout=600) |
| print(f"[{tag}] {name}: HTTP {r.status_code}, {len(r.content)} bytes", flush=True) |
| if r.status_code == 200: |
| open(f"{outdir}/{name}.wav", "wb").write(r.content) |
| else: |
| print(f"[{tag}] {name} FAILED: {r.text[:400]}", flush=True) |
| proc.terminate() |
| time.sleep(10) |
|
|
|
|
| def main(): |
| t0 = time.time() |
| print("=== eval.py v12: int4-only, setuptools fix + fault-tolerant scorer ===", flush=True) |
| from huggingface_hub import hf_hub_download, snapshot_download, HfApi |
| serve_env = install_stack() |
| |
| patch = hf_hub_download("akhaliq/drama-3-int4", "apply_patch.py") |
| sh([VENV_PY, patch, "/root/sglang-omni"]) |
| int4_dir = snapshot_download("akhaliq/drama-3-int4") |
| ref_wav = requests.get(REF_URL, timeout=120).content |
| open("/root/ref.wav", "wb").write(ref_wav) |
| ref = {"audio_path": "/root/ref.wav", "text": REF_TEXT} |
|
|
| os.makedirs(OUT, exist_ok=True) |
| |
| serve_and_generate(int4_dir, "int4", serve_env, ref, "0.62") |
|
|
| |
| scorer = "/root/score.py" |
| with open(scorer, "w") as fh: |
| fh.write(SCORER_SRC) |
| sh([VENV_PY, scorer, OUT, "/root/ref.wav", json.dumps(PROMPTS)]) |
|
|
| HfApi(token=os.environ["HF_TOKEN"]).upload_folder( |
| repo_id="akhaliq/drama-3-int4", folder_path=OUT, path_in_repo="eval", |
| commit_message="eval: FP8 vs int4-hybrid, Whisper WER + speaker similarity") |
| print(f"EVAL DONE in {time.time()-t0:.0f}s", flush=True) |
|
|
|
|
| SCORER_SRC = r''' |
| import sys, os, json, re |
| out_dir, ref_path, prompts_json = sys.argv[1], sys.argv[2], sys.argv[3] |
| PROMPTS = json.loads(prompts_json) |
| |
| def ref_text(t): |
| return re.sub(r"\[[^\]]+\]", "", t).strip() |
| |
| from faster_whisper import WhisperModel |
| from jiwer import wer |
| import numpy as np, soundfile as sf |
| |
| wm = WhisperModel("large-v3", device="cuda", compute_type="float16") |
| try: |
| from resemblyzer import VoiceEncoder, preprocess_wav |
| enc = VoiceEncoder() |
| ref_emb = enc.embed_utterance(preprocess_wav(ref_path)) |
| except Exception as e: |
| print(f"speaker-sim disabled (resemblyzer unavailable: {e})", flush=True) |
| enc = ref_emb = None |
| |
| results = {} |
| for tag in ("int4",): |
| rows = {} |
| for name, text, use_ref in PROMPTS: |
| p = f"{out_dir}/{tag}/{name}.wav" |
| if not os.path.exists(p): |
| rows[name] = {"error": "no clip"} |
| continue |
| lang = "zh" if name.startswith("zh") else ("ja" if name.startswith("ja") else "en") |
| segs, _ = wm.transcribe(p, language=lang) |
| hyp = " ".join(s.text for s in segs).strip() |
| row = {"hyp": hyp, "ref": ref_text(text), "wer": round(wer(ref_text(text).lower(), hyp.lower()), 4)} |
| if use_ref and enc is not None: |
| try: |
| emb = enc.embed_utterance(preprocess_wav(p)) |
| row["speaker_sim"] = round(float(np.dot(emb, ref_emb) / |
| (np.linalg.norm(emb) * np.linalg.norm(ref_emb))), 4) |
| except Exception as e: |
| row["speaker_sim_error"] = str(e) |
| rows[name] = row |
| wers = [r["wer"] for r in rows.values() if "wer" in r] |
| sims = [r["speaker_sim"] for r in rows.values() if "speaker_sim" in r] |
| results[tag] = {"rows": rows, |
| "mean_wer": round(sum(wers) / len(wers), 4) if wers else None, |
| "mean_speaker_sim": round(sum(sims) / len(sims), 4) if sims else None} |
| with open(f"{out_dir}/eval_results.json", "w") as fh: |
| json.dump(results, fh, indent=2, ensure_ascii=False) |
| print(json.dumps({t: {"mean_wer": r["mean_wer"], "mean_speaker_sim": r["mean_speaker_sim"]} |
| for t, r in results.items()}, indent=2)) |
| ''' |
|
|
| if __name__ == "__main__": |
| main() |
|
|
