diff --git a/notebooks/luke-scribe-colab.ipynb b/notebooks/luke-scribe-colab.ipynb index 311af09..30e6425 100644 --- a/notebooks/luke-scribe-colab.ipynb +++ b/notebooks/luke-scribe-colab.ipynb @@ -109,19 +109,19 @@ "execution_count": null, "metadata": {}, "outputs": [], - "source": "# 7) 실전 전사 — GPU 자동 감지 + 모델 다운로드 (large-v3-turbo)\n# 첫 실행 시 Hugging Face에서 모델을 다운로드합니다 (turbo ≈ 1.6GB, 1~2분)\n# v0.1: 'cuda:N' device를 분리해 CTranslate2 계약에 맞게 전달하도록 수정됨\n# (기존: unsupported device cuda:0). 그래도 실패하면 CPU 폴백 안내가 출력됩니다.\nimport subprocess, json\nr = subprocess.run(\n ['luke-scribe', 'transcribe', 'samples/colab-ko-en.mp3', '--language', 'ko', '--device', 'auto'],\n capture_output=True, text=True,\n)\nprint(r.stdout[-2500:] if r.stdout else '')\nprint(r.stderr[-800:] if r.stderr else '')\n" + "source": "# 7) 실전 전사 — GPU 자동 감지 + 모델 다운로드 (large-v3-turbo)\n# 첫 실행 시 Hugging Face에서 모델을 다운로드합니다 (turbo ≈ 1.6GB, 1~2분)\n# v0.1: 'cuda:N' device를 분리해 CTranslate2 계약에 맞게 전달하도록 수정됨\n# (기존: unsupported device cuda:0). 그래도 실패하면 CPU 폴백 안내가 출력됩니다.\n# 후처리: 기본 rules가 vLLM→BLM 같은 흔한 오인식을 복원하고,\n# --glossary 'BLM=vLLM'으로 도메인 용어를 명시적으로 보강할 수 있다.\nimport subprocess, json\nr = subprocess.run(\n ['luke-scribe', 'transcribe', 'samples/colab-ko-en.mp3', '--language', 'ko', '--device', 'auto',\n '--glossary', 'BLM=vLLM'],\n capture_output=True, text=True,\n)\nprint(r.stdout[-2500:] if r.stdout else '')\nprint(r.stderr[-800:] if r.stderr else '')\n" }, { "cell_type": "markdown", "metadata": {}, - "source": "### 5-2) 후처리 검증\n\nglossary(오인식 용어 복원) + hotword(용어 사전 주입) 동작을 확인합니다.\n`--hotword vLLM Kubernetes`를 주면 initial_prompt에 용어가 주입되어 보존률이 올라갑니다." + "source": "### 5-2) 후처리 검증\n\n후처리(rules/glossary) + hotword(용어 사전 주입) 동작을 확인합니다.\n- **rules (기본)**: `BLM → vLLM`, `v l l m → vLLM` 같은 흔한 오인식을 결정적으로 복원.\n- **glossary**: `--glossary '오인식=표준'`으로 도메인 용어를 명시적으로 보강 (반복 가능).\n- **hotword**: `--hotword vLLM Kubernetes` → initial_prompt 주입으로 보존률 향상." }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], - "source": "# 8) hotword 포함 전사 (용어 보존 강화)\n!luke-scribe transcribe samples/colab-ko-en.mp3 --language ko --device auto --hotword vLLM --hotword Kubernetes\n" + "source": "# 8) hotword + glossary 포함 전사 (용어 보존 강화)\n!luke-scribe transcribe samples/colab-ko-en.mp3 --language ko --device auto --hotword vLLM --hotword Kubernetes --glossary BLM=vLLM\n" }, { "cell_type": "markdown", @@ -171,7 +171,7 @@ "execution_count": null, "metadata": {}, "outputs": [], - "source": "# 13) 벤치마크 — 샘플 manifest 사용 (선택, 시간 소요)\n# entities는 {canonical, surface, start_char, end_char} dict여야 한다\n# (문자열이면 entity_retention이 .get() 호출에 실패해 clip이 실패 처리됨)\nimport yaml\n\nREF_TEXT = '오늘은 vLLM 서버를 Kubernetes 클러스터에 배포하는 방법을 설명합니다. GPU 가속 추론으로 대기 시간을 줄일 수 있습니다.'\nwith open('/content/reference.txt', 'w', encoding='utf-8') as f:\n f.write(REF_TEXT)\n\nentities = []\nfor name in ('vLLM', 'Kubernetes', 'GPU'):\n idx = REF_TEXT.index(name)\n entities.append({\n 'canonical': name, 'surface': name,\n 'start_char': idx, 'end_char': idx + len(name),\n })\n\nmanifest = {\n 'name': 'colab-quick',\n 'dataset_version': '1.0',\n 'language': 'ko',\n 'targets': {'entity_preservation': 0.95, 'cer': 0.15},\n 'clips': [\n {'id': 'ko-en-tech', 'audio_path': 'samples/colab-ko-en.mp3',\n 'reference_path': '/content/reference.txt',\n 'duration_sec': 10.5, 'entities': entities},\n ],\n}\nyaml.safe_dump(manifest, open('/content/manifest.yaml', 'w'))\n\n# 실행 (기본: turbo만 → 빠름)\n!luke-scribe bench /content/manifest.yaml --models large-v3-turbo --device auto --repeats 2 --output /content/bench-report.json 2>&1 | tail -20\n" + "source": "# 13) 벤치마크 — 샘플 manifest 사용 (선택, 시간 소요)\n# entities는 {canonical, surface, start_char, end_char} dict여야 한다\n# (문자열이면 entity_retention이 .get() 호출에 실패해 clip이 실패 처리됨)\nimport yaml\n\nREF_TEXT = '오늘은 vLLM 서버를 Kubernetes 클러스터에 배포하는 방법을 설명합니다. GPU 가속 추론으로 대기 시간을 줄일 수 있습니다.'\nwith open('/content/reference.txt', 'w', encoding='utf-8') as f:\n f.write(REF_TEXT)\n\nentities = []\nfor name in ('vLLM', 'Kubernetes', 'GPU'):\n idx = REF_TEXT.index(name)\n entities.append({\n 'canonical': name, 'surface': name,\n 'start_char': idx, 'end_char': idx + len(name),\n })\n\nmanifest = {\n 'name': 'colab-quick',\n 'dataset_version': '1.0',\n 'language': 'ko',\n 'targets': {'entity_preservation': 0.95, 'cer': 0.15},\n # 벤치도 후처리를 적용해 실사용 지표를 측정 (vLLM→BLM 복원 포함)\n 'glossary': {'BLM': 'vLLM'},\n 'clips': [\n {'id': 'ko-en-tech', 'audio_path': 'samples/colab-ko-en.mp3',\n 'reference_path': '/content/reference.txt',\n 'duration_sec': 10.5, 'entities': entities},\n ],\n}\nyaml.safe_dump(manifest, open('/content/manifest.yaml', 'w'))\n\n# 실행 (기본: turbo만 → 빠름)\n!luke-scribe bench /content/manifest.yaml --models large-v3-turbo --device auto --repeats 2 --output /content/bench-report.json 2>&1 | tail -20\n" }, { "cell_type": "markdown", diff --git a/scripts/build_colab_notebook.py b/scripts/build_colab_notebook.py index 0311cb1..ab05e66 100644 --- a/scripts/build_colab_notebook.py +++ b/scripts/build_colab_notebook.py @@ -192,9 +192,12 @@ def cells() -> list[dict]: "# 첫 실행 시 Hugging Face에서 모델을 다운로드합니다 (turbo ≈ 1.6GB, 1~2분)\n" "# v0.1: 'cuda:N' device를 분리해 CTranslate2 계약에 맞게 전달하도록 수정됨\n" "# (기존: unsupported device cuda:0). 그래도 실패하면 CPU 폴백 안내가 출력됩니다.\n" + "# 후처리: 기본 rules가 vLLM→BLM 같은 흔한 오인식을 복원하고,\n" + "# --glossary 'BLM=vLLM'으로 도메인 용어를 명시적으로 보강할 수 있다.\n" "import subprocess, json\n" "r = subprocess.run(\n" - " ['luke-scribe', 'transcribe', 'samples/colab-ko-en.mp3', '--language', 'ko', '--device', 'auto'],\n" + " ['luke-scribe', 'transcribe', 'samples/colab-ko-en.mp3', '--language', 'ko', '--device', 'auto',\n" + " '--glossary', 'BLM=vLLM'],\n" " capture_output=True, text=True,\n" ")\n" "print(r.stdout[-2500:] if r.stdout else '')\n" @@ -202,12 +205,14 @@ def cells() -> list[dict]: ), md( "### 5-2) 후처리 검증\n\n" - "glossary(오인식 용어 복원) + hotword(용어 사전 주입) 동작을 확인합니다.\n" - "`--hotword vLLM Kubernetes`를 주면 initial_prompt에 용어가 주입되어 보존률이 올라갑니다." + "후처리(rules/glossary) + hotword(용어 사전 주입) 동작을 확인합니다.\n" + "- **rules (기본)**: `BLM → vLLM`, `v l l m → vLLM` 같은 흔한 오인식을 결정적으로 복원.\n" + "- **glossary**: `--glossary '오인식=표준'`으로 도메인 용어를 명시적으로 보강 (반복 가능).\n" + "- **hotword**: `--hotword vLLM Kubernetes` → initial_prompt 주입으로 보존률 향상." ), code( - "# 8) hotword 포함 전사 (용어 보존 강화)\n" - "!luke-scribe transcribe samples/colab-ko-en.mp3 --language ko --device auto --hotword vLLM --hotword Kubernetes\n" + "# 8) hotword + glossary 포함 전사 (용어 보존 강화)\n" + "!luke-scribe transcribe samples/colab-ko-en.mp3 --language ko --device auto --hotword vLLM --hotword Kubernetes --glossary BLM=vLLM\n" ), md( "## 6) REST API 스모크\n\n" @@ -454,6 +459,8 @@ def cells() -> list[dict]: " 'dataset_version': '1.0',\n" " 'language': 'ko',\n" " 'targets': {'entity_preservation': 0.95, 'cer': 0.15},\n" + " # 벤치도 후처리를 적용해 실사용 지표를 측정 (vLLM→BLM 복원 포함)\n" + " 'glossary': {'BLM': 'vLLM'},\n" " 'clips': [\n" " {'id': 'ko-en-tech', 'audio_path': 'samples/colab-ko-en.mp3',\n" " 'reference_path': '/content/reference.txt',\n" diff --git a/src/luke_scribe/benchmark/runner.py b/src/luke_scribe/benchmark/runner.py index 32387e6..71133a5 100644 --- a/src/luke_scribe/benchmark/runner.py +++ b/src/luke_scribe/benchmark/runner.py @@ -56,6 +56,8 @@ def run_benchmark( from ..errors import InvalidInput raise InvalidInput("manifest에 clips가 없습니다") + # manifest 최상위 glossary: {오인식 패턴: 표준 표기} — 벤치도 후처리를 적용한다 + glossary = data.get("glossary") or None report: dict = { "report_version": REPORT_VERSION, @@ -71,6 +73,8 @@ def run_benchmark( "beam_size": 5, "temperature": 0.0, "vad_filter": True, + "post_mode": settings.post_mode if settings.post_enabled else "none", + "glossary": glossary or {}, }, "models": [], "decision": {"status": "pending", "default_model": None, "reasons": []}, @@ -81,7 +85,9 @@ def run_benchmark( owner = EngineOwner.get(settings) model_results = {} for model in models: - agg = _run_model(owner, clips, model, device, compute_type, repeats, hotwords, report) + agg = _run_model( + owner, clips, model, device, compute_type, repeats, hotwords, report, settings, glossary + ) model_results[model] = agg report["models"].append(agg["summary"]) @@ -102,7 +108,9 @@ def run_benchmark( return report -def _run_model(owner, clips, model, device, compute_type, repeats, hotwords, report) -> dict: +def _run_model( + owner, clips, model, device, compute_type, repeats, hotwords, report, settings, glossary +) -> dict: from ..engine.base import TranscriptionOptions options = TranscriptionOptions( @@ -114,7 +122,7 @@ def _run_model(owner, clips, model, device, compute_type, repeats, hotwords, rep ) # warm-up (비평가 클립 1회) clip0 = clips[0] - _transcribe_clip(owner, options, clip0) + _transcribe_clip(owner, options, clip0, settings, glossary) rtf_samples: list[float] = [] rss_samples: list[float] = [] @@ -141,7 +149,7 @@ def _run_model(owner, clips, model, device, compute_type, repeats, hotwords, rep clip_success = False for _ in range(repeats): try: - result = _transcribe_clip(owner, options, clip) + result = _transcribe_clip(owner, options, clip, settings, glossary) m = clip_metrics(ref_text, result["text"], entities) clip_rtfs.append(result["rtf"]) rss_samples.append(peak_process_rss_mb()) @@ -185,9 +193,15 @@ def _run_model(owner, clips, model, device, compute_type, repeats, hotwords, rep return {"summary": summary} -def _transcribe_clip(owner, options, clip) -> dict: - """클립 전사 — 세그먼트 소비 + text/rtf 반환.""" +def _transcribe_clip(owner, options, clip, settings, glossary) -> dict: + """클립 전사 — 후처리(glossary/rules) 적용 + text/rtf 반환. + + 벤치 지표는 원시 전사가 아니라 사용자가 실제로 받는 후처리 결과를 측정해야 + 한다 (vLLM→BLM 같은 오인식은 rules/glossary에서 복원된다). + """ from ..engine.owner import InferenceRequest + from ..postprocess.pipeline import run_postprocess + from ..results.models import Segment audio_path = clip.get("audio_path") if not audio_path: @@ -196,11 +210,22 @@ def _transcribe_clip(owner, options, clip) -> dict: t0 = time.time() req = InferenceRequest(audio_path=audio_path, options=options, lane="batch") outcome = owner.transcribe(req) - texts = [] - for seg in outcome["segments"]: - texts.append(seg.get("text", "")) + segments: list[Segment] = [] + for idx, seg in enumerate(outcome["segments"]): + segments.append( + Segment( + index=idx, + start=float(seg.get("start", 0.0)), + end=float(seg.get("end", 0.0)), + text=seg.get("text", ""), + avg_logprob=seg.get("avg_logprob"), + no_speech_prob=seg.get("no_speech_prob"), + ) + ) + post = run_postprocess(segments, options, settings, glossary=glossary) + text = " ".join(s.text.strip() for s in post["segments"] if s.text.strip()) elapsed = time.time() - t0 - return {"text": " ".join(t for t in texts if t), "rtf": elapsed / duration} + return {"text": text, "rtf": elapsed / duration} def _decide(model_results: dict) -> dict: diff --git a/src/luke_scribe/cli.py b/src/luke_scribe/cli.py index 2fb038f..c1fbc00 100644 --- a/src/luke_scribe/cli.py +++ b/src/luke_scribe/cli.py @@ -86,6 +86,9 @@ def transcribe( ), vad: bool = typer.Option(True, "--vad/--no-vad"), hotword: list[str] = typer.Option([], "--hotword", help="반복 가능"), + glossary: list[str] = typer.Option( + [], "--glossary", help="오인식 패턴=표준 표기, 반복 가능 (예: --glossary BLM=vLLM)" + ), output: Path | None = typer.Option(None, "--output", "-", help="결과 파일 (기본 stdout)"), force: bool = typer.Option(False, "--force", help="기존 출력 파일 overwrite"), word_timestamps: bool = typer.Option(False, "--word-timestamps"), @@ -113,9 +116,17 @@ def transcribe( hotwords=hotword, word_timestamps=word_timestamps, ) + glossary_dict: dict[str, str] = {} + for item in glossary: + if "=" not in item: + _fail(EXIT_INPUT, f"--glossary는 'KEY=VALUE' 형식이어야 합니다: {item}") + key, _, value = item.partition("=") + glossary_dict[key.strip()] = value.strip() try: pipeline = BatchPipeline(settings=settings, token=token) - result = pipeline.run(source, options, source_name=source.name) + result = pipeline.run( + source, options, source_name=source.name, glossary=glossary_dict or None + ) except LukeScribeError as exc: if output is not None: failed = TranscriptResult( diff --git a/src/luke_scribe/jobqueue/worker.py b/src/luke_scribe/jobqueue/worker.py index d7db075..b3622b2 100644 --- a/src/luke_scribe/jobqueue/worker.py +++ b/src/luke_scribe/jobqueue/worker.py @@ -110,7 +110,11 @@ class Worker: if k in TranscriptionOptions.__slots__ } ) - result = pipeline.run(job, options, progress_cb=progress_cb) + # API 옵션의 glossary/post_correction(dict)을 후처리 glossary로 전달 + glossary = (job.options or {}).get("glossary") or (job.options or {}).get( + "post_correction" + ) + result = pipeline.run(job, options, progress_cb=progress_cb, glossary=glossary) # 결과를 먼저 영속화한 뒤 상태 전이 (실패 시 FAILED로 전이 가능하게) self.store.write_result(job.id, result) current = self._transition(job, JobStatus.COMPLETED) diff --git a/src/luke_scribe/pipeline/batch.py b/src/luke_scribe/pipeline/batch.py index 61824c9..657909e 100644 --- a/src/luke_scribe/pipeline/batch.py +++ b/src/luke_scribe/pipeline/batch.py @@ -54,9 +54,13 @@ class BatchPipeline: *, progress_cb: Callable[[float, float], None] | None = None, source_name: str | None = None, + glossary: dict[str, str] | None = None, ) -> TranscriptResult: """job(Job) 또는 source(Path)를 받아 전사. + Args: + glossary: {오인식 패턴: 표준 표기} — run_postprocess에 전달. + Returns: completed TranscriptResult (후처리 포함). """ @@ -116,7 +120,7 @@ class BatchPipeline: # 4) 후처리 (glossary/rules/LLM/confidence) t2 = time.time() - post_result = run_postprocess(segments, options, self.settings) + post_result = run_postprocess(segments, options, self.settings, glossary=glossary) postprocess_sec = time.time() - t2 text = " ".join(s.text.strip() for s in post_result["segments"] if s.text.strip()) diff --git a/src/luke_scribe/postprocess/rules.py b/src/luke_scribe/postprocess/rules.py index 1d49d99..b936128 100644 --- a/src/luke_scribe/postprocess/rules.py +++ b/src/luke_scribe/postprocess/rules.py @@ -13,6 +13,8 @@ from ..results.models import Segment # 흔한 오인식 패턴 → 표준 표기 (정규식) DEFAULT_RULES: list[tuple[re.Pattern, str]] = [ (re.compile(r"\bv ?l ?l ?m\b", re.IGNORECASE), "vLLM"), + # 흔한 오인식: vLLM → BLM (GPU 실전에서 재현). 기술 STT 도메인 전제로 복원. + (re.compile(r"\bblm\b", re.IGNORECASE), "vLLM"), (re.compile(r"\bk ?u ?b ?e ?r ?n ?e ?t ?e ?s\b", re.IGNORECASE), "Kubernetes"), (re.compile(r"\bf ?a ?s ?t ?a ?p ?i\b", re.IGNORECASE), "FastAPI"), (re.compile(r"\bg ?p ?u\b", re.IGNORECASE), "GPU"), diff --git a/tests/unit/test_benchmark.py b/tests/unit/test_benchmark.py new file mode 100644 index 0000000..af354b0 --- /dev/null +++ b/tests/unit/test_benchmark.py @@ -0,0 +1,143 @@ +"""벤치마크 단위 테스트 — 후처리 적용 + glossary 지원. + +벤치 지표는 원시 전사가 아니라 후처리(rules/glossary)를 거친 결과를 측정해야 +한다 (vLLM→BLM 같은 오인식 복원 포함). +""" + +from __future__ import annotations + +from luke_scribe.benchmark.runner import _run_model, _transcribe_clip +from luke_scribe.config import Settings +from luke_scribe.engine.base import TranscriptionOptions + +REF_TEXT = "오늘은 vLLM 서버를 Kubernetes 클러스터에 배포합니다" +ENTITIES = [ + {"canonical": "vLLM", "surface": "vLLM", "start_char": 4, "end_char": 8}, + {"canonical": "Kubernetes", "surface": "Kubernetes", "start_char": 13, "end_char": 23}, +] + + +class _FakeOwner: + """세그먼트 dict를 반환하는 가짜 owner (dict 계약 사용).""" + + def __init__(self, texts: list[str]) -> None: + self._texts = texts + + def transcribe(self, req): # noqa: ANN001 + segs = [ + { + "index": i, + "start": i * 2.0, + "end": i * 2.0 + 2.0, + "text": t, + "avg_logprob": -0.2, + "no_speech_prob": 0.01, + } + for i, t in enumerate(self._texts) + ] + return { + "segments": iter(segs), + "device": "cpu", + "compute_type": "int8", + "attempted_profiles": [{}], + "info": {"language": "ko"}, + } + + +class TestTranscribeClipPostprocess: + def test_rules_fix_blm_to_vllm(self): + owner = _FakeOwner(["오늘은 BLM 서버를 Kubernetes 클러스터에 배포합니다"]) + settings = Settings(_env_file=None, post_mode="rules", post_enabled=True) + clip = {"audio_path": "x.mp3", "duration_sec": 10.0} + out = _transcribe_clip(owner, TranscriptionOptions(), clip, settings, None) + assert "vLLM" in out["text"] + assert "BLM" not in out["text"] + + def test_glossary_applied(self): + owner = _FakeOwner(["브이엘엘엠 서버"]) + settings = Settings(_env_file=None, post_mode="glossary", post_enabled=True) + clip = {"audio_path": "x.mp3", "duration_sec": 10.0} + out = _transcribe_clip( + owner, TranscriptionOptions(), clip, settings, {"브이엘엘엠": "vLLM"} + ) + assert "vLLM" in out["text"] + + def test_postprocess_disabled_keeps_raw(self): + owner = _FakeOwner(["오늘은 BLM 서버"]) + settings = Settings(_env_file=None, post_mode="rules", post_enabled=False) + clip = {"audio_path": "x.mp3", "duration_sec": 10.0} + out = _transcribe_clip(owner, TranscriptionOptions(), clip, settings, None) + assert "BLM" in out["text"] + + +class TestRunModelPostprocess: + def _report(self) -> dict: + return {"run_config": {"hotword_set": []}} + + def _clip(self, ref_path: str) -> dict: + return { + "id": "c1", + "audio_path": "x.mp3", + "reference_path": ref_path, + "duration_sec": 10.0, + "entities": ENTITIES, + } + + def test_entity_retention_full_with_rules(self, tmp_path): + ref = tmp_path / "ref.txt" + ref.write_text(REF_TEXT, encoding="utf-8") + owner = _FakeOwner(["오늘은 BLM 서버를 Kubernetes 클러스터에 배포합니다"]) + settings = Settings(_env_file=None, post_mode="rules", post_enabled=True) + agg = _run_model( + owner, + [self._clip(str(ref))], + "large-v3-turbo", + "cpu", + "int8", + 1, + [], + self._report(), + settings, + None, + ) + assert agg["summary"]["entity_retention"] == 1.0 + assert agg["summary"]["failure_rate"] == 0.0 + + def test_glossary_raises_entity_retention(self, tmp_path): + ref = tmp_path / "ref.txt" + ref.write_text(REF_TEXT, encoding="utf-8") + owner = _FakeOwner(["오늘은 BLM 서버를 Kubernetes 클러스터에 배포합니다"]) + settings = Settings(_env_file=None, post_mode="glossary", post_enabled=True) + agg = _run_model( + owner, + [self._clip(str(ref))], + "large-v3-turbo", + "cpu", + "int8", + 1, + [], + self._report(), + settings, + {"BLM": "vLLM"}, + ) + assert agg["summary"]["entity_retention"] == 1.0 + + def test_raw_text_fails_entity_retention_without_postprocess(self, tmp_path): + # 후처리 없이 raw 전사("BLM")를 측정하면 vLLM 엔티티가 보존되지 않는다 + ref = tmp_path / "ref.txt" + ref.write_text(REF_TEXT, encoding="utf-8") + owner = _FakeOwner(["오늘은 BLM 서버를 Kubernetes 클러스터에 배포합니다"]) + settings = Settings(_env_file=None, post_mode="none", post_enabled=False) + agg = _run_model( + owner, + [self._clip(str(ref))], + "large-v3-turbo", + "cpu", + "int8", + 1, + [], + self._report(), + settings, + None, + ) + assert agg["summary"]["entity_retention"] < 1.0 diff --git a/tests/unit/test_postprocess.py b/tests/unit/test_postprocess.py index 084a6f7..0caaef7 100644 --- a/tests/unit/test_postprocess.py +++ b/tests/unit/test_postprocess.py @@ -43,6 +43,25 @@ class TestRules: out = apply_rules(segs) assert "vLLM" in out["segments"][0].text + def test_vllm_restored_from_blm(self): + # GPU 실전에서 재현된 오인식: vLLM → BLM + segs = _segments(["오늘은 BLM 서버를 배포합니다"]) + out = apply_rules(segs) + assert "vLLM" in out["segments"][0].text + assert "BLM" not in out["segments"][0].text + + def test_blm_boundary_required(self): + # 단어 경계가 없으면 교정하지 않는다 (부분 문자열 보호) + segs = _segments(["sublm 단어"]) + out = apply_rules(segs) + assert "sublm" in out["segments"][0].text + assert "vLLM" not in out["segments"][0].text + + def test_blm_case_insensitive(self): + segs = _segments(["blm 서버"]) + out = apply_rules(segs) + assert "vLLM" in out["segments"][0].text + def test_whitespace_collapse(self): segs = _segments(["오늘 API 서버"]) out = apply_rules(segs)