From f9fd86464d6b73b9d2ddedd0282fe2c233194ca8 Mon Sep 17 00:00:00 2001 From: EJClaw Date: Fri, 25 Sep 2026 23:01:31 +0900 Subject: [PATCH] =?UTF-8?q?fix:=20.gitignore=20=EA=B0=80=20=EC=86=8C?= =?UTF-8?q?=EC=8A=A4=20=ED=8C=A8=ED=82=A4=EC=A7=80=20src/livesub/models=20?= =?UTF-8?q?=EB=A5=BC=20=ED=86=B5=EC=A7=B8=EB=A1=9C=20=EB=A8=B9=EA=B3=A0=20?= =?UTF-8?q?=EC=9E=88=EC=97=88=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `models/` 처럼 슬래시 없이 쓴 규칙은 깊이에 상관없이 같은 이름의 폴더를 전부 무시한다. 의도는 루트의 모델 캐시였는데 소스 패키지까지 같이 걸렸고, 그래서 asr/translator/manager/tiers/glossary/packs/speech_level 8개 파일이 한 번도 커밋된 적이 없다. 증상: Gitea 에서 clone 하면 `ModuleNotFoundError: No module named 'livesub.models'` 로 앱이 아예 임포트되지 않는다. 로컬 작업 트리에는 파일이 있으니 개발 중에는 절대 안 보인다. 방금 붙인 Windows CI VM 이 첫 검증에서 바로 잡아냈다. /models/, /adapters/, /data/ 로 루트에 고정하고 빠진 파일을 추가한다. --- .gitignore | 11 +- src/livesub/models/__init__.py | 47 ++++ src/livesub/models/asr.py | 95 ++++++++ src/livesub/models/glossary.py | 235 ++++++++++++++++++ src/livesub/models/manager.py | 242 +++++++++++++++++++ src/livesub/models/packs.py | 152 ++++++++++++ src/livesub/models/speech_level.py | 310 ++++++++++++++++++++++++ src/livesub/models/tiers.py | 209 ++++++++++++++++ src/livesub/models/translator.py | 376 +++++++++++++++++++++++++++++ 9 files changed, 1674 insertions(+), 3 deletions(-) create mode 100644 src/livesub/models/__init__.py create mode 100644 src/livesub/models/asr.py create mode 100644 src/livesub/models/glossary.py create mode 100644 src/livesub/models/manager.py create mode 100644 src/livesub/models/packs.py create mode 100644 src/livesub/models/speech_level.py create mode 100644 src/livesub/models/tiers.py create mode 100644 src/livesub/models/translator.py diff --git a/.gitignore b/.gitignore index ea53f71..2eb8840 100644 --- a/.gitignore +++ b/.gitignore @@ -15,9 +15,14 @@ src/hearo/resources/bin/*.exe src/hearo/resources/bin/*.pdb # 모델 캐시 / 학습 산출물 / 사용자 데이터 -models/ -adapters/ -data/ +# +# ⚠️ 반드시 슬래시로 시작해 저장소 루트에 고정할 것. `models/` 로 쓰면 깊이에 +# 상관없이 같은 이름의 폴더를 전부 먹는다. 실제로 이것 때문에 소스 패키지 +# src/livesub/models/ 8개 파일이 한 번도 커밋되지 않았고, Gitea 에서 clone 한 +# 쪽은 앱이 아예 임포트되지 않았다. Windows CI VM 첫 검증에서 잡혔다. +/models/ +/adapters/ +/data/ *.jsonl config.json glossary.json diff --git a/src/livesub/models/__init__.py b/src/livesub/models/__init__.py new file mode 100644 index 0000000..22c901c --- /dev/null +++ b/src/livesub/models/__init__.py @@ -0,0 +1,47 @@ +"""모델 레이어 (음성인식 · 번역 · 용어집).""" + +from .asr import SpeechRecognizer, Transcript +from .glossary import Glossary, GlossaryEntry +from .manager import GpuInfo, ModelManager, detect_gpu +from .tiers import ( + DEFAULT_TIER, + TIERS, + MTBackend, + PromptStyle, + Tier, + get_tier, + ordered_tiers, +) +from .speech_level import ( + Politeness, + SpeechLevel, + apply_speech_level, + detect_politeness, + to_casual, +) +from .translator import Translator, build_seedx_prompt, create_translator + +__all__ = [ + "DEFAULT_TIER", + "Glossary", + "GlossaryEntry", + "GpuInfo", + "MTBackend", + "ModelManager", + "Politeness", + "PromptStyle", + "SpeechLevel", + "SpeechRecognizer", + "TIERS", + "Tier", + "Transcript", + "Translator", + "apply_speech_level", + "build_seedx_prompt", + "detect_politeness", + "create_translator", + "detect_gpu", + "get_tier", + "ordered_tiers", + "to_casual", +] diff --git a/src/livesub/models/asr.py b/src/livesub/models/asr.py new file mode 100644 index 0000000..6d0a01a --- /dev/null +++ b/src/livesub/models/asr.py @@ -0,0 +1,95 @@ +"""음성인식(ASR) 래퍼 — faster-whisper(CTranslate2) 기반.""" + +from __future__ import annotations + +import logging +import threading +from dataclasses import dataclass + +import numpy as np + +from ..constants import models_dir +from .tiers import AsrSpec + +log = logging.getLogger(__name__) + + +@dataclass +class Transcript: + text: str + language: str + confidence: float = 0.0 + duration_s: float = 0.0 + + +class SpeechRecognizer: + """faster-whisper 모델 한 개를 감싼다. 스레드 안전.""" + + def __init__(self, spec: AsrSpec, device: str = "cuda", device_index: int = 0) -> None: + self.spec = spec + self.device = device + self.device_index = device_index + self._model = None + self._lock = threading.Lock() + + @property + def loaded(self) -> bool: + return self._model is not None + + def load(self) -> None: + if self._model is not None: + return + from faster_whisper import WhisperModel + + compute_type = self.spec.compute_type + if self.device == "cpu" and compute_type.endswith("float16"): + compute_type = "int8" # CPU에서는 float16이 오히려 느리다 + log.info("음성인식 모델 로드: %s (%s/%s)", self.spec.repo, self.device, compute_type) + self._model = WhisperModel( + self.spec.repo, + device=self.device, + device_index=self.device_index, + compute_type=compute_type, + download_root=str(models_dir()), + ) + + def unload(self) -> None: + with self._lock: + self._model = None + + def transcribe( + self, + audio: np.ndarray, + language: str | None = None, + fast: bool = False, + prompt: str = "", + ) -> Transcript: + """오디오 한 덩어리를 텍스트로. + + `fast=True` 는 중간 결과용으로 beam search를 끄고 빠르게 돌린다. + """ + if self._model is None: + self.load() + assert self._model is not None + + with self._lock: + segments, info = self._model.transcribe( + audio.astype(np.float32, copy=False), + language=language, + beam_size=1 if fast else self.spec.beam_size, + temperature=0.0, + condition_on_previous_text=False, + initial_prompt=prompt or None, + vad_filter=not fast, + vad_parameters={"min_silence_duration_ms": 300}, + word_timestamps=False, + ) + parts = [s.text for s in segments] + + text = " ".join(p.strip() for p in parts if p.strip()).strip() + return Transcript( + text=text, + language=getattr(info, "language", language or "") or "", + confidence=float(getattr(info, "language_probability", 0.0) or 0.0), + duration_s=float(getattr(info, "duration", 0.0) or 0.0), + ) diff --git a/src/livesub/models/glossary.py b/src/livesub/models/glossary.py new file mode 100644 index 0000000..b38726f --- /dev/null +++ b/src/livesub/models/glossary.py @@ -0,0 +1,235 @@ +"""용어집(고유명사·게임/방송 용어) 처리. + +추가학습 없이도 즉시 효과를 내는 레이어다. 두 가지 방식으로 동작한다. + +1. 플레이스홀더 보호 (`protect` / `restore`) + 원문의 등록 용어를 `⟦0⟧` 같은 토큰으로 치환해 번역 모델이 건드리지 못하게 하고, + 번역이 끝난 뒤 지정한 역어로 되돌린다. 어떤 모델에서도 쓸 수 있다. +2. 프롬프트 주입 (`prompt_hint`) + 해당 문장에 등장한 용어만 골라 프롬프트에 "이 용어는 이렇게 옮겨라"로 넣는다. + 조사·어미까지 문맥에 맞게 붙어 결과가 더 자연스럽다. + +어느 쪽을 쓸지는 모델이 정한다. 2번은 **지시문을 이해하는 instruct 모델에서만** +동작한다. Seed-X 처럼 번역만 하도록 학습된 completion 모델은 지시문을 넣으면 +학습 분포를 벗어나 오히려 번역이 망가지므로 1번을 쓴다. NLLB 같은 seq2seq 도 +마찬가지다. 판정은 `MTSpec.supports_prompt_glossary` 가 한다. +""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass, field +from pathlib import Path + +#: 번역 모델이 원문 그대로 통과시켜야 하는 자리표시자. +#: +#: 형식 선택은 취향이 아니라 실측 결과다. 처음에 쓰던 `⟦0⟧` 는 NLLB 가 +#: 괄호를 통째로 날려 `0` 만 남겼고(생존률 0/3), 그 결과 용어가 자막에서 +#: 사라졌다. 후보 8종을 실제 모델로 돌려 비교한 끝에 `#0#` 를 골랐다 +#: (생존률 3/3, 다중 치환 4/5). 입력이 음성인식 결과라 `#` 가 자연스럽게 +#: 등장할 일이 거의 없다는 점도 같이 봤다. +PLACEHOLDER = "#{}#" # #0# +PLACEHOLDER_RE = re.compile(r"#(\d+)#") + + +@dataclass +class GlossaryEntry: + """용어 하나. `targets` 는 언어코드 → 역어.""" + + source: str + targets: dict[str, str] = field(default_factory=dict) + note: str = "" + #: 비우면 모든 소스 언어에 적용 + source_lang: str = "" + #: 문장 전체가 이 표현일 때만 치환한다. + #: + #: "get in the car" 같은 절을 긴 문장 한가운데서 통째로 치환하면 + #: 번역 모델이 문법을 세울 근거를 잃어 "어부 들어오는, 차 타!" 같은 + #: 결과가 나온다. 반대로 그 말만 단독으로 나왔을 때는 통째로 바꾸는 + #: 것이 가장 자연스럽다. 그래서 적용 범위를 문장 전체로 한정한다. + whole_only: bool = False + + def target_for(self, lang: str) -> str: + return self.targets.get(lang, "") + + +class Glossary: + def __init__(self, entries: list[GlossaryEntry] | None = None, case_sensitive: bool = False): + self.case_sensitive = case_sensitive + self.entries: list[GlossaryEntry] = list(entries or []) + self._pattern: re.Pattern[str] | None = None + self._index: dict[str, GlossaryEntry] = {} + self._rebuild() + + # --- 영속화 -------------------------------------------------------- + @classmethod + def load(cls, path: str | Path, case_sensitive: bool = False) -> "Glossary": + p = Path(path) + if not p.is_file(): + return cls(case_sensitive=case_sensitive) + raw = json.loads(p.read_text(encoding="utf-8")) + entries = [ + GlossaryEntry( + source=item["source"], + targets=dict(item.get("targets", {})), + note=item.get("note", ""), + source_lang=item.get("source_lang", ""), + whole_only=bool(item.get("whole_only", False)), + ) + for item in raw.get("entries", []) + if item.get("source") + ] + return cls(entries, case_sensitive=raw.get("case_sensitive", case_sensitive)) + + def save(self, path: str | Path) -> Path: + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + payload = { + "case_sensitive": self.case_sensitive, + "entries": [ + { + "source": e.source, + "targets": e.targets, + "note": e.note, + "source_lang": e.source_lang, + "whole_only": e.whole_only, + } + for e in self.entries + ], + } + p.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") + return p + + # --- 편집 ---------------------------------------------------------- + def add(self, entry: GlossaryEntry) -> None: + key = self._key(entry.source) + self.entries = [e for e in self.entries if self._key(e.source) != key] + self.entries.append(entry) + self._rebuild() + + def remove(self, source: str) -> None: + key = self._key(source) + self.entries = [e for e in self.entries if self._key(e.source) != key] + self._rebuild() + + def __len__(self) -> int: + return len(self.entries) + + # --- 사용 ---------------------------------------------------------- + def matches(self, text: str, target_lang: str) -> list[GlossaryEntry]: + """문장에 실제로 등장하고 해당 언어 역어가 있는 항목만.""" + if not self._pattern or not text: + return [] + found: list[GlossaryEntry] = [] + seen: set[str] = set() + for m in self._pattern.finditer(text): + entry = self._index.get(self._key(m.group(0))) + if entry is None or not entry.target_for(target_lang): + continue + key = self._key(entry.source) + if key not in seen: + seen.add(key) + found.append(entry) + return found + + def protect(self, text: str, target_lang: str) -> tuple[str, list[str]]: + """등록 용어를 자리표시자로 치환. (치환된 문장, 역어 목록) 반환.""" + replacements: list[str] = [] + if not text: + return text, replacements + + whole = self._match_whole(text, target_lang) + if whole is not None: + # 문장 전체가 등록된 표현 — 통째로 바꾸는 게 가장 자연스럽다. + return PLACEHOLDER.format(0), [whole] + + if not self._pattern: + return text, replacements + + def _sub(m: re.Match[str]) -> str: + entry = self._index.get(self._key(m.group(0))) + if entry is None: + return m.group(0) + target = entry.target_for(target_lang) + if not target: + return m.group(0) + replacements.append(target) + return PLACEHOLDER.format(len(replacements) - 1) + + return self._pattern.sub(_sub, text), replacements + + @staticmethod + def surviving_placeholders(text: str) -> set[int]: + """번역 결과에 살아남은 자리표시자 번호.""" + return {int(m) for m in PLACEHOLDER_RE.findall(text)} + + @classmethod + def missing_placeholders(cls, text: str, replacements: list[str]) -> list[int]: + """번역 중에 사라진 자리표시자 번호. + + 번역 모델이 문장 일부를 통째로 누락하면 자리표시자도 같이 사라진다. + 그대로 복원하면 해당 용어가 자막에서 증발하므로, 호출부가 이걸로 + 확인해 용어집 없이 다시 번역하는 쪽을 택할 수 있게 한다. + """ + present = cls.surviving_placeholders(text) + return [i for i in range(len(replacements)) if i not in present] + + @staticmethod + def restore(text: str, replacements: list[str]) -> str: + """플레이스홀더를 역어로 되돌린다.""" + if not replacements: + return text + + def _sub(m: re.Match[str]) -> str: + idx = int(m.group(1)) + return replacements[idx] if 0 <= idx < len(replacements) else "" + + return PLACEHOLDER_RE.sub(_sub, text) + + def prompt_hint(self, text: str, target_lang: str, limit: int = 12) -> str: + """LLM 프롬프트에 붙일 용어 지시문. 해당 없으면 빈 문자열.""" + hits = self.matches(text, target_lang)[:limit] + if not hits: + return "" + lines = [f"- {e.source} -> {e.target_for(target_lang)}" for e in hits] + return "다음 용어는 반드시 지정된 역어로 번역하세요:\n" + "\n".join(lines) + + def _match_whole(self, text: str, target_lang: str) -> str | None: + """문장 전체가 등록 표현과 같으면 그 역어를 돌려준다.""" + stripped = self._key(text.strip().rstrip(".!?…~,")) + entry = self._index.get(stripped) + if entry is None: + return None + return entry.target_for(target_lang) or None + + # --- 내부 ---------------------------------------------------------- + def _key(self, text: str) -> str: + return text if self.case_sensitive else text.casefold() + + def _rebuild(self) -> None: + self._index = {self._key(e.source): e for e in self.entries} + if not self.entries: + self._pattern = None + return + # whole_only 항목은 부분 매칭에서 제외한다 (문장 전체일 때만 쓴다). + partial = [e.source for e in self.entries if not e.whole_only] + if not partial: + self._pattern = None + return + # 긴 용어를 먼저 매칭해야 "헤드샷"이 "헤드"에 잡아먹히지 않는다. + sources = sorted(partial, key=len, reverse=True) + flags = 0 if self.case_sensitive else re.IGNORECASE + self._pattern = re.compile( + "|".join(_bounded(re.escape(s)) for s in sources), flags + ) + + +def _bounded(escaped: str) -> str: + """영문/숫자로 시작·끝나는 용어에만 단어 경계를 건다. + + 한국어·일본어·중국어에는 \\b 가 의미 없으므로 적용하지 않는다. + """ + prefix = r"\b" if re.match(r"[A-Za-z0-9]", escaped.replace("\\", "")) else "" + suffix = r"\b" if re.search(r"[A-Za-z0-9]$", escaped.replace("\\", "")) else "" + return f"{prefix}{escaped}{suffix}" diff --git a/src/livesub/models/manager.py b/src/livesub/models/manager.py new file mode 100644 index 0000000..c74aa91 --- /dev/null +++ b/src/livesub/models/manager.py @@ -0,0 +1,242 @@ +"""티어에 맞는 ASR + MT 모델 쌍을 관리한다.""" + +from __future__ import annotations + +import logging +import threading +from dataclasses import dataclass, replace + +from ..config import ModelConfig +from .asr import SpeechRecognizer +from .tiers import TIERS, MTBackend, Tier, get_tier +from .translator import Translator, create_translator + +log = logging.getLogger(__name__) + + +@dataclass +class GpuInfo: + available: bool + name: str = "" + total_vram_mb: int = 0 + free_vram_mb: int = 0 + reason: str = "" + + +def detect_gpu() -> GpuInfo: + """CUDA 사용 가능 여부와 VRAM. torch 가 없으면 nvidia-smi 로 대체 조회.""" + try: + import torch + + if torch.cuda.is_available(): + idx = torch.cuda.current_device() + free, total = torch.cuda.mem_get_info(idx) + return GpuInfo( + available=True, + name=torch.cuda.get_device_name(idx), + total_vram_mb=total // (1024 * 1024), + free_vram_mb=free // (1024 * 1024), + ) + return GpuInfo(False, reason="CUDA를 사용할 수 없습니다 (드라이버/빌드 확인).") + except ImportError: + pass + + import shutil + import subprocess + + smi = shutil.which("nvidia-smi") + if not smi: + return GpuInfo(False, reason="PyTorch와 nvidia-smi 모두 찾지 못했습니다.") + try: + out = subprocess.run( + [smi, "--query-gpu=name,memory.total,memory.free", + "--format=csv,noheader,nounits"], + capture_output=True, text=True, timeout=5, check=True, + ).stdout.strip().splitlines() + name, total, free = (v.strip() for v in out[0].split(",")) + return GpuInfo(True, name, int(total), int(free), + reason="PyTorch 미설치 — 정보만 조회했습니다.") + except Exception as exc: # noqa: BLE001 + return GpuInfo(False, reason=f"GPU 조회 실패: {exc}") + + +def apply_vram_limit(ratio: float, device_index: int = 0) -> bool: + """이 프로세스가 쓸 수 있는 VRAM 상한을 건다. + + 게임과 동시에 돌 때 가장 큰 문제는 연산 시간이 아니라 **VRAM 고갈**이다. + 게임이 쓸 메모리를 우리가 먹어버리면 텍스처 스트리밍이 시스템 메모리로 + 밀려나면서 프레임이 뚝뚝 끊긴다. 상한을 걸면 그 대신 우리 쪽이 먼저 + OOM 을 맞으므로, 티어를 낮추라는 안내로 이어진다. + """ + if ratio <= 0 or ratio >= 1: + return False + try: + import torch + + if not torch.cuda.is_available(): + return False + torch.cuda.set_per_process_memory_fraction(ratio, device_index) + log.info("VRAM 상한 적용: 전체의 %d%%", int(ratio * 100)) + return True + except Exception as exc: # noqa: BLE001 - 상한 실패가 치명적이지는 않다 + log.warning("VRAM 상한을 걸지 못했습니다: %s", exc) + return False + + +def has_torch() -> bool: + """PyTorch 가 설치돼 있는가. + + 포터블 exe 빌드는 torch 를 넣지 않는다. torch+CUDA 만 2.5GB 라 + 한 파일로 묶으면 실행할 때마다 그걸 임시폴더에 풀어야 해서 쓸 수 없다. + CTranslate2 기반 1~3티어는 torch 없이 동작하므로, 포터블은 그쪽만 담는다. + """ + import importlib.util + + return importlib.util.find_spec("torch") is not None + + +def has_awq() -> bool: + """4티어(AWQ Int4) 로드에 필요한 autoawq 가 있는지.""" + import importlib.util + + return importlib.util.find_spec("awq") is not None + + +def tier_availability(tier: Tier, gpu: GpuInfo | None = None) -> tuple[bool, str]: + """이 설치 환경에서 해당 티어를 쓸 수 있는지와, 안 되면 그 이유. + + torch 유무만 보면 부족하다. 실제로 막히는 경우가 둘 더 있다. + + - 4티어(AWQ Int4)는 autoawq 가 있어야 로드된다. requirements.txt 에 + 없는 선택 의존성이라 없을 때가 많다. + - VRAM 이 모자라면 설치가 아무리 멀쩡해도 안 올라간다. 기존에도 + min_vram_gb 로 경고 배지는 띄웠지만 선택은 막지 않아서, 골랐다가 + 로드에서 죽었다. 예: RTX 5050(7.5GB)은 정밀 10GB / 극한 16GB 라 + 둘 다 영영 못 쓴다. + + "사용 가능"이라고 해놓고 로드에서 죽으면 사용자는 원인을 알 수 없다. + 그래서 고를 수 있다고 말하기 전에 여기서 걸러낸다. + """ + if tier.mt.backend is not MTBackend.TRANSFORMERS: + return True, "" + if not has_torch(): + return False, "PyTorch가 없어 사용할 수 없습니다 (포터블 버전은 1~3티어만 지원)" + if tier.mt.compute_type == "int4" and not has_awq(): + return False, "autoawq가 없어 사용할 수 없습니다 (pip install autoawq)" + + info = detect_gpu() if gpu is None else gpu + # free 가 아니라 total 로 본다. free 는 다른 프로그램 때문에 수시로 + # 변해서, 그걸로 막으면 "아까는 되던 티어가 지금은 안 보인다"가 된다. + have_gb = info.total_vram_mb / 1024 + if info.available and have_gb < tier.min_vram_gb: + return False, ( + f"VRAM이 부족합니다 " + f"({tier.min_vram_gb:g}GB 필요, 이 GPU는 {have_gb:.1f}GB)" + ) + return True, "" + + +def first_available_tier(preferred: str) -> str: + """선호 티어가 불가능하면 쓸 수 있는 가장 좋은 티어로 내린다.""" + tier = get_tier(preferred) + gpu = detect_gpu() + if tier_availability(tier, gpu)[0]: + return tier.key + usable = [t for t in TIERS.values() if tier_availability(t, gpu)[0]] + if not usable: + return preferred + return max(usable, key=lambda t: t.order).key + + +def downgrade_compute(compute_type: str) -> str: + """저부하 모드에서 쓸 더 가벼운 연산 정밀도.""" + if compute_type in ("float16", "float32"): + return "int8_float16" + return compute_type + + +class ModelManager: + """현재 선택된 티어의 모델 두 개를 들고 있는다.""" + + def __init__(self, config: ModelConfig, low_power: bool = False) -> None: + self.config = config + #: 저부하(게임) 모드. 연산 정밀도를 낮춰 VRAM과 GPU 점유를 줄인다. + self.low_power = low_power + self._tier: Tier = get_tier(config.tier) + self._asr: SpeechRecognizer | None = None + self._mt: Translator | None = None + self._lock = threading.Lock() + + @property + def tier(self) -> Tier: + return self._tier + + @property + def ready(self) -> bool: + return bool(self._asr and self._asr.loaded and self._mt and self._mt.loaded) + + def effective_asr_spec(self): + """저부하 모드가 켜져 있으면 정밀도를 낮춘 사양을 돌려준다.""" + spec = self._tier.asr + if not self.low_power: + return spec + return replace( + spec, compute_type=downgrade_compute(spec.compute_type), beam_size=1 + ) + + def effective_mt_spec(self): + spec = self._tier.mt + # 양자화 LLM(int4)은 이미 최소 상태라 더 낮출 수 없다. + if not self.low_power or spec.compute_type == "int4": + return spec + return replace(spec, compute_type=downgrade_compute(spec.compute_type)) + + def recognizer(self) -> SpeechRecognizer: + with self._lock: + if self._asr is None: + self._asr = SpeechRecognizer( + self.effective_asr_spec(), self.config.device, self.config.device_index + ) + return self._asr + + def translator(self) -> Translator: + with self._lock: + if self._mt is None: + self._mt = create_translator( + self.effective_mt_spec(), + self.config.device, + self.config.device_index, + self.config.lora_adapter_path, + ) + return self._mt + + def preload(self, progress=None) -> None: + """두 모델을 미리 올린다. progress(str, float) 콜백으로 상태를 보고.""" + def report(msg: str, value: float) -> None: + if progress: + progress(msg, value) + + report(f"'{self._tier.name}' 음성인식 모델 준비 중…", 0.1) + self.recognizer().load() + report(f"'{self._tier.name}' 번역 모델 준비 중…", 0.6) + self.translator().load() + report("준비 완료", 1.0) + + def switch_tier(self, tier_key: str) -> Tier: + """티어 변경. 기존 모델은 내려서 VRAM을 즉시 돌려준다.""" + if tier_key == self._tier.key: + return self._tier + self.unload() + self.config.tier = tier_key + self._tier = get_tier(tier_key) + log.info("모델 티어 변경: %s", self._tier.name) + return self._tier + + def unload(self) -> None: + with self._lock: + if self._asr is not None: + self._asr.unload() + self._asr = None + if self._mt is not None: + self._mt.unload() + self._mt = None diff --git a/src/livesub/models/packs.py b/src/livesub/models/packs.py new file mode 100644 index 0000000..b81b39d --- /dev/null +++ b/src/livesub/models/packs.py @@ -0,0 +1,152 @@ +"""기본 제공 게임 용어집 팩. + +`resources/glossaries/*.json` 에 게임별로 묶어둔 용어를 읽어, 사용자가 켠 팩만 +사용자 용어집과 합쳐 하나의 `Glossary` 로 만든다. + +합칠 때 **사용자가 직접 등록한 항목이 항상 이긴다.** 기본 팩의 역어가 마음에 +안 들면 용어집 화면에서 같은 단어를 등록하면 그쪽이 적용된다. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from functools import lru_cache +from pathlib import Path + +from .glossary import Glossary, GlossaryEntry + +log = logging.getLogger(__name__) + +PACK_DIR = Path(__file__).resolve().parent.parent / "resources" / "glossaries" + + +#: 여러 팩을 동시에 켰을 때 같은 단어의 역어가 갈리는 것을 막기 위한 계층. +#: 숫자가 클수록 나중에 적용되어 이긴다. +SCOPE_PRIORITY = {"generic": 0, "game": 1} + + +@dataclass(frozen=True) +class GlossaryPack: + key: str + name: str + description: str + entries: tuple[GlossaryEntry, ...] + #: "generic" = 장르 공통, "game" = 특정 게임 전용 + scope: str = "game" + + def __len__(self) -> int: + return len(self.entries) + + @property + def priority(self) -> int: + return SCOPE_PRIORITY.get(self.scope, 1) + + @property + def is_generic(self) -> bool: + return self.scope == "generic" + + +@lru_cache(maxsize=1) +def available_packs() -> tuple[GlossaryPack, ...]: + """번들된 팩 목록. 파일이 깨져 있으면 그 팩만 건너뛴다.""" + if not PACK_DIR.is_dir(): + return () + packs: list[GlossaryPack] = [] + for path in sorted(PACK_DIR.glob("*.json")): + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + log.warning("용어집 팩을 읽지 못했습니다 (%s): %s", path.name, exc) + continue + entries = tuple( + GlossaryEntry( + source=item["source"], + targets=dict(item.get("targets", {})), + note=raw.get("name", path.stem), + whole_only=bool(item.get("whole_only", False)), + ) + for item in raw.get("entries", []) + if item.get("source") + ) + if not entries: + continue + packs.append( + GlossaryPack( + key=raw.get("key", path.stem), + name=raw.get("name", path.stem), + description=raw.get("description", ""), + entries=entries, + scope=raw.get("scope", "game"), + ) + ) + return tuple(packs) + + +def get_pack(key: str) -> GlossaryPack | None: + for pack in available_packs(): + if pack.key == key: + return pack + return None + + +def resolve_packs(enabled_packs: list[str]) -> list[GlossaryPack]: + """켜둔 팩을 적용 순서대로 돌려준다 (범용 먼저, 게임 전용 나중). + + `knocked` 는 FPS 공통에서 "기절", 배그에서 "눕혔다"다. 어느 쪽이 맞는지는 + 지금 무슨 게임을 하느냐로 갈리므로, **게임 전용 팩이 항상 범용을 이겨야** + 한다. 그렇지 않으면 체크박스를 누른 순서에 따라 결과가 달라진다. + """ + packs: list[GlossaryPack] = [] + for key in enabled_packs: + pack = get_pack(key) + if pack is None: + log.warning("알 수 없는 용어집 팩입니다: %s", key) + continue + packs.append(pack) + # 같은 계층 안에서는 사용자가 켠 순서를 유지한다 (stable sort). + return sorted(packs, key=lambda p: p.priority) + + +def find_conflicts(enabled_packs: list[str], lang: str = "ko") -> dict[str, list[tuple[str, str]]]: + """켜둔 게임 전용 팩끼리 역어가 갈리는 단어를 찾는다. + + 범용 팩은 게임 팩에 확실히 지므로 충돌로 보지 않는다. 게임 팩 둘 이상이 + 같은 단어를 다르게 번역하면 그건 사용자가 판단해야 할 문제라 UI로 알린다. + """ + seen: dict[str, list[tuple[str, str]]] = {} + for pack in resolve_packs(enabled_packs): + if pack.is_generic: + continue + for entry in pack.entries: + target = entry.target_for(lang) + if not target: + continue + seen.setdefault(entry.source.casefold(), []).append((pack.name, target)) + return { + source: owners + for source, owners in seen.items() + if len({t for _, t in owners}) > 1 + } + + +def build_glossary( + user_glossary: Glossary, + enabled_packs: list[str], + case_sensitive: bool = False, +) -> Glossary: + """켜둔 팩 + 사용자 용어집을 하나로 합친다. + + 적용 순서: 범용 팩 → 게임 전용 팩 → 사용자 항목. + 뒤에 오는 쪽이 이기므로 사용자가 직접 등록한 것이 항상 최우선이다. + """ + merged: list[GlossaryEntry] = [] + for pack in resolve_packs(enabled_packs): + merged.extend(pack.entries) + merged.extend(user_glossary.entries) + + combined = Glossary(case_sensitive=case_sensitive) + for entry in merged: + combined.add(entry) # add() 가 같은 원문을 교체해준다 + return combined diff --git a/src/livesub/models/speech_level.py b/src/livesub/models/speech_level.py new file mode 100644 index 0000000..0c2f931 --- /dev/null +++ b/src/livesub/models/speech_level.py @@ -0,0 +1,310 @@ +"""말투(존댓말/반말) 처리. + +규칙은 두 줄로 요약된다. + +1. 원문이 **실제로 존댓말**이면 반말 모드여도 존댓말을 지킨다. + 상대가 정중하게 말했는데 자막이 반말로 나오면 뉘앙스가 통째로 뒤집힌다. +2. 원문에 높임이 **없거나 알 수 없으면** 사용자가 고른 모드를 따른다. + 영어·중국어는 문법적 높임이 없으므로 항상 여기에 해당한다. + +한국어로 번역할 때만 동작한다. 영어·중국어는 대상 언어일 때 적용할 문법이 +없고, 일본어 경어 변환은 동사 활용이 필요해 규칙만으로는 위험하다. + +변환 방향도 한쪽만 한다. 번역 모델의 한국어 출력은 거의 항상 격식체 +(합니다체/해요체)이므로 **존댓말 모드는 손대지 않고**, 반말 모드일 때만 +낮춤 변환을 건다. 반대 방향(반말→존댓말)은 어간 정보가 없으면 훨씬 자주 +틀리므로 시도하지 않는다. +""" + +from __future__ import annotations + +import re +from enum import Enum + + +class SpeechLevel(str, Enum): + """사용자가 고르는 출력 말투.""" + + POLITE = "polite" # 존댓말 (기본) + CASUAL = "casual" # 반말 + + +class Politeness(str, Enum): + """원문에서 감지한 높임 정도.""" + + UNKNOWN = "unknown" # 높임이 없는 언어이거나 판단 불가 + POLITE = "polite" # 확실히 존댓말 + CASUAL = "casual" # 확실히 반말 + + +#: 문법적 높임 체계가 있는 언어. 나머지는 항상 UNKNOWN 이다. +HONORIFIC_LANGUAGES = frozenset({"ko", "ja"}) + +# --- 원문 높임 감지 --------------------------------------------------------- + +_KO_POLITE = re.compile( + r"(습니다|습니까|ㅂ니다|ㅂ니까|십시오|십시요|세요|셔요|시죠" + r"|어요|아요|에요|예요|해요|네요|지요|죠|드려|드립|주세|감사합)" +) +_JA_POLITE = re.compile( + r"(です|ます|ません|でした|ました|ましょう|ください|下さい" + r"|ございま|でしょう|いたし|申し上げ|なさい)" +) + +_HAS_HANGUL = re.compile(r"[가-힣]") +_HAS_JAPANESE = re.compile(r"[぀-ヿ一-鿿]") + + +def detect_politeness(text: str, lang: str) -> Politeness: + """원문의 높임 정도를 판단한다. + + 높임 체계가 없는 언어(영어·중국어)는 항상 UNKNOWN 이다. "please" 같은 + 공손 표현은 문법적 높임이 아니라 어조라, 이걸로 존댓말을 확정하면 + 오탐이 너무 많아진다. + + 한국어·일본어는 존댓말 표지가 있으면 POLITE, 없으면 CASUAL 로 본다. + 두 언어 모두 문장은 둘 중 하나로 끝나므로 "중립"이라는 선택지가 사실상 + 없고, 어미를 일일이 열거하는 것보다 이 편이 덜 틀린다. + """ + text = (text or "").strip() + if not text or lang not in HONORIFIC_LANGUAGES: + return Politeness.UNKNOWN + + polite, script = ( + (_KO_POLITE, _HAS_HANGUL) if lang == "ko" else (_JA_POLITE, _HAS_JAPANESE) + ) + if polite.search(text): + return Politeness.POLITE + if script.search(text): + return Politeness.CASUAL + return Politeness.UNKNOWN + + +# --- 한국어 낮춤 변환 ------------------------------------------------------- + +_HANGUL_BASE = 0xAC00 +_JONGSUNG_COUNT = 28 +_JUNGSUNG_COUNT = 21 +_HANGUL_COUNT = 11172 + +# 중성 인덱스 +_A, _AE, _EO, _E, _YEO, _O, _WA, _U, _WO, _EU, _I = 0, 1, 4, 5, 6, 8, 9, 13, 14, 18, 20 +#: '아'로 이어지는 밝은 모음 (모음조화) +_BRIGHT = frozenset({_A, _O, 2, _WA}) +#: 받침 없는 어간 + 어/아 의 축약. 오+아 -> 와, 지+어 -> 져 +_CONTRACTION = {_O: _WA, _U: _WO, _I: _YEO, _EU: _EO} +#: 붙여도 모양이 그대로인 어간 모음. 가+아 -> 가, 서+어 -> 서 +_ABSORBING = frozenset({_A, _EO, _AE, _E, _YEO, _WA, _WO}) +_RIEUL = 8 # 종성 ㄹ +_SSANG_SIOT = 20 # 종성 ㅆ — 과거형 '았/었' 판별용 + + +def _decompose(char: str) -> tuple[int, int, int] | None: + code = ord(char) - _HANGUL_BASE + if not (0 <= code < _HANGUL_COUNT): + return None + return ( + code // (_JUNGSUNG_COUNT * _JONGSUNG_COUNT), + (code // _JONGSUNG_COUNT) % _JUNGSUNG_COUNT, + code % _JONGSUNG_COUNT, + ) + + +def _compose(cho: int, jung: int, jong: int) -> str: + return chr( + _HANGUL_BASE + (cho * _JUNGSUNG_COUNT + jung) * _JONGSUNG_COUNT + jong + ) + + +def _attach_eo(stem: str) -> str: + """어간에 어/아 를 붙인다 (해체 활용). + + 먹->먹어, 좋->좋아, 하->해, 오->와, 주->줘, 빠지->빠져, 가->가. + 한국어 활용의 실제 규칙(모음조화 + 축약)을 그대로 구현한 것이라 + 어미 목록을 나열하는 방식보다 훨씬 넓게 맞는다. + """ + if not stem: + return stem + parts = _decompose(stem[-1]) + if parts is None: + return stem + "어" + cho, jung, jong = parts + + if jong == _SSANG_SIOT: + # 과거형 '았/었/였' 뒤에는 모음조화와 무관하게 항상 '어'가 온다. + # (남았아 X -> 남았어 O) + return stem + "어" + if jong: # 받침이 있으면 그냥 붙인다 + return stem + ("아" if jung in _BRIGHT else "어") + if cho == 18 and jung == _A: # 하 -> 해 + return stem[:-1] + _compose(cho, _AE, 0) + if jung in _CONTRACTION: # 오+아 -> 와, 지+어 -> 져 + return stem[:-1] + _compose(cho, _CONTRACTION[jung], 0) + if jung in _ABSORBING: # 가+아 -> 가 + return stem + return stem + ("아" if jung in _BRIGHT else "어") + + +def _attach_lge(stem: str) -> str: + """'~겠습니다' -> '~ㄹ게'. 치->칠게, 가->갈게, 먹->먹을게.""" + if not stem: + return stem + parts = _decompose(stem[-1]) + if parts is None: + return stem + "을게" + cho, jung, jong = parts + if jong: + return stem + "을게" + return stem[:-1] + _compose(cho, jung, _RIEUL) + "게" + + +#: 받침 없는 어간 + ㅂ니다 축약형. 규칙으로 풀기 어려운 것만 표로 둔다. +_CONTRACTED = { + "입니다": "야", + "됩니다": "돼", + "드립니다": "줄게", +} + +#: (패턴, 치환) — 긴 어미부터 적용해야 짧은 어미가 먼저 먹지 않는다. +_CASUAL_RULES: list[tuple[re.Pattern[str], str]] = [ + (re.compile(r"있습니다"), "있어"), + (re.compile(r"없습니다"), "없어"), + (re.compile(r"이에요"), "이야"), + (re.compile(r"예요"), "야"), + (re.compile(r"네요"), "네"), + (re.compile(r"지요"), "지"), + (re.compile(r"죠"), "지"), +] + +#: 어간을 활용해야 하는 어미들. 앞 글자를 잡아 _attach_* 로 넘긴다. +_GESSEUMNIDA = re.compile(r"([가-힣])겠습니다") +_SEUMNIDA = re.compile(r"([가-힣])습니다") +_BNIDA = re.compile(r"([가-힣])ᆸ니다|([가-힣])ㅂ니다") +_SEYO = re.compile(r"([가-힣])세요") +_SEUMNIKKA = re.compile(r"([가-힣])습니까") +#: 명령형 존대 '~십시오' 도 어간을 활용해야 한다. 조심하십시오 -> 조심해 +_SIPSIO = re.compile(r"([가-힣])십시[오요]") + +#: 문장 끝의 '요'만 떼어낸다. 문장 중간의 '요'(중요, 필요)는 건드리면 안 된다. +_TRAILING_YO = re.compile(r"([가-힣])요(\s*[.!?…,~]*)\s*$") +_TRAILING_YO_MID = re.compile(r"([가-힣])요(\s*[.!?…,~]+\s+)") + + +def _split_bnida(text: str) -> str: + """'갑니다' 처럼 어간에 ㅂ 이 합쳐진 형태를 푼다. 갑니다 -> 가 + 어.""" + + def _sub(match: re.Match[str]) -> str: + syllable = match.group(1) + parts = _decompose(syllable) + if parts is None: + return match.group(0) + cho, jung, jong = parts + if jong != 17: # ㅂ 받침이 아니면 대상이 아니다 + return match.group(0) + return _attach_eo(_compose(cho, jung, 0)) + + return re.sub(r"([가-힣])니다", _sub, text) + + +def to_casual(text: str) -> str: + """한국어 존댓말 문장을 반말로 낮춘다. + + 번역 모델이 내놓는 격식체 어미를 대상으로 한 규칙 기반 변환이라 + 100%는 아니다. 다만 게임 자막에서 실제로 나오는 어미는 종류가 + 많지 않고, 활용은 실제 문법 규칙으로 처리하므로 대부분을 덮는다. + """ + if not text: + return text + + out = text + for source, target in _CONTRACTED.items(): + out = out.replace(source, target) + + out = _GESSEUMNIDA.sub(lambda m: _attach_lge(m.group(1)), out) + out = _SIPSIO.sub(lambda m: _attach_eo(m.group(1)), out) + for pattern, target in _CASUAL_RULES: + out = pattern.sub(target, out) + out = _SEUMNIKKA.sub(lambda m: _attach_eo(m.group(1)) + "?", out) + out = _SEUMNIDA.sub(lambda m: _attach_eo(m.group(1)), out) + out = _SEYO.sub(lambda m: _attach_eo(m.group(1)), out) + out = _split_bnida(out) + + # 문장 끝 '요' 떼기 — 문장 중간의 '중요/필요'가 다치면 안 된다. + out = _TRAILING_YO_MID.sub(r"\1\2", out) + out = _TRAILING_YO.sub(r"\1\2", out) + return out.strip() + + +def apply_speech_level( + translated: str, + target_lang: str, + level: SpeechLevel, + source_politeness: Politeness = Politeness.UNKNOWN, +) -> str: + """번역 결과에 말투 설정을 적용한다. + + 존댓말 모드는 아무것도 하지 않는다. 번역 모델의 한국어 출력이 이미 + 격식체이고, 한국어에서 존댓말은 어떤 상황에서도 틀린 선택이 아니다. + """ + if target_lang != "ko" or level is not SpeechLevel.CASUAL: + return translated + # 사용자 규칙: 원문이 진짜 존댓말이면 반말 모드여도 존댓말을 지킨다. + if source_politeness is Politeness.POLITE: + return translated + return to_casual(translated) + + +def prompt_instruction(level: SpeechLevel, source_politeness: Politeness) -> str: + """지시문을 이해하는 LLM(Qwen3 등)에 넣을 말투 지시. + + Seed-X 처럼 지시문을 못 알아듣는 번역 전용 모델에는 쓰지 않는다. + 그쪽은 `apply_speech_level` 후처리로만 말투를 맞춘다. + """ + if source_politeness is Politeness.POLITE: + return "The speaker is being polite, so use Korean polite speech (존댓말)." + if level is SpeechLevel.CASUAL: + return "Use casual Korean speech (반말), the way friends talk while gaming." + return "Use polite Korean speech (존댓말)." + + +# --- 조사 교정 -------------------------------------------------------------- + +#: 앞말의 받침 유무로 갈리는 조사 쌍 (받침 있음, 받침 없음) +_PARTICLE_PAIRS = [("을", "를"), ("이", "가"), ("은", "는"), ("과", "와"), ("으로", "로")] +_PARTICLE_RE = re.compile( + r"([가-힣])(" + "|".join(sorted( + {p for pair in _PARTICLE_PAIRS for p in pair}, key=len, reverse=True + )) + r")(?=\s|$|[.!?,…~])" +) +_PARTICLE_LOOKUP = {} +for _with, _without in _PARTICLE_PAIRS: + _PARTICLE_LOOKUP[_with] = (_with, _without) + _PARTICLE_LOOKUP[_without] = (_with, _without) + + +def fix_particles(text: str) -> str: + """앞말 받침에 맞게 조사를 고친다. 자기장를 -> 자기장을. + + 용어집이 단어를 통째로 바꿔치기하기 때문에 생기는 문제다. 번역 모델은 + 자리표시자를 기준으로 조사를 골랐는데, 복원된 역어의 받침이 그와 다르면 + "자기장를" 같은 어긋남이 남는다. 받침은 한글 코드에서 바로 읽을 수 있어 + 규칙만으로 정확히 고칠 수 있다. + """ + if not text: + return text + + def _sub(match: re.Match[str]) -> str: + head, particle = match.group(1), match.group(2) + pair = _PARTICLE_LOOKUP.get(particle) + if pair is None: + return match.group(0) + parts = _decompose(head) + if parts is None: + return match.group(0) + has_batchim = parts[2] != 0 + # '으로/로' 는 ㄹ 받침이면 '로'를 쓴다 (서울로, 칼로). + if pair == ("으로", "로") and parts[2] == _RIEUL: + return head + "로" + return head + (pair[0] if has_batchim else pair[1]) + + return _PARTICLE_RE.sub(_sub, text) diff --git a/src/livesub/models/tiers.py b/src/livesub/models/tiers.py new file mode 100644 index 0000000..13bf279 --- /dev/null +++ b/src/livesub/models/tiers.py @@ -0,0 +1,209 @@ +"""모델 품질 티어 정의. + +사용자에게는 실제 모델 이름 대신 5단계 티어(속도우선 → 품질우선)로 노출한다. +각 티어는 ASR(음성인식) 1개 + MT(번역) 1개의 조합이다. + +선정 근거는 docs/MODELS.md 참고. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + +from ..constants import DEFAULT_TIER + + +class MTBackend(str, Enum): + """번역 모델 실행 백엔드.""" + + CTRANSLATE2 = "ctranslate2" # NLLB 계열 seq2seq + TRANSFORMERS = "transformers" # LLM 계열 (Seed-X / Qwen3) + + +class PromptStyle(str, Enum): + """LLM 백엔드의 프롬프트 형식. 모델 계열마다 요구사항이 다르다.""" + + #: 해당 없음 (seq2seq) + NONE = "none" + #: Seed-X 전용. chat template 이 없는 번역 전용 completion 모델이라 + #: 모델 카드에 적힌 고정 형식 + 끝의 `<언어코드>` 태그를 그대로 지켜야 한다. + SEEDX = "seedx" + #: Qwen3 등 일반 instruct 모델. chat template + 자유 지시문 사용 가능. + INSTRUCT = "instruct" + + +@dataclass(frozen=True) +class AsrSpec: + repo: str + compute_type: str + beam_size: int = 1 + vram_mb: int = 0 + + +@dataclass(frozen=True) +class MTSpec: + repo: str + backend: MTBackend + compute_type: str = "float16" + vram_mb: int = 0 + prompt_style: PromptStyle = PromptStyle.NONE + #: LoRA 추가학습 난이도 1(쉬움) ~ 3(까다로움) + finetune_ease: int = 2 + + @property + def supports_prompt_glossary(self) -> bool: + """프롬프트로 용어집을 지시할 수 있는가. + + 지시문을 이해하는 instruct 모델만 가능하다. Seed-X 는 번역만 하도록 + 학습된 completion 모델이라 "이 용어는 이렇게 옮겨라" 같은 문장을 + 넣으면 학습 분포를 벗어나 오히려 번역이 망가진다. seq2seq 도 마찬가지다. + 이 둘은 플레이스홀더 치환 방식으로 용어를 보호한다. + """ + return self.prompt_style is PromptStyle.INSTRUCT + + +@dataclass(frozen=True) +class Tier: + key: str + order: int + name: str + tagline: str + asr: AsrSpec + mt: MTSpec + approx_latency_s: float + min_vram_gb: float + notes: str = "" + recommended: bool = False + best_after_finetune: bool = False + + @property + def total_vram_mb(self) -> int: + return self.asr.vram_mb + self.mt.vram_mb + + @property + def download_gb(self) -> float: + return round(self.total_vram_mb / 1024 * 1.25, 1) + + +TIERS: dict[str, Tier] = {} + + +def _register(tier: Tier) -> Tier: + TIERS[tier.key] = tier + return tier + + +LIGHTNING = _register( + Tier( + key="lightning", + order=1, + name="번개", + tagline="속도 최우선 · 저사양 GPU", + asr=AsrSpec("Systran/faster-whisper-small", "int8_float16", 1, 700), + mt=MTSpec( + "entai2965/nllb-200-distilled-600M-ctranslate2", + MTBackend.CTRANSLATE2, + "int8_float16", + 800, + finetune_ease=1, + ), + approx_latency_s=0.6, + min_vram_gb=2, + notes="짧은 대사·게임 음성에 충분. 긴 문장이나 전문용어는 품질이 떨어질 수 있습니다.", + ) +) + +SWIFT = _register( + Tier( + key="swift", + order=2, + name="신속", + tagline="빠르면서 인식률은 최상급", + asr=AsrSpec("Systran/faster-whisper-large-v3-turbo", "int8_float16", 1, 1600), + mt=MTSpec( + "entai2965/nllb-200-distilled-600M-ctranslate2", + MTBackend.CTRANSLATE2, + "float16", + 1400, + finetune_ease=1, + ), + approx_latency_s=0.9, + min_vram_gb=4, + notes="음성인식을 최상급으로 올리고 번역만 경량으로 유지한 구성. 가성비가 가장 좋습니다.", + ) +) + +BALANCE = _register( + Tier( + key="balance", + order=3, + name="균형", + tagline="기본값 · 속도와 품질의 중간", + asr=AsrSpec("Systran/faster-whisper-large-v3-turbo", "float16", 2, 2400), + mt=MTSpec( + "entai2965/nllb-200-distilled-1.3B-ctranslate2", + MTBackend.CTRANSLATE2, + "float16", + 2900, + finetune_ease=2, + ), + approx_latency_s=1.2, + min_vram_gb=6, + notes="8GB VRAM 환경에서 가장 무난한 선택.", + ) +) + +PRECISION = _register( + Tier( + key="precision", + order=4, + name="정밀", + tagline="추천 · 현 시점 최고 번역 품질", + asr=AsrSpec("Systran/faster-whisper-large-v3-turbo", "float16", 2, 2400), + mt=MTSpec( + "ByteDance-Seed/Seed-X-PPO-7B-AWQ-Int4", + MTBackend.TRANSFORMERS, + "int4", + 5600, + prompt_style=PromptStyle.SEEDX, + finetune_ease=3, + ), + approx_latency_s=2.0, + min_vram_gb=10, + notes="번역 전용으로 학습된 7B 모델. 한/영/일/중 구어체와 문맥 처리가 확연히 좋습니다.", + recommended=True, + ) +) + +ULTIMATE = _register( + Tier( + key="ultimate", + order=5, + name="극한", + tagline="품질 최우선 · 추가학습(LoRA)에 최적", + asr=AsrSpec("Systran/faster-whisper-large-v3", "float16", 5, 3100), + mt=MTSpec( + "Qwen/Qwen3-8B", + MTBackend.TRANSFORMERS, + "float16", + 11000, + prompt_style=PromptStyle.INSTRUCT, + finetune_ease=1, + ), + approx_latency_s=3.0, + min_vram_gb=16, + notes="게임·방송 용어를 LoRA로 추가학습해 붙일 때 가장 잘 먹히는 구성입니다.", + best_after_finetune=True, + ) +) + +assert DEFAULT_TIER in TIERS, f"constants.DEFAULT_TIER({DEFAULT_TIER}) 가 티어 목록에 없습니다" + + +def ordered_tiers() -> list[Tier]: + return sorted(TIERS.values(), key=lambda t: t.order) + + +def get_tier(key: str) -> Tier: + return TIERS.get(key, TIERS[DEFAULT_TIER]) diff --git a/src/livesub/models/translator.py b/src/livesub/models/translator.py new file mode 100644 index 0000000..f5e8c74 --- /dev/null +++ b/src/livesub/models/translator.py @@ -0,0 +1,376 @@ +"""번역(MT) 래퍼. + +두 백엔드를 지원한다. + +* CTranslate2 — NLLB 계열 seq2seq. 가볍고 빠르며 추가학습이 쉽다. +* Transformers — Seed-X / Qwen3 같은 LLM. 품질이 좋고 LoRA 어댑터를 얹을 수 있다. + +LLM 백엔드는 다시 두 가지 프롬프트 형식으로 갈린다. 같은 "LLM"이라도 요구사항이 +정반대라 하나로 합치면 한쪽이 반드시 망가진다. + +* PromptStyle.SEEDX — Seed-X 는 chat template 이 없는 **번역 전용 completion 모델**이다. + 모델 카드가 정한 고정 문장과 끝의 `<언어코드>` 태그를 그대로 지켜야 하고, + 그 밖의 지시문을 끼워 넣으면 학습 분포를 벗어난다. 따라서 용어집은 프롬프트가 + 아니라 **플레이스홀더 치환**으로 넣는다. +* PromptStyle.INSTRUCT — Qwen3 등. chat template 을 쓰고 용어집을 지시문으로 넣는다. + +세 경우 모두 `Translator` 인터페이스를 따르므로 상위 레이어는 구분하지 않는다. +""" + +from __future__ import annotations + +import abc +import logging +import re +import threading + +from ..constants import LANGUAGES, models_dir +from .glossary import Glossary +from .speech_level import Politeness, SpeechLevel, fix_particles, prompt_instruction +from .tiers import MTBackend, MTSpec, PromptStyle + +log = logging.getLogger(__name__) + +#: 문장 끝에 LLM이 붙이곤 하는 군더더기를 걷어내기 위한 패턴 +_TRAILING_NOISE = re.compile(r"^\s*(?:번역[:：]|Translation[:：])\s*", re.IGNORECASE) + + +def _restore_or_retranslate(out: str, replacements: list[str], retry) -> str: + """자리표시자를 역어로 되돌린다. 하나라도 사라졌으면 용어집 없이 다시 번역. + + 번역 모델이 문장 일부를 통째로 누락하면 그 안의 자리표시자도 같이 + 사라진다. 그대로 복원하면 용어가 자막에서 증발하는데, 그건 용어집을 + 아예 안 쓴 것보다 나쁘다. 그래서 그럴 때는 보호 없이 한 번 더 번역해 + 원문 내용이 살아 있는 결과를 쓴다. + """ + missing = Glossary.missing_placeholders(out, replacements) + if not missing: + return Glossary.restore(out, replacements) + log.debug("자리표시자 %s 소실 — 용어집 없이 재번역", missing) + return retry() + + +def build_seedx_prompt(text: str, source_lang: str, target_lang: str) -> str: + """Seed-X 모델 카드가 요구하는 정확한 프롬프트. + + Translate the following English sentence into Chinese: + May the force be with you + + 끝의 `<언어코드>` 태그는 PPO 학습에 쓰인 것이라 **필수**다. 빠지면 번역 + 품질이 크게 흔들린다. 형식을 바꾸면 안 되므로 지시문이나 용어집 설명을 + 여기 끼워 넣지 말 것. + + 출처: https://huggingface.co/ByteDance-Seed/Seed-X-PPO-7B + """ + src = LANGUAGES[source_lang]["english"] + tgt = LANGUAGES[target_lang]["english"] + tag = LANGUAGES[target_lang]["seedx"] + return f"Translate the following {src} sentence into {tgt}:\n{text} <{tag}>" + + +class Translator(abc.ABC): + def __init__(self, spec: MTSpec, device: str = "cuda", device_index: int = 0) -> None: + self.spec = spec + self.device = device + self.device_index = device_index + self._lock = threading.Lock() + + @property + @abc.abstractmethod + def loaded(self) -> bool: ... + + @abc.abstractmethod + def load(self) -> None: ... + + @abc.abstractmethod + def unload(self) -> None: ... + + @abc.abstractmethod + def _translate( + self, + text: str, + source_lang: str, + target_lang: str, + glossary: Glossary | None, + speech_level: SpeechLevel, + source_politeness: Politeness, + ) -> str: ... + + def translate( + self, + text: str, + source_lang: str, + target_lang: str, + glossary: Glossary | None = None, + speech_level: SpeechLevel = SpeechLevel.POLITE, + source_politeness: Politeness = Politeness.UNKNOWN, + ) -> str: + """말투(speech_level)는 지시문을 이해하는 모델에만 프롬프트로 전달된다. + + 나머지 백엔드는 상위에서 `apply_speech_level` 후처리로 맞춘다. + """ + text = text.strip() + if not text: + return "" + if source_lang == target_lang: + return text + out = self._translate( + text, source_lang, target_lang, glossary, speech_level, source_politeness + ) + out = _TRAILING_NOISE.sub("", out).strip() + if target_lang == "ko": + # 용어집이 단어를 바꿔치기하면 앞말 받침이 달라져 조사가 어긋난다. + # (자기장를 -> 자기장을) + out = fix_particles(out) + return out + + +class CTranslate2Translator(Translator): + """NLLB 계열. 용어집은 플레이스홀더 보호 방식으로 적용한다.""" + + def __init__(self, spec: MTSpec, device: str = "cuda", device_index: int = 0) -> None: + super().__init__(spec, device, device_index) + self._model = None + self._tokenizer = None + + @property + def loaded(self) -> bool: + return self._model is not None + + def load(self) -> None: + if self._model is not None: + return + import ctranslate2 + from huggingface_hub import snapshot_download + from transformers import AutoTokenizer + + local = snapshot_download(self.spec.repo, cache_dir=str(models_dir())) + compute = self.spec.compute_type + if self.device == "cpu" and "float16" in compute: + compute = "int8" + log.info("번역 모델 로드: %s (%s/%s)", self.spec.repo, self.device, compute) + self._model = ctranslate2.Translator( + local, + device=self.device, + device_index=self.device_index, + compute_type=compute, + ) + self._tokenizer = AutoTokenizer.from_pretrained(local, cache_dir=str(models_dir())) + + def unload(self) -> None: + with self._lock: + self._model = None + self._tokenizer = None + + def _run_model(self, text: str, source_lang: str, target_lang: str) -> str: + """용어집 처리를 뺀 순수 번역 한 번.""" + src_code = LANGUAGES[source_lang]["nllb"] + tgt_code = LANGUAGES[target_lang]["nllb"] + with self._lock: + self._tokenizer.src_lang = src_code + tokens = self._tokenizer.convert_ids_to_tokens( + self._tokenizer.encode(text, truncation=True, max_length=512) + ) + results = self._model.translate_batch( + [tokens], + target_prefix=[[tgt_code]], + beam_size=2, + max_decoding_length=512, + repetition_penalty=1.1, + ) + hyp = results[0].hypotheses[0] + if hyp and hyp[0] == tgt_code: + hyp = hyp[1:] + return self._tokenizer.decode( + self._tokenizer.convert_tokens_to_ids(hyp), skip_special_tokens=True + ) + + def _translate(self, text, source_lang, target_lang, glossary, + speech_level=SpeechLevel.POLITE, source_politeness=Politeness.UNKNOWN): + # seq2seq 는 프롬프트가 없어 말투를 지시할 수 없다. 후처리로 맞춘다. + if self._model is None: + self.load() + assert self._model is not None and self._tokenizer is not None + + if glossary is None or not len(glossary): + return self._run_model(text, source_lang, target_lang) + + protected, replacements = glossary.protect(text, target_lang) + if not replacements: + return self._run_model(text, source_lang, target_lang) + + out = self._run_model(protected, source_lang, target_lang) + return _restore_or_retranslate( + out, replacements, + lambda: self._run_model(text, source_lang, target_lang), + ) + + +class LLMTranslator(Translator): + """Seed-X / Qwen3 등 LLM 백엔드. 용어집은 프롬프트로 직접 지시한다.""" + + def __init__(self, spec: MTSpec, device: str = "cuda", device_index: int = 0, + lora_path: str = "") -> None: + super().__init__(spec, device, device_index) + self.lora_path = lora_path + self._model = None + self._tokenizer = None + + @property + def loaded(self) -> bool: + return self._model is not None + + def load(self) -> None: + if self._model is not None: + return + import torch + from transformers import AutoModelForCausalLM, AutoTokenizer + + dtype = torch.float16 if self.device == "cuda" else torch.float32 + log.info("번역 LLM 로드: %s (%s)", self.spec.repo, self.device) + self._tokenizer = AutoTokenizer.from_pretrained( + self.spec.repo, cache_dir=str(models_dir()) + ) + try: + self._model = AutoModelForCausalLM.from_pretrained( + self.spec.repo, + cache_dir=str(models_dir()), + torch_dtype=dtype, + device_map={"": self.device_index} if self.device == "cuda" else "cpu", + ) + except (ImportError, RuntimeError, ValueError) as exc: + if self.spec.compute_type == "int4": + # AWQ 가중치는 전용 커널이 있어야 읽힌다. 메시지가 모호해서 + # 사용자가 원인을 못 찾는 일이 잦으므로 해결 방법을 같이 알려준다. + raise RuntimeError( + f"양자화 번역 모델({self.spec.repo})을 불러오지 못했습니다.\n" + "AWQ 커널이 필요합니다: pip install autoawq\n" + "설치가 어려우면 '모델' 화면에서 다른 티어를 선택하세요.\n" + f"원인: {exc}" + ) from exc + raise + if self.lora_path: + from peft import PeftModel + + log.info("LoRA 어댑터 적용: %s", self.lora_path) + self._model = PeftModel.from_pretrained(self._model, self.lora_path) + self._model.eval() + + def unload(self) -> None: + with self._lock: + self._model = None + self._tokenizer = None + try: + import torch + + torch.cuda.empty_cache() + except Exception: # noqa: BLE001 + pass + + def _build_prompt(self, text, source_lang, target_lang, glossary, + speech_level=SpeechLevel.POLITE, + source_politeness=Politeness.UNKNOWN) -> str: + if self.spec.prompt_style is PromptStyle.SEEDX: + # Seed-X 는 지시문을 못 알아듣는다. 말투는 후처리로만 맞춘다. + return build_seedx_prompt(text, source_lang, target_lang) + return self._build_instruct_prompt( + text, source_lang, target_lang, glossary, speech_level, source_politeness + ) + + def _build_instruct_prompt(self, text, source_lang, target_lang, glossary, + speech_level=SpeechLevel.POLITE, + source_politeness=Politeness.UNKNOWN) -> str: + """지시문을 이해하는 모델용. 가능하면 chat template 을 태운다.""" + src = LANGUAGES[source_lang]["english"] + tgt = LANGUAGES[target_lang]["english"] + hint = glossary.prompt_hint(text, target_lang) if glossary else "" + parts = [ + f"Translate the following {src} text into {tgt}. " + "It is a live spoken line from a game or broadcast, so keep the tone " + "casual and natural. Output only the translation." + ] + if target_lang == "ko": + parts.append(prompt_instruction(speech_level, source_politeness)) + if hint: + parts.append(hint) + parts.append(f"{src}: {text}\n{tgt}:") + instruction = "\n\n".join(parts) + + template = getattr(self._tokenizer, "chat_template", None) + if not template: + return instruction + # Qwen3 는 기본적으로 thinking 모드가 켜져 실시간 자막에는 너무 느리다. + try: + return self._tokenizer.apply_chat_template( + [{"role": "user", "content": instruction}], + tokenize=False, + add_generation_prompt=True, + enable_thinking=False, + ) + except TypeError: + # enable_thinking 을 모르는 템플릿 + return self._tokenizer.apply_chat_template( + [{"role": "user", "content": instruction}], + tokenize=False, + add_generation_prompt=True, + ) + + def _generate(self, prompt: str) -> str: + """프롬프트 하나를 돌려 첫 줄만 돌려준다.""" + import torch + + with self._lock: + inputs = self._tokenizer(prompt, return_tensors="pt").to(self._model.device) + with torch.inference_mode(): + generated = self._model.generate( + **inputs, + max_new_tokens=256, + do_sample=False, + num_beams=1, + repetition_penalty=1.05, + pad_token_id=self._tokenizer.eos_token_id, + ) + new_tokens = generated[0][inputs["input_ids"].shape[-1] :] + out = self._tokenizer.decode(new_tokens, skip_special_tokens=True) + return out.strip().split("\n")[0] + + def _translate(self, text, source_lang, target_lang, glossary, + speech_level=SpeechLevel.POLITE, source_politeness=Politeness.UNKNOWN): + if self._model is None: + self.load() + assert self._model is not None and self._tokenizer is not None + + def run(source: str, gloss) -> str: + return self._generate( + self._build_prompt( + source, source_lang, target_lang, gloss, + speech_level, source_politeness, + ) + ) + + # Seed-X 는 지시문을 못 알아들으므로 용어집을 자리표시자로 보호한다. + # 지시문을 이해하는 모델은 프롬프트로 넣으므로 보호가 필요 없다. + needs_placeholder = ( + self.spec.prompt_style is PromptStyle.SEEDX + and glossary is not None + and len(glossary) + ) + if not needs_placeholder: + return run(text, glossary) + + protected, replacements = glossary.protect(text, target_lang) + if not replacements: + return run(text, glossary) + + out = run(protected, None) + return _restore_or_retranslate( + out, replacements, lambda: run(text, None) + ) + + +def create_translator( + spec: MTSpec, device: str = "cuda", device_index: int = 0, lora_path: str = "" +) -> Translator: + if spec.backend is MTBackend.CTRANSLATE2: + return CTranslate2Translator(spec, device, device_index) + return LLMTranslator(spec, device, device_index, lora_path)