"""한국어 TTS 사람 청취평가 — 블라인드 받아쓰기 + 0~10점. 왜 필요한가 ----------- Open Ko-S2S 리더보드의 TTS 발음 지표는 합성음을 ASR 로 되받아쓴 CER 이다. 4090 실측에서 이 지표가 발음이 아니라 **숫자 표기**를 재고 있다는 것이 드러났다. (기존 20문장 세트, 표기 보정 없음) Qwen3-TTS 전체 CER 11.08% ├ 숫자 표기 문장(n=9) 23.67% ← ASR 이 "삼십 분"을 "30분"으로 씀 └ 숫자 없는 문장(n=11) 0.50% ← 발음은 사실상 완벽 허용 표기(variants)를 도입해 이 편향을 걷어내자 숫자문장 CER 이 23.67% → 3.09% 로 떨어졌다(의원 접수대 100문장 기준). 남은 것이 진짜 발음 오차다. (의원 100문장, variants 적용) Qwen3-TTS 전체 1.50% 숫자 3.09% 숫자없음 1.02% MMS-TTS 전체 6.28% 숫자 7.74% 숫자없음 5.84% 다만 이 보정이 옳은지, 남은 오차가 정말 발음 문제인지는 자동 지표로 확정할 수 없다. 사람 귀가 유일한 기준선이라 이 앱으로 human CER 과 주관 점수를 모은다. 설계 ---- - **블라인드**: 어떤 모델의 음성인지, 정답 문장이 무엇인지 제출 전에는 감춘다. 정답을 먼저 보여주면 받아쓰기가 베끼기가 되어 명료도 측정이 무의미해진다. - **표기 통제**: 사람도 "삼십 분"을 듣고 "30분"이라 적는다. 그래서 안내로 한글 표기를 요청하고, 채점 시에도 숫자를 정규화해 표기 차이를 제거한다. - **저장**: HF_TOKEN 시크릿이 있으면 데이터셋 repo 로 누적, 없으면 로컬 파일로 받아 두고 화면에 경고한다(조용히 유실되지 않게). """ from __future__ import annotations import json import os import random import re import unicodedata import uuid from datetime import datetime, timezone from pathlib import Path import gradio as gr ROOT = Path(__file__).resolve().parent AUDIO_DIR = ROOT / "audio" MANIFEST = json.loads((AUDIO_DIR / "manifest.json").read_text(encoding="utf-8")) ITEMS = MANIFEST["items"] # 도메인: 의원(1차 의료기관) 접수대 100문장 × TTS 2종 = 200 샘플. # 음성봇이 가장 먼저 투입되는 구간이라 이 도메인의 발음이 실사용을 좌우한다. DATASET_REPO = os.environ.get("EVAL_DATASET_REPO", "baryonlabs/ko-tts-human-eval-votes") HF_TOKEN = os.environ.get("HF_TOKEN") LOCAL_LOG = Path(os.environ.get("EVAL_LOCAL_LOG", "/tmp/ko_tts_votes.jsonl")) # ── 채점 ──────────────────────────────────────────────────────────────── # 한글 수사 → 숫자. 사람도 ASR 도 "삼십 분"을 "30분"으로 적을 수 있어서, 표기 # 차이가 명료도 점수를 오염시키지 않도록 양쪽을 같은 형태로 만든 뒤 비교한다. _SINO = {"영": 0, "공": 0, "일": 1, "이": 2, "삼": 3, "사": 4, "오": 5, "육": 6, "칠": 7, "팔": 8, "구": 9} _UNIT = {"십": 10, "백": 100, "천": 1000} _NATIVE = {"한": 1, "두": 2, "세": 3, "네": 4, "다섯": 5, "여섯": 6, "일곱": 7, "여덟": 8, "아홉": 9, "열": 10, "스물": 20, "서른": 30, "마흔": 40, "쉰": 50, "예순": 60, "일흔": 70, "여든": 80, "아흔": 90} _PCT = re.compile(r"퍼센트|프로") def _sino_to_int(chunk: str): """'삼십사' → 34. 해석 불가하면 None.""" total, current, seen = 0, 0, False for ch in chunk: if ch in _SINO: current = _SINO[ch] seen = True elif ch in _UNIT: total += (current or 1) * _UNIT[ch] current = 0 seen = True else: return None return total + current if seen else None def normalize_numbers(text: str) -> str: """숫자 표기를 통일한다: 아라비아 숫자·퍼센트 기호·한글 수사를 한 형태로.""" t = _PCT.sub("%", text) t = re.sub(r"(\d),(?=\d{3})", r"\1", t) # 4,500 → 4500 t = t.replace("m", "미터").replace("M", "미터") for word, val in sorted(_NATIVE.items(), key=lambda kv: -len(kv[0])): t = t.replace(word, str(val)) def repl(m): v = _sino_to_int(m.group(0)) return str(v) if v is not None else m.group(0) t = re.sub(r"[영공일이삼사오육칠팔구십백천]{1,8}", repl, t) return t def normalize(text: str, fold_numbers: bool = True) -> str: t = unicodedata.normalize("NFC", (text or "").strip()) if fold_numbers: t = normalize_numbers(t) t = re.sub(r"[^\w가-힣%]", "", t) return t def _edit_distance(a: str, b: str) -> int: if not a: return len(b) prev = list(range(len(b) + 1)) for i, ca in enumerate(a, 1): cur = [i] for j, cb in enumerate(b, 1): cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb))) prev = cur return prev[-1] def char_error_rate(ref: str, hyp: str, fold_numbers: bool = True) -> float: r, h = normalize(ref, fold_numbers), normalize(hyp, fold_numbers) if not r: return 0.0 return _edit_distance(r, h) / len(r) # ── 저장 ──────────────────────────────────────────────────────────────── def _scheduler(): """HF_TOKEN 이 있으면 데이터셋 repo 로 누적 저장하는 스케줄러를 만든다.""" if not HF_TOKEN: return None try: from huggingface_hub import CommitScheduler LOCAL_LOG.parent.mkdir(parents=True, exist_ok=True) return CommitScheduler( repo_id=DATASET_REPO, repo_type="dataset", folder_path=LOCAL_LOG.parent, path_in_repo="data", every=5, token=HF_TOKEN, private=False, ) except Exception as exc: # 저장 실패를 조용히 삼키지 않는다 print(f"[warn] CommitScheduler 초기화 실패: {exc}") return None SCHEDULER = _scheduler() STORAGE_NOTE = ( f"✅ 응답이 `{DATASET_REPO}` 데이터셋에 누적됩니다." if SCHEDULER else "⚠️ **HF_TOKEN 시크릿이 없어 응답이 이 컨테이너에만 임시 저장됩니다.** " "Space 재시작 시 사라집니다. Settings → Secrets 에 write 권한 HF_TOKEN 을 넣어주세요." ) def save_vote(record: dict) -> None: LOCAL_LOG.parent.mkdir(parents=True, exist_ok=True) line = json.dumps(record, ensure_ascii=False) if SCHEDULER is not None: with SCHEDULER.lock: with LOCAL_LOG.open("a", encoding="utf-8") as f: f.write(line + "\n") else: with LOCAL_LOG.open("a", encoding="utf-8") as f: f.write(line + "\n") def load_votes() -> list[dict]: if not LOCAL_LOG.exists(): return [] out = [] for line in LOCAL_LOG.read_text(encoding="utf-8").splitlines(): if line.strip(): try: out.append(json.loads(line)) except json.JSONDecodeError: continue return out # ── 출제 ──────────────────────────────────────────────────────────────── def pick_item(seen_ids: list[str]) -> dict: """아직 안 들은 것 우선, 그 안에서 응답 수가 적은 것 우선.""" counts = {} for v in load_votes(): counts[v["item_id"]] = counts.get(v["item_id"], 0) + 1 pool = [it for it in ITEMS if it["id"] not in seen_ids] or ITEMS fewest = min(counts.get(it["id"], 0) for it in pool) return random.choice([it for it in pool if counts.get(it["id"], 0) == fewest]) def start(seen_ids): item = pick_item(seen_ids or []) return ( str(AUDIO_DIR / item["audio"]), item, gr.update(value="", interactive=True), gr.update(value=5, interactive=True), gr.update(visible=False, value=""), gr.update(visible=True), gr.update(visible=False), ) def _best_cer(item, heard, fold_numbers): """참조에 허용 표기(variants)가 있으면 오류가 가장 적은 것을 택한다. `삼십 분` / `30분` 처럼 발음이 같고 표기만 다른 형태를 하나만 정답으로 쓰면, 그 표기를 적은 평가자만 유리해진다. KsponSpeech 가 이중전사를 제공하는 것과 같은 이유다. """ refs = item.get("variants") or [item["ref"]] return min(char_error_rate(r, heard, fold_numbers=fold_numbers) for r in refs) def submit(item, heard, score, seen_ids, rater_id): if not item: return (gr.update(), gr.update(visible=True, value="먼저 '다음 문장'을 눌러주세요."), gr.update(), gr.update(), seen_ids) if not (heard or "").strip(): return (gr.update(), gr.update(visible=True, value="⚠️ 들린 내용을 입력해주세요."), gr.update(visible=True), gr.update(visible=False), seen_ids) cer_folded = _best_cer(item, heard, True) cer_raw = _best_cer(item, heard, False) record = { "vote_id": str(uuid.uuid4()), "rater_id": rater_id, "ts": datetime.now(timezone.utc).isoformat(), "item_id": item["id"], "model_key": item["model_key"], "model_name": item["model_name"], "sentence_id": item.get("sentence_id", item.get("id")), "category": item.get("category"), "has_number": item.get("has_number"), "ref": item["ref"], "heard": heard.strip(), "score": int(score), "human_cer": round(cer_folded, 4), "human_cer_raw": round(cer_raw, 4), } save_vote(record) seen = (seen_ids or []) + [item["id"]] reveal = ( f"### 제출 완료\n" f"**정답** : {item['ref']}\n\n" f"**입력** : {heard.strip()}\n\n" f"| 지표 | 값 |\n| --- | --- |\n" f"| 사람 CER (숫자 정규화) | **{cer_folded * 100:.1f}%** |\n" f"| 사람 CER (원문 그대로) | {cer_raw * 100:.1f}% |\n" f"| 내 점수 | {int(score)} / 10 |\n\n" f"두 CER 이 다르면 그 차이가 곧 **표기 때문에 생긴 오차**입니다.\n\n" f"누적 응답 {len(load_votes())}개 · 계속하려면 **다음 문장**" ) return (gr.update(visible=True, value=reveal), gr.update(visible=False, value=""), gr.update(visible=False), gr.update(visible=True), seen) def leaderboard(): votes = load_votes() if not votes: return "아직 응답이 없습니다. 몇 개 평가하면 여기에 모델별 집계가 나옵니다." agg = {} for v in votes: a = agg.setdefault(v["model_name"], {"n": 0, "score": 0.0, "cer": 0.0, "cer_raw": 0.0, "n_num": 0, "cer_num": 0.0}) a["n"] += 1 a["score"] += v["score"] a["cer"] += v["human_cer"] a["cer_raw"] += v["human_cer_raw"] if v.get("has_number"): a["n_num"] += 1 a["cer_num"] += v["human_cer"] rows = sorted(agg.items(), key=lambda kv: kv[1]["cer"] / kv[1]["n"]) out = ["| 모델 | 응답수 | 평균 점수 (0~10) | 사람 CER | 사람 CER(원문) | 숫자문장 CER |", "| --- | ---: | ---: | ---: | ---: | ---: |"] for name, a in rows: n = a["n"] num = f"{a['cer_num'] / a['n_num'] * 100:.2f}%" if a["n_num"] else "—" out.append(f"| {name} | {n} | {a['score'] / n:.2f} | " f"{a['cer'] / n * 100:.2f}% | {a['cer_raw'] / n * 100:.2f}% | {num} |") out.append(f"\n총 {len(votes)}개 응답 · 평가자 {len({v['rater_id'] for v in votes})}명") out.append("\n※ '사람 CER'은 숫자 표기를 정규화한 값, '원문'은 적힌 그대로입니다. " "둘의 차이가 표기 관습이 만드는 오차의 크기입니다.") return "\n".join(out) INTRO = f"""# 🎧 한국어 TTS 사람 청취평가 합성음을 듣고 **들리는 대로** 받아쓴 뒤 0~10점을 매겨주세요. 어떤 모델인지, 정답 문장이 무엇인지는 제출 후에 공개됩니다. **받아쓰기 요령** - 들린 그대로 적어주세요. 못 알아들은 부분은 비워두거나 `?` 로 두면 됩니다. - **숫자는 들린 대로 한글로** 적어주세요. (`30분`이 아니라 `삼십 분`) 자동 채점이 숫자 표기 때문에 발음 오류를 과대평가하는 문제를 확인하려는 것이 이 평가의 목적입니다. **0~10점 기준** — 0: 전혀 못 알아들음 · 5: 알아듣지만 어색함 · 10: 사람 발화와 구분 안 됨 {STORAGE_NOTE} """ with gr.Blocks(title="한국어 TTS 사람 청취평가") as demo: seen_state = gr.State([]) item_state = gr.State(None) rater_state = gr.State(lambda: str(uuid.uuid4())[:8]) gr.Markdown(INTRO) with gr.Tab("평가하기"): audio = gr.Audio(label="합성음 (모델명 비공개)", interactive=False, autoplay=False) heard_box = gr.Textbox(label="들린 대로 받아쓰기", lines=2, placeholder="예) 오늘 회의는 오후 세 시 삼십 분에 시작합니다") score_slider = gr.Slider(0, 10, value=5, step=1, label="자연스러움·명료도 (0~10)") warn = gr.Markdown(visible=False) with gr.Row(): submit_btn = gr.Button("제출", variant="primary", visible=False) next_btn = gr.Button("다음 문장", variant="secondary") reveal_box = gr.Markdown(visible=False) with gr.Tab("집계"): board = gr.Markdown() gr.Button("새로고침").click(leaderboard, outputs=board) demo.load(leaderboard, outputs=board) next_btn.click(start, inputs=[seen_state], outputs=[audio, item_state, heard_box, score_slider, reveal_box, submit_btn, next_btn]) submit_btn.click(submit, inputs=[item_state, heard_box, score_slider, seen_state, rater_state], outputs=[reveal_box, warn, submit_btn, next_btn, seen_state]) if __name__ == "__main__": demo.launch()