3 Commits

Author SHA1 Message Date
EJClaw
f9fd86464d fix: .gitignore 가 소스 패키지 src/livesub/models 를 통째로 먹고 있었다
Some checks failed
Windows / verify (push) Has been cancelled
`models/` 처럼 슬래시 없이 쓴 규칙은 깊이에 상관없이 같은 이름의 폴더를 전부
무시한다. 의도는 루트의 모델 캐시였는데 소스 패키지까지 같이 걸렸고, 그래서
asr/translator/manager/tiers/glossary/packs/speech_level 8개 파일이 한 번도
커밋된 적이 없다.

증상: Gitea 에서 clone 하면 `ModuleNotFoundError: No module named
'livesub.models'` 로 앱이 아예 임포트되지 않는다. 로컬 작업 트리에는 파일이
있으니 개발 중에는 절대 안 보인다. 방금 붙인 Windows CI VM 이 첫 검증에서
바로 잡아냈다.

/models/, /adapters/, /data/ 로 루트에 고정하고 빠진 파일을 추가한다.
2026-09-25 23:01:31 +09:00
EJClaw
ae295dc308 fix: 무인 프로비저닝이 통째로 안 돌던 인코딩/파이프라인 결함 3건
실제 설치에서 드러난 것들이다. 셋 다 증상이 원인과 전혀 안 닮아서 기록해 둔다.

1) provision.ps1 이 1행부터 파싱 실패 - 'Missing closing }'
   Windows PowerShell 5.1 은 BOM 없는 .ps1 을 ANSI(CP1252)로 읽는다.
   한글 주석의 UTF-8 바이트가 CP1252 에서 스마트따옴표(0x93/0x94 = " ")로
   변하고, 파서가 그걸 문자열 시작으로 보고 뒤를 전부 삼킨다.
   -> build.sh 가 ISO 에 넣을 때 UTF-8 BOM 을 붙인다.

2) stage2.ps1 의 1행이 주석으로 인식되지 않음
   agent 가 받은 본문에 BOM 이 있으면 Set-Content -Encoding UTF8 이 BOM 을
   한 번 더 붙여 BOM 이 두 개가 된다. PowerShell 은 첫 개만 벗겨내고 남은
   U+FEFF 때문에 '#' 이 줄 맨 앞이 아니게 된다.
   -> agent 가 TrimStart(U+FEFF), 서버는 charset=utf-8 을 명시한다.

3) Git 설치가 'Cannot convert System.Object[] to System.String' 로 실패
   Log 가 Write-Output 이라 함수 안에서 부르면 그 함수의 반환값에 로그가
   섞인다. Get-File 이 배열을 돌려주니 Start-Process 가 죽었다.
   -> Log 를 Write-Host 로 바꿨다.

덤으로 시리얼/HTTP 로그의 한글이 '?' 로 깨지던 것도 고쳤다 (SerialPort 기본
인코딩이 ASCII). 그리고 러너 토큰이 없어도 워크플로와 같은 단계를 VM 안에서
직접 돌리는 검증 절차를 stage2 에 붙였다.
2026-09-25 22:50:08 +09:00
EJClaw
e41d0cb441 ci: .9 안 Windows CI VM 무인 구축 자동화
Windows 설치 화면을 사람이 클릭할 수 없으므로 전 과정을 무인으로 짰다.
autounattend.xml 이 설치와 OOBE 를 끝내고, provision.ps1 이 virtio 드라이버를
오프라인으로 넣어 네트워크를 살리고, 상주 agent.ps1 이 호스트 HTTP 에서
stage2.ps1 을 받아 Git/pwsh7/Python/Node/CMake/act_runner 를 설치한다.

설치 내용을 ISO 에 박지 않고 호스트가 HTTP 로 내려주는 구조로 한 이유는
ISO 를 한 번 구우면 못 고쳐서, 한 줄 고칠 때마다 Windows 재설치가 되기
때문이다. stage2.ps1 만 고치면 게스트가 20초 안에 반영한다.

- 시스템 디스크는 SATA. virtio-blk 로 하면 Setup 이 디스크를 못 봐서
  WinPE 드라이버 주입이 필요해진다
- VM 은 systemd --user 트랜지언트 유닛으로 띄워 봇 cgroup 밖에 둔다
  (MemoryHigh=16G 안에 8G VM 이 들어가면 봇이 OOM 으로 죽는 전례)
- Node 포함. act_runner host 모드가 JS 액션을 PATH 의 node 로 돌린다
2026-09-25 22:13:55 +09:00
23 changed files with 2812 additions and 3 deletions

11
.gitignore vendored
View File

@@ -15,9 +15,14 @@ src/hearo/resources/bin/*.exe
src/hearo/resources/bin/*.pdb src/hearo/resources/bin/*.pdb
# 모델 캐시 / 학습 산출물 / 사용자 데이터 # 모델 캐시 / 학습 산출물 / 사용자 데이터
models/ #
adapters/ # ⚠️ 반드시 슬래시로 시작해 저장소 루트에 고정할 것. `models/` 로 쓰면 깊이에
data/ # 상관없이 같은 이름의 폴더를 전부 먹는다. 실제로 이것 때문에 소스 패키지
# src/livesub/models/ 8개 파일이 한 번도 커밋되지 않았고, Gitea 에서 clone 한
# 쪽은 앱이 아예 임포트되지 않았다. Windows CI VM 첫 검증에서 잡혔다.
/models/
/adapters/
/data/
*.jsonl *.jsonl
config.json config.json
glossary.json glossary.json

View File

@@ -142,6 +142,68 @@ IOMMU 그룹 수 0개
넣은 뒤 GPU를 `vfio-pci`에 묶어야 하는데, 그러면 **`.9`가 GPU를 잃습니다.** 넣은 뒤 GPU를 `vfio-pci`에 묶어야 하는데, 그러면 **`.9`가 GPU를 잃습니다.**
결국 "GPU를 누가 가질 것인가" 문제로 되돌아오므로 중첩으로는 해결되지 않습니다. 결국 "GPU를 누가 가질 것인가" 문제로 되돌아오므로 중첩으로는 해결되지 않습니다.
#### 만드는 법 — 스크립트 한 줄, 사람 손 없음
`scripts/win-ci/` 에 전부 들어 있습니다. Windows 설치 화면을 사람이 클릭할
방법이 없으므로 **무인 설치**로 짰습니다.
```bash
scripts/win-ci/build.sh # ISO 받기 + 무인설치 ISO 굽기 + 디스크 생성
/home/claude/win-ci/start-vm.sh # VM + 제어 서버 기동
/home/claude/win-ci/boot-cd.sh reset # CD 부팅 프롬프트 넘기기 (최초 1회)
/home/claude/win-ci/status.sh # 진행 상황
/home/claude/win-ci/shot.sh a.png # 화면 스냅샷
```
네 부분이 이어달리기를 합니다.
| 파일 | 어디서 도나 | 하는 일 |
|---|---|---|
| `unattend/autounattend.xml` | Windows Setup | 디스크 분할·설치·OOBE 건너뛰기·`ci` 자동 로그온 |
| `unattend/setup/provision.ps1` | 첫 로그온 1회 | virtio 드라이버(네트워크·화면), 화면꺼짐/잠금 해제 |
| `unattend/setup/agent.ps1` | 로그온마다 상주 | 호스트에서 `stage2.ps1` 을 받아 실행 |
| `serve/stage2.ps1` | 호스트가 HTTP 로 제공 | Git·pwsh7·Python·Node·CMake·act_runner 설치 |
**왜 설치 내용을 ISO 에 안 넣었나** — ISO 는 한 번 구우면 못 고칩니다. 설치
스크립트를 ISO 에 박으면 한 줄 고칠 때마다 Windows 를 다시 깔아야 합니다.
그래서 ISO 에는 껍데기(agent)만 넣고, 실제 설치 내용은 호스트의 HTTP 로
가져옵니다. `serve/stage2.ps1` 을 고치면 게스트가 20초 안에 알아서 새로
받아 실행합니다.
호스트와 게스트를 잇는 통로는 두 개입니다.
| 방향 | 통로 | 왜 |
|---|---|---|
| 호스트 → 게스트 | HTTP `10.0.2.2:8099` (QEMU user networking) | 스크립트·러너 토큰 전달 |
| 게스트 → 호스트 | COM1 → `serial.log`, HTTP POST → `guest.log` | 네트워크가 **없는** 드라이버 설치 단계도 봐야 하므로 시리얼이 필수 |
접속 정보: 계정 `ci` / 비밀번호는 `~/.config/ejclaw/secrets.json` 의
`win-ci.9` 항목. RDP 는 호스트 `127.0.0.1:13389`, VNC 는 `127.0.0.1:5919`.
##### 무인 설치에서 실제로 걸린 것들
- **`Press any key to boot from CD`** — UEFI 는 이걸 5초 안에 안 누르면 CD 를
포기합니다. 무인인데 누를 사람이 없어서 QMP `send-key` 로 밀어 넣습니다.
단, **설치가 시작된 뒤에는 절대 키를 더 보내면 안 됩니다.** 남는 Enter 가
설치 화면의 `Cancel` 을 눌러 "Are you sure you want to quit?" 를 띄웁니다.
반대로 설치가 끝난 뒤 재부팅 때는 아무 키도 안 누르는 게 정답입니다.
그래야 CD 를 건너뛰고 디스크로 부팅합니다.
- **시스템 디스크는 SATA(AHCI)** 로 둡니다. virtio-blk/scsi 로 하면 Windows
Setup 이 디스크 자체를 못 봐서 WinPE 단계 드라이버 주입이 필요해집니다.
네트워크·화면만 virtio 로 가고, 그 드라이버는 설치 후 ISO 에서 오프라인으로
넣습니다.
- **Windows ISO 는 UDF** 입니다. ISO9660 층에는 `README.TXT` 하나뿐이라
`xorriso` 로 열면 비어 보입니다. `mount -t udf` 로 봐야 합니다.
- `install.wim` 의 **인덱스 1** 이 `Windows 11 Enterprise LTSC 2024 Evaluation`
입니다(이미지 2개). `wiminfo` 로 확인하고 박았습니다.
- **LTSC 에는 winget 이 없습니다.** 스토어가 빠져 있어서 설치 파일을 직접
받습니다.
- **Node.js 가 필요합니다.** `act_runner` 의 host 모드는 `actions/checkout`,
`upload-artifact` 같은 JS 액션을 PATH 의 `node` 로 돌립니다. 빠뜨리면 그
단계들만 조용히 실패합니다.
- **러너는 서비스가 아니라 로그온 예약 작업**이어야 합니다. 서비스로 돌리면
데스크톱 세션이 없어 GUI 테스트와 스크린샷이 전부 실패합니다.
#### VM 에 얼마를 줄까 — 실측 근거 #### VM 에 얼마를 줄까 — 실측 근거
**vCPU 4개 / RAM 8GB / 디스크 80GB.** 추측이 아니라 재보고 정한 값입니다. **vCPU 4개 / RAM 8GB / 디스크 80GB.** 추측이 아니라 재보고 정한 값입니다.

44
scripts/win-ci/boot-cd.sh Executable file
View File

@@ -0,0 +1,44 @@
#!/bin/bash
# "Press any key to boot from CD or DVD" 를 넘긴다.
#
# UEFI 부팅 시 Windows ISO 는 매번 이 프롬프트를 띄우고 약 5초 안에 키가
# 없으면 CD 부팅을 포기한다. 무인 설치인데 사람이 눌러줄 수 없으므로
# 리셋 직후부터 키를 계속 밀어 넣는다.
#
# ⚠️ 설치가 시작된 뒤에는 절대 실행하지 말 것. 남는 Enter 가 설치 화면의
# Cancel 버튼을 눌러 "Are you sure you want to quit?" 가 뜬다 (실제로 겪음).
# 설치가 끝난 뒤 재부팅 때는 키를 안 누르는 게 정답이다 - 그래야 CD 를
# 건너뛰고 디스크로 부팅한다.
#
# 사용법: ./boot-cd.sh [reset] - reset 을 주면 VM 을 먼저 리셋한다
set -euo pipefail
MODE="${1:-}"
python3 - "$MODE" <<'PY'
import json, socket, sys, time
mode = sys.argv[1] if len(sys.argv) > 1 else ""
s = socket.socket(socket.AF_UNIX); s.settimeout(20)
s.connect("/home/claude/win-ci/qmp.sock")
f = s.makefile("rwb")
f.readline()
def cmd(o):
f.write((json.dumps(o) + "\n").encode()); f.flush()
while True:
line = f.readline()
if not line: raise SystemExit("qmp closed")
m = json.loads(line)
if "return" in m or "error" in m: return m
cmd({"execute": "qmp_capabilities"})
if mode == "reset":
print(cmd({"execute": "system_reset"}))
time.sleep(3)
# 프롬프트 창은 5초뿐이지만 펌웨어 POST 시간이 들쭉날쭉하다. 25초면 충분하고,
# 그 이상 밀면 설치 화면까지 Enter 가 새어 들어간다.
deadline = time.time() + 25
n = 0
while time.time() < deadline:
cmd({"execute": "send-key",
"arguments": {"keys": [{"type": "qcode", "data": "ret"}]}})
n += 1
time.sleep(0.4)
print("sent", n, "keys")
PY

64
scripts/win-ci/build.sh Executable file
View File

@@ -0,0 +1,64 @@
#!/bin/bash
# win-ci Windows VM 을 처음부터 만든다 (.9 안 중첩 KVM, 무인 설치).
#
# ./build.sh ISO 내려받기 + 무인설치 ISO 굽기 + 디스크 생성
# ./start-vm.sh VM + 제어 서버 기동 (systemd 트랜지언트 유닛)
# ./boot-cd.sh reset "Press any key to boot from CD" 넘기기 (최초 1회만)
# ./status.sh 진행 상황
# ./shot.sh out.png 화면 스냅샷
#
# 사람이 클릭할 일은 없다. autounattend.xml 이 설치를 끝내고, provision.ps1 이
# virtio 드라이버를 넣고, agent.ps1 이 호스트에서 stage2.ps1 을 받아 개발도구와
# act_runner 를 깐다.
#
# 필요 패키지: qemu-system-x86 ovmf swtpm swtpm-tools xorriso
# 그리고 실행 계정이 kvm 그룹에 있어야 한다 (없으면 sg kvm -c '...' 로 감싼다).
set -euo pipefail
D="${WIN_CI_DIR:-/home/claude/win-ci}"
HERE="$(cd "$(dirname "$0")" && pwd)"
WIN_ISO_URL='https://go.microsoft.com/fwlink/p/?linkid=2289029' # Win11 Ent LTSC 2024 평가판(90일, 키 불필요)
VIRTIO_URL='https://fedorapeople.org/groups/virt/virtio-win/direct-downloads/stable-virtio/virtio-win.iso'
mkdir -p "$D/iso" "$D/serve" "$D/swtpm" "$D/unattend-root/setup"
fetch() { # url dest 최소크기MB
if [ -f "$2" ] && [ "$(stat -c%s "$2")" -gt $(( $3 * 1024 * 1024 )) ]; then
echo "이미 있음: $2"; return
fi
echo "다운로드: $2"
curl -fL --retry 3 -o "$2" "$1"
}
fetch "$WIN_ISO_URL" "$D/iso/win11-ltsc.iso" 4000
fetch "$VIRTIO_URL" "$D/iso/virtio-win.iso" 500
# 무인 설치 미디어. Windows Setup 이 붙어 있는 모든 드라이브 루트에서
# autounattend.xml 을 찾으므로 CD 로 물려주면 된다.
cp "$HERE/unattend/autounattend.xml" "$D/unattend-root/"
# ⚠️ .ps1 은 반드시 UTF-8 BOM 으로 넣는다.
# Windows PowerShell 5.1 은 BOM 이 없으면 현재 ANSI 코드페이지(en-US -> CP1252)로
# 읽는다. 한글 주석의 UTF-8 바이트가 CP1252 에서 스마트따옴표(“ ” ‘)로 변하고,
# PowerShell 은 그걸 문자열 시작으로 보고 스크립트 뒤를 전부 삼킨다. 증상은
# 엉뚱하게도 "Missing closing '}' in statement block" 이다. 실제로 이것 때문에
# 첫 설치의 provision.ps1 이 통째로 안 돌았다.
for f in "$HERE/unattend/setup/"*.ps1; do
out="$D/unattend-root/setup/$(basename "$f")"
printf '\xef\xbb\xbf' > "$out"
cat "$f" >> "$out"
done
xorriso -as mkisofs -V UNATTEND -J -joliet-long -R -o "$D/unattend.iso" "$D/unattend-root/"
# 호스트 제어 채널이 내려줄 스크립트
cp "$HERE/serve/stage2.ps1" "$D/serve/"
cp "$HERE"/*.sh "$HERE"/serve-host.py "$D/"
chmod +x "$D"/*.sh
[ -f "$D/win-ci.qcow2" ] || qemu-img create -f qcow2 "$D/win-ci.qcow2" 80G
# SecureBoot 용 - MS 키가 미리 들어 있는 변수 이미지여야 Windows 부트로더가 통과한다
[ -f "$D/OVMF_VARS.fd" ] || cp /usr/share/OVMF/OVMF_VARS_4M.ms.fd "$D/OVMF_VARS.fd"
echo
echo "준비 끝. 다음: $D/start-vm.sh 그리고 $D/boot-cd.sh reset"

28
scripts/win-ci/click.sh Executable file
View File

@@ -0,0 +1,28 @@
#!/bin/bash
# VM 화면의 좌표를 클릭한다 (usb-tablet 절대좌표).
# 사용법: ./click.sh <x> <y> [화면폭 화면높이] 기본 1280x800
set -euo pipefail
python3 - "$@" <<'PY'
import json, socket, sys, time
x, y = int(sys.argv[1]), int(sys.argv[2])
W = int(sys.argv[3]) if len(sys.argv) > 3 else 1280
H = int(sys.argv[4]) if len(sys.argv) > 4 else 800
s = socket.socket(socket.AF_UNIX); s.settimeout(20)
s.connect("/home/claude/win-ci/qmp.sock"); f = s.makefile("rwb"); f.readline()
def cmd(o):
f.write((json.dumps(o)+"\n").encode()); f.flush()
while True:
m = json.loads(f.readline())
if "return" in m or "error" in m: return m
cmd({"execute": "qmp_capabilities"})
ax, ay = int(x / W * 32767), int(y / H * 32767)
ev = [{"type": "abs", "data": {"axis": "x", "value": ax}},
{"type": "abs", "data": {"axis": "y", "value": ay}}]
print(cmd({"execute": "input-send-event", "arguments": {"events": ev}}))
time.sleep(0.3)
print(cmd({"execute": "input-send-event", "arguments": {"events": [
{"type": "btn", "data": {"down": True, "button": "left"}}]}}))
time.sleep(0.15)
print(cmd({"execute": "input-send-event", "arguments": {"events": [
{"type": "btn", "data": {"down": False, "button": "left"}}]}}))
PY

55
scripts/win-ci/run-vm.sh Executable file
View File

@@ -0,0 +1,55 @@
#!/bin/bash
# win-ci Windows VM 기동 (.9 안 중첩 KVM)
#
# 이 스크립트를 직접 실행하지 말 것. start-vm.sh 가 systemd 트랜지언트 유닛으로
# 띄운다. 에이전트 턴에서 그냥 qemu 를 띄우면 ejclaw.service 의 cgroup 안에서
# 돌아 MemoryHigh=16G 상한을 쳐서 봇이 OOM 으로 죽는다 (2026-07-26 전례).
set -euo pipefail
D=/home/claude/win-ci
cd "$D"
# TPM 2.0 - Windows 11 요구사항
if [ ! -S "$D/swtpm/sock" ]; then
swtpm socket --tpmstate "dir=$D/swtpm" \
--ctrl "type=unixio,path=$D/swtpm/sock" --tpm2 -d
for i in $(seq 1 20); do [ -S "$D/swtpm/sock" ] && break; sleep 0.3; done
fi
rm -f "$D/qmp.sock"
exec qemu-system-x86_64 \
-name win-ci,process=win-ci \
-machine q35,smm=on,vmport=off \
-accel kvm \
-cpu host,hv_relaxed,hv_spinlocks=0x1fff,hv_vapic,hv_time \
-smp 4,sockets=1,cores=4,threads=1 \
-m 8192 \
-rtc base=localtime,driftfix=slew \
-global driver=cfi.pflash01,property=secure,value=on \
-global ICH9-LPC.disable_s3=1 \
-drive if=pflash,format=raw,unit=0,readonly=on,file=/usr/share/OVMF/OVMF_CODE_4M.secboot.fd \
-drive if=pflash,format=raw,unit=1,file="$D/OVMF_VARS.fd" \
-chardev "socket,id=chrtpm,path=$D/swtpm/sock" \
-tpmdev emulator,id=tpm0,chardev=chrtpm \
-device tpm-tis,tpmdev=tpm0 \
-device virtio-vga,max_outputs=2,xres=1920,yres=1080 \
-audiodev none,id=aud0 \
-device intel-hda \
-device hda-duplex,audiodev=aud0 \
-netdev user,id=net0,hostfwd=tcp:127.0.0.1:13389-:3389 \
-device virtio-net-pci,netdev=net0 \
-device qemu-xhci,id=xhci \
-device usb-kbd -device usb-tablet \
-drive "file=$D/win-ci.qcow2,if=none,id=sysdisk,format=qcow2,cache=writeback,discard=unmap,aio=threads" \
-device ide-hd,drive=sysdisk,bus=ide.0,bootindex=2 \
-drive "file=$D/iso/win11-ltsc.iso,if=none,id=cdwin,media=cdrom,readonly=on" \
-device ide-cd,drive=cdwin,bus=ide.1,bootindex=1 \
-drive "file=$D/iso/virtio-win.iso,if=none,id=cdvirtio,media=cdrom,readonly=on" \
-device ide-cd,drive=cdvirtio,bus=ide.2 \
-drive "file=$D/unattend.iso,if=none,id=cdunattend,media=cdrom,readonly=on" \
-device ide-cd,drive=cdunattend,bus=ide.3 \
-serial "file:$D/serial.log" \
-qmp "unix:$D/qmp.sock,server=on,wait=off" \
-vnc 127.0.0.1:19 \
-boot menu=off,strict=on

75
scripts/win-ci/serve-host.py Executable file
View File

@@ -0,0 +1,75 @@
#!/usr/bin/env python3
"""win-ci VM 과의 통신 채널.
게스트(Windows)는 QEMU user networking 안에 있어서 호스트를 10.0.2.2 로 본다.
이 서버는 그 반대편이다.
GET /ping 게스트가 네트워크 올라왔는지 확인용
GET /stage20.ps1 게스트가 실행할 설치 스크립트 (serve/ 안의 파일 그대로)
GET /<file> serve/ 안의 아무 파일 (러너 등록 토큰 등)
POST /log 게스트 로그 -> guest.log 에 append
127.0.0.1 만 바인딩한다. slirp 의 10.0.2.2 는 호스트 루프백으로 들어오므로
외부에 열 필요가 없다.
"""
import http.server
import os
import socketserver
import sys
from datetime import datetime
ROOT = "/home/claude/win-ci/serve"
LOG = "/home/claude/win-ci/guest.log"
PORT = 8099
class Handler(http.server.SimpleHTTPRequestHandler):
def __init__(self, *a, **kw):
super().__init__(*a, directory=ROOT, **kw)
def guess_type(self, path):
# .ps1 은 charset 을 반드시 명시한다. 없으면 PowerShell 쪽에서
# ISO-8859-1 로 디코딩해 한글 주석이 깨지고, 깨진 바이트가 CP1252 의
# 스마트따옴표가 되어 파서가 스크립트를 통째로 삼킨다.
if str(path).endswith(".ps1"):
return "text/plain; charset=utf-8"
return super().guess_type(path)
def do_GET(self):
if self.path.rstrip("/") == "/ping":
body = b"pong\n"
self.send_response(200)
self.send_header("Content-Type", "text/plain")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
return
super().do_GET()
def do_POST(self):
if self.path.rstrip("/") != "/log":
self.send_error(404)
return
n = int(self.headers.get("Content-Length") or 0)
data = self.rfile.read(n).decode("utf-8", "replace").strip()
with open(LOG, "a", encoding="utf-8") as f:
f.write(f"{datetime.now():%Y-%m-%d %H:%M:%S} {data}\n")
self.send_response(204)
self.end_headers()
def log_message(self, fmt, *args):
# 접근 로그는 stdout 으로만. journald 가 받는다.
sys.stdout.write("%s %s\n" % (self.address_string(), fmt % args))
sys.stdout.flush()
class Server(socketserver.ThreadingTCPServer):
allow_reuse_address = True
daemon_threads = True
if __name__ == "__main__":
os.makedirs(ROOT, exist_ok=True)
with Server(("127.0.0.1", PORT), Handler) as httpd:
print(f"win-ci control server on 127.0.0.1:{PORT}, root={ROOT}", flush=True)
httpd.serve_forever()

View File

@@ -0,0 +1,264 @@
# win-ci VM 설치 스크립트 (호스트가 HTTP 로 내려준다 - agent.ps1 이 실행)
#
# 이 파일은 호스트의 /home/claude/win-ci/serve/stage2.ps1 이다. 고치면 게스트가
# 20초 안에 알아서 새로 받아 실행한다. Windows 재설치 필요 없다.
#
# 규칙: 반드시 idempotent. 이미 깔린 건 건너뛴다. 중간에 죽어도 다시 돌 수 있다.
#
# LTSC 2024 에는 winget(스토어)이 없다. 그래서 설치 파일을 직접 받는다.
$ErrorActionPreference = 'Stop'
[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
$Dl = 'C:\win-ci\dl'
New-Item -ItemType Directory -Force -Path $Dl | Out-Null
# Write-Output 를 쓰면 안 된다. 함수 안에서 부르면 그 함수의 반환값에
# 로그 문자열이 섞여 들어가 $f 가 배열이 되고, Start-Process 가
# "Cannot convert System.Object[] to System.String" 으로 죽는다.
# 실제로 이것 때문에 Git 설치가 실패했다.
function Log($m) { Write-Host "$m" }
function Get-File($url, $name) {
$out = Join-Path $Dl $name
if ((Test-Path $out) -and (Get-Item $out).Length -gt 1mb) { Log "캐시 사용 $name"; return $out }
for ($i = 1; $i -le 3; $i++) {
try {
Log "다운로드 ($i/3) $name"
Invoke-WebRequest -Uri $url -OutFile $out -UseBasicParsing -TimeoutSec 600
return $out
} catch {
Log "실패: $($_.Exception.Message)"
if ($i -eq 3) { throw }
Start-Sleep -Seconds 10
}
}
}
function Sync-Path {
$m = [Environment]::GetEnvironmentVariable('Path', 'Machine')
$u = [Environment]::GetEnvironmentVariable('Path', 'User')
$env:Path = "$m;$u"
}
function Have($exe) { Sync-Path; return [bool](Get-Command $exe -ErrorAction SilentlyContinue) }
# ------------------------------------------------------------------ Git
# actions/checkout@v4 가 git 을 쓴다. 없으면 소스를 못 받는다.
if (Have 'git') {
Log "git 이미 있음: $(git --version)"
} else {
$f = Get-File 'https://github.com/git-for-windows/git/releases/download/v2.55.0.windows.5/Git-2.55.0.5-64-bit.exe' 'git.exe'
Log "Git 설치 중"
Start-Process $f -ArgumentList '/VERYSILENT', '/NORESTART', '/NOCANCEL', '/SP-' -Wait
if (-not (Have 'git')) { throw 'git 설치 실패' }
Log "git 설치 완료: $(git --version)"
}
# ------------------------------------------------------------------ pwsh 7
# 워크플로가 shell: pwsh 를 쓴다. Windows 기본은 5.1 이라 별도로 필요하다.
if (Have 'pwsh') {
Log "pwsh 이미 있음: $(pwsh --version)"
} else {
$f = Get-File 'https://github.com/PowerShell/PowerShell/releases/download/v7.4.6/PowerShell-7.4.6-win-x64.msi' 'pwsh.msi'
Log "PowerShell 7 설치 중"
Start-Process 'msiexec.exe' -ArgumentList '/i', "`"$f`"", '/qn', '/norestart', 'ADD_PATH=1' -Wait
if (-not (Have 'pwsh')) { throw 'pwsh 설치 실패' }
Log "pwsh 설치 완료: $(pwsh --version)"
}
# ------------------------------------------------------------------ Python 3.12
if (Have 'python') {
Log "python 이미 있음: $(python --version 2>&1)"
} else {
$f = Get-File 'https://www.python.org/ftp/python/3.12.10/python-3.12.10-amd64.exe' 'python.exe'
Log "Python 3.12 설치 중"
Start-Process $f -ArgumentList '/quiet', 'InstallAllUsers=1', 'PrependPath=1', `
'Include_test=0', 'Include_launcher=1', 'AssociateFiles=0' -Wait
if (-not (Have 'python')) { throw 'python 설치 실패' }
Log "python 설치 완료: $(python --version 2>&1)"
}
# ------------------------------------------------------------------ Node
# act_runner 의 host 모드는 JS 액션(checkout, upload-artifact)을 돌리려고
# PATH 의 node 를 쓴다. 없으면 그 단계들이 전부 실패한다.
if (Have 'node') {
Log "node 이미 있음: $(node --version)"
} else {
$f = Get-File 'https://nodejs.org/dist/v22.20.0/node-v22.20.0-x64.msi' 'node.msi'
Log "Node.js 설치 중"
Start-Process 'msiexec.exe' -ArgumentList '/i', "`"$f`"", '/qn', '/norestart' -Wait
if (-not (Have 'node')) { throw 'node 설치 실패' }
Log "node 설치 완료: $(node --version)"
}
# ------------------------------------------------------------------ CMake
# native\process_loopback 빌드용. 워크플로에서 continue-on-error 지만 있으면 검증된다.
if (Have 'cmake') {
Log "cmake 이미 있음: $(cmake --version | Select-Object -First 1)"
} else {
$f = Get-File 'https://github.com/Kitware/CMake/releases/download/v3.31.6/cmake-3.31.6-windows-x86_64.msi' 'cmake.msi'
Log "CMake 설치 중"
Start-Process 'msiexec.exe' -ArgumentList '/i', "`"$f`"", '/qn', '/norestart', 'ADD_CMAKE_TO_PATH=System' -Wait
Sync-Path
if (Have 'cmake') { Log "cmake 설치 완료: $(cmake --version | Select-Object -First 1)" }
else { Log "WARN cmake 설치 실패 - 워크플로의 native 빌드 단계는 건너뛰어진다" }
}
# ------------------------------------------------------------------ act_runner
$RunnerDir = 'C:\gitea-runner'
$RunnerExe = Join-Path $RunnerDir 'act_runner.exe'
New-Item -ItemType Directory -Force -Path $RunnerDir | Out-Null
if (Test-Path $RunnerExe) {
Log "act_runner 이미 있음"
} else {
Log "act_runner 다운로드"
Invoke-WebRequest -Uri 'https://dl.gitea.com/act_runner/0.2.13/act_runner-0.2.13-windows-amd64.exe' `
-OutFile $RunnerExe -UseBasicParsing -TimeoutSec 300
Log "act_runner 받음: $((Get-Item $RunnerExe).Length) bytes"
}
# GPU 러너와 달리 이건 CPU 러너지만, 같은 원칙으로 capacity 1 로 고정한다.
# 이 VM 은 vCPU 4개뿐이라 두 작업이 동시에 돌면 둘 다 느려지고 타임아웃 위험이 생긴다.
$cfg = Join-Path $RunnerDir 'config.yaml'
if (-not (Test-Path $cfg)) {
@'
log:
level: info
runner:
capacity: 1
timeout: 60m
shutdown_timeout: 3m
insecure: false
fetch_timeout: 5s
fetch_interval: 2s
container:
network: ""
privileged: false
force_pull: false
host:
workdir_parent: C:\gitea-runner\work
'@ | Set-Content -Path $cfg -Encoding ASCII
Log "config.yaml 생성 (capacity 1)"
}
# ------------------------------------------- 러너 등록 (토큰이 호스트에 있을 때만)
# 호스트의 serve/runner-token.txt 에 등록 토큰을 넣어주면 여기서 등록한다.
# 없으면 이 단계만 건너뛰고 나머지는 전부 완료로 본다.
$registered = Test-Path (Join-Path $RunnerDir '.runner')
if ($registered) {
Log "러너 이미 등록됨"
} else {
$tok = $null
try {
$tok = (Invoke-WebRequest -Uri 'http://10.0.2.2:8099/runner-token.txt' -UseBasicParsing -TimeoutSec 10).Content
if ($tok -is [byte[]]) { $tok = [Text.Encoding]::ASCII.GetString($tok) }
$tok = $tok.Trim()
} catch { $tok = $null }
if ([string]::IsNullOrWhiteSpace($tok)) {
Log "등록 토큰 없음 (serve/runner-token.txt) - 러너 등록은 건너뜀. 사전 설치는 전부 끝났다."
} else {
Log "러너 등록 시도 (토큰 길이 $($tok.Length))"
Push-Location $RunnerDir
$out = & $RunnerExe register --no-interactive `
--instance 'https://git.tkrmagid.kr' `
--token $tok `
--name 'win-ci' `
--labels 'windows:host' 2>&1 | Out-String
Pop-Location
Log "register -> $($out.Trim())"
if (-not (Test-Path (Join-Path $RunnerDir '.runner'))) { throw "러너 등록 실패" }
# 서비스로 만들지 않는다. SYSTEM 권한이 되고, 데스크톱 세션이 없어
# GUI 테스트/스크린샷이 전부 실패한다. 로그온 예약 작업이어야 한다.
& schtasks.exe /Create /TN 'gitea-runner' /SC ONLOGON /RL HIGHEST /F `
/TR "cmd.exe /c cd /d C:\gitea-runner && act_runner.exe daemon --config config.yaml" | Out-Null
& schtasks.exe /Run /TN 'gitea-runner' | Out-Null
Log "러너 등록 + 로그온 작업 등록 완료, daemon 시작"
}
}
# ------------------------------------------------------------------ 요약
Sync-Path
Log "=== 설치 요약 ==="
foreach ($c in 'git', 'pwsh', 'python', 'node', 'cmake') {
$v = try { (& $c --version 2>&1 | Select-Object -First 1) } catch { '없음' }
Log ("{0,-8} {1}" -f $c, $v)
}
Log "act_runner $(if (Test-Path $RunnerExe) { 'OK' } else { '없음' }) 등록=$(Test-Path (Join-Path $RunnerDir '.runner'))"
# 모니터 2개가 실제로 잡혔는지 - virtio-gpu 드라이버가 들어갔는지 확인하는 지표다.
try {
Add-Type -AssemblyName System.Windows.Forms -ErrorAction Stop
$screens = [System.Windows.Forms.Screen]::AllScreens
Log "화면 $($screens.Count)개: $(($screens | ForEach-Object { $_.Bounds.Width.ToString() + 'x' + $_.Bounds.Height }) -join ', ')"
$gpu = Get-CimInstance Win32_VideoController | ForEach-Object { $_.Name }
Log "비디오 어댑터: $($gpu -join ', ')"
} catch {
Log "화면 열거 실패: $($_.Exception.Message)"
}
Log "=== stage2 완료 ==="
# ==================================================================
# Windows 전용 경로 실제 검증
#
# 러너 등록 토큰이 없어도 여기서 워크플로와 똑같은 단계를 직접 돌린다.
# 이게 VM 을 만든 진짜 이유다 - 리눅스에서 못 돌리던 경로(WASAPI, 전역
# 단축키, 모니터 열거, PyInstaller exe)가 실제로 통과하는지 본다.
#
# 통째로 try/catch 로 감싼다. 검증이 실패해도 stage2 자체는 성공이어야
# agent 가 무한 재시도에 빠지지 않는다. 결과는 로그로 남는다.
# ==================================================================
$Repo = 'C:\win-ci\live-app-translator'
function Step($name, $block) {
Log "--- [$name] 시작"
try {
& $block
Log "--- [$name] 통과"
return $true
} catch {
Log "--- [$name] 실패: $($_.Exception.Message)"
return $false
}
}
$ErrorActionPreference = 'Continue'
try {
Sync-Path
if (Test-Path $Repo) {
Log "저장소 갱신"
Push-Location $Repo; & git fetch --all -q; & git reset --hard origin/main -q; Pop-Location
} else {
Log "저장소 clone"
& git clone -q https://git.tkrmagid.kr/tkrmagid/live-app-translator $Repo
}
Push-Location $Repo
$py = Join-Path $Repo '.venv-ci\Scripts\python.exe'
if (-not (Test-Path $py)) { & python -m venv .venv-ci }
& $py -m pip install --upgrade pip --quiet
& $py -m pip install -r packaging\requirements-portable.txt --quiet
& $py -m pip install pytest ruff --quiet
# 워크플로와 같은 조건인지 - torch 가 없어야 "포터블에서 1~3티어가 된다"를
# 실제로 검증한 셈이 된다.
& $py -c "import importlib.util,sys; sys.exit(0 if importlib.util.find_spec('torch') is None else 1)"
Log "포터블 조건(torch 없음) = $($LASTEXITCODE -eq 0)"
$env:PYTHONPATH = 'src'
$r = @{}
$r['pytest'] = Step 'pytest' { & $py -m pytest tests -q 2>&1 | Select-Object -Last 8 | ForEach-Object { Log " $_" }; if ($LASTEXITCODE -ne 0) { throw "exit $LASTEXITCODE" } }
$r['ruff'] = Step 'ruff' { & $py -m ruff check src tests scripts 2>&1 | Select-Object -Last 5 | ForEach-Object { Log " $_" }; if ($LASTEXITCODE -ne 0) { throw "exit $LASTEXITCODE" } }
$r['windows경로'] = Step 'windows_smoke' { & $py packaging\windows_smoke.py 2>&1 | ForEach-Object { Log " $_" }; if ($LASTEXITCODE -ne 0) { throw "exit $LASTEXITCODE" } }
$r['exe빌드'] = Step 'pyinstaller' { & $py -m pip install pyinstaller --quiet; & $py -m PyInstaller packaging\livesub.spec --noconfirm 2>&1 | Select-Object -Last 4 | ForEach-Object { Log " $_" }; if (-not (Test-Path 'dist\LiveSub.exe')) { throw 'dist\LiveSub.exe 없음' }; Log " exe 크기 $([math]::Round((Get-Item 'dist\LiveSub.exe').Length/1MB,1)) MB" }
$r['exe기동'] = Step 'exe 실행' { & $py packaging\windows_smoke.py --exe dist\LiveSub.exe 2>&1 | ForEach-Object { Log " $_" }; if ($LASTEXITCODE -ne 0) { throw "exit $LASTEXITCODE" } }
Log "=== Windows 검증 결과 ==="
foreach ($k in $r.Keys) { Log ("{0,-14} {1}" -f $k, $(if ($r[$k]) { 'PASS' } else { 'FAIL' })) }
Log "=== 검증 끝 ==="
Pop-Location
} catch {
Log "검증 준비 단계에서 실패: $($_.Exception.Message)"
try { Pop-Location } catch {}
}

31
scripts/win-ci/shot.sh Executable file
View File

@@ -0,0 +1,31 @@
#!/bin/bash
# win-ci VM 화면 스냅샷. QMP screendump 로 찍는다.
# 사용법: ./shot.sh [출력경로.png]
set -euo pipefail
OUT="${1:-/home/claude/EJClaw/data/attachments/generated/win-ci-$(date +%H%M%S).png}"
mkdir -p "$(dirname "$OUT")"
python3 - "$OUT" <<'PY'
import json, socket, sys, time
out = sys.argv[1]
s = socket.socket(socket.AF_UNIX)
s.settimeout(20)
s.connect("/home/claude/win-ci/qmp.sock")
f = s.makefile("rwb")
f.readline() # greeting
def cmd(o):
f.write((json.dumps(o) + "\n").encode()); f.flush()
while True:
line = f.readline()
if not line: raise SystemExit("qmp closed")
m = json.loads(line)
if "return" in m or "error" in m: return m
cmd({"execute": "qmp_capabilities"})
r = cmd({"execute": "screendump", "arguments": {"filename": out, "format": "png"}})
if "error" in r:
# 구버전 qemu 는 format 인자를 모른다 -> ppm 으로 받는다
r = cmd({"execute": "screendump", "arguments": {"filename": out + ".ppm"}})
print(json.dumps(r))
else:
print("ok", out)
PY
ls -l "$OUT"* 2>/dev/null

33
scripts/win-ci/start-vm.sh Executable file
View File

@@ -0,0 +1,33 @@
#!/bin/bash
# win-ci VM + 호스트 제어 서버를 systemd 트랜지언트 유닛으로 띄운다.
#
# 반드시 봇(ejclaw.service) cgroup 밖에서 돌아야 한다. 안에서 돌면 8GB VM 이
# MemoryHigh=16G 를 쳐서 봇이 OOM 으로 죽는다. --user 유닛으로 띄우면
# systemd --user 의 자식이 되어 app.slice 형제로 빠지고, 에이전트 턴이
# 끝나도 계속 살아 있는다.
#
# CPUQuota=400% / CPUWeight=50 - 물리 호스트가 i7-11700(16스레드)뿐이고 .5 에서
# 운영 컨테이너 19개가 돈다. 4코어를 넘기면 .5 응답이 +140% 느려진다(실측).
set -euo pipefail
systemctl --user reset-failed win-ci.service win-ci-http.service 2>/dev/null || true
# 게스트 <-> 호스트 제어 채널 (게스트는 10.0.2.2:8099 로 본다)
if ! systemctl --user is-active --quiet win-ci-http.service; then
systemd-run --user --unit=win-ci-http \
--description="win-ci guest control channel" \
-p Restart=always \
/usr/bin/python3 /home/claude/win-ci/serve-host.py
fi
# VM
systemd-run --user --unit=win-ci \
--description="win-ci Windows CI VM (nested KVM)" \
-p MemoryMax=10G \
-p CPUQuota=400% \
-p CPUWeight=50 \
-p WorkingDirectory=/home/claude/win-ci \
/bin/bash /home/claude/win-ci/run-vm.sh
sleep 2
systemctl --user --no-pager status win-ci-http.service win-ci.service | head -40

19
scripts/win-ci/status.sh Executable file
View File

@@ -0,0 +1,19 @@
#!/bin/bash
# win-ci VM 현재 상태를 한 번에 본다.
cd /home/claude/win-ci
echo "=== 유닛 ==="
systemctl --user is-active win-ci.service win-ci-http.service | paste -sd' ' -
systemctl --user show win-ci.service -p MemoryCurrent -p CPUUsageNSec --value | paste -sd' ' -
echo
echo "=== 디스크 사용량 (설치가 진행되면 커진다) ==="
du -h --apparent-size win-ci.qcow2 2>/dev/null | cut -f1
du -h win-ci.qcow2 | cut -f1
echo
echo "=== 게스트 로그 (HTTP, 최근 25줄) ==="
tail -n 25 guest.log 2>/dev/null || echo "(아직 없음 - 네트워크 올라오기 전)"
echo
echo "=== 시리얼 로그 (최근 25줄) ==="
tail -n 25 serial.log 2>/dev/null | tr -d '\r' || echo "(아직 없음)"
echo
echo "=== 러너 등록 토큰 ==="
if [ -s serve/runner-token.txt ]; then echo "있음 ($(wc -c < serve/runner-token.txt) bytes)"; else echo "없음 - serve/runner-token.txt 에 넣으면 게스트가 20초 안에 자동 등록"; fi

67
scripts/win-ci/type.sh Executable file
View File

@@ -0,0 +1,67 @@
#!/bin/bash
# VM 게스트에 키보드 입력을 보낸다 (QMP send-key).
#
# ./type.sh text "hello world" 문자열 입력
# ./type.sh key ret 단일 키 (qcode)
# ./type.sh key alt-f4 조합키 (하이픈으로 연결)
#
# 게스트에 네트워크가 없을 때 유일한 제어 수단이므로 지우지 말 것.
set -euo pipefail
python3 - "$@" <<'PY'
import json, socket, sys, time
MODE = sys.argv[1]
ARG = sys.argv[2]
s = socket.socket(socket.AF_UNIX); s.settimeout(30)
s.connect("/home/claude/win-ci/qmp.sock"); f = s.makefile("rwb"); f.readline()
def cmd(o):
f.write((json.dumps(o) + "\n").encode()); f.flush()
while True:
m = json.loads(f.readline())
if "return" in m or "error" in m:
if "error" in m: print("ERR", m, file=sys.stderr)
return m
cmd({"execute": "qmp_capabilities"})
def send(qcodes):
cmd({"execute": "send-key", "arguments": {
"keys": [{"type": "qcode", "data": q} for q in qcodes]}})
time.sleep(0.04)
SHIFTED = {
'!': '1', '@': '2', '#': '3', '$': '4', '%': '5', '^': '6', '&': '7',
'*': '8', '(': '9', ')': '0', '_': 'minus', '+': 'equal', '{': 'bracket_left',
'}': 'bracket_right', '|': 'backslash', ':': 'semicolon', '"': 'apostrophe',
'<': 'comma', '>': 'dot', '?': 'slash', '~': 'grave_accent',
}
PLAIN = {
' ': 'spc', '-': 'minus', '=': 'equal', '[': 'bracket_left',
']': 'bracket_right', '\\': 'backslash', ';': 'semicolon',
"'": 'apostrophe', ',': 'comma', '.': 'dot', '/': 'slash',
'`': 'grave_accent', '\n': 'ret', '\t': 'tab',
}
def char_keys(c):
if c.isalpha():
q = c.lower()
return ['shift', q] if c.isupper() else [q]
if c.isdigit():
return [c]
if c in SHIFTED:
return ['shift', SHIFTED[c]]
if c in PLAIN:
return [PLAIN[c]]
raise SystemExit("unsupported char: %r" % c)
if MODE == "text":
for c in ARG:
send(char_keys(c))
elif MODE == "key":
send(ARG.split("-"))
else:
raise SystemExit("usage: type.sh text <str> | type.sh key <qcode>")
print("ok")
PY

View File

@@ -0,0 +1,190 @@
<?xml version="1.0" encoding="utf-8"?>
<unattend xmlns="urn:schemas-microsoft-com:unattend" xmlns:wcm="http://schemas.microsoft.com/WMIConfig/2002/State">
<settings pass="windowsPE">
<component name="Microsoft-Windows-International-Core-WinPE" processorArchitecture="amd64"
publicKeyToken="31bf3856ad364e35" language="neutral" versionScope="nonSxS">
<SetupUILanguage>
<UILanguage>en-US</UILanguage>
</SetupUILanguage>
<InputLocale>0409:00000409</InputLocale>
<SystemLocale>en-US</SystemLocale>
<UILanguage>en-US</UILanguage>
<UserLocale>en-US</UserLocale>
</component>
<component name="Microsoft-Windows-Setup" processorArchitecture="amd64"
publicKeyToken="31bf3856ad364e35" language="neutral" versionScope="nonSxS">
<!-- Win11 하드웨어 검사 우회. TPM/SecureBoot 는 실제로 제공하지만
중첩 가상화라 CPU/스토리지 검사가 튕길 수 있으므로 안전망으로 둔다. -->
<RunSynchronous>
<RunSynchronousCommand wcm:action="add">
<Order>1</Order>
<Path>reg add HKLM\SYSTEM\Setup\LabConfig /v BypassTPMCheck /t REG_DWORD /d 1 /f</Path>
</RunSynchronousCommand>
<RunSynchronousCommand wcm:action="add">
<Order>2</Order>
<Path>reg add HKLM\SYSTEM\Setup\LabConfig /v BypassSecureBootCheck /t REG_DWORD /d 1 /f</Path>
</RunSynchronousCommand>
<RunSynchronousCommand wcm:action="add">
<Order>3</Order>
<Path>reg add HKLM\SYSTEM\Setup\LabConfig /v BypassRAMCheck /t REG_DWORD /d 1 /f</Path>
</RunSynchronousCommand>
<RunSynchronousCommand wcm:action="add">
<Order>4</Order>
<Path>reg add HKLM\SYSTEM\Setup\LabConfig /v BypassCPUCheck /t REG_DWORD /d 1 /f</Path>
</RunSynchronousCommand>
<RunSynchronousCommand wcm:action="add">
<Order>5</Order>
<Path>reg add HKLM\SYSTEM\Setup\LabConfig /v BypassStorageCheck /t REG_DWORD /d 1 /f</Path>
</RunSynchronousCommand>
</RunSynchronous>
<DiskConfiguration>
<WillShowUI>OnError</WillShowUI>
<Disk wcm:action="add">
<DiskID>0</DiskID>
<WillWipeDisk>true</WillWipeDisk>
<CreatePartitions>
<CreatePartition wcm:action="add">
<Order>1</Order>
<Type>EFI</Type>
<Size>512</Size>
</CreatePartition>
<CreatePartition wcm:action="add">
<Order>2</Order>
<Type>MSR</Type>
<Size>16</Size>
</CreatePartition>
<CreatePartition wcm:action="add">
<Order>3</Order>
<Type>Primary</Type>
<Extend>true</Extend>
</CreatePartition>
</CreatePartitions>
<ModifyPartitions>
<ModifyPartition wcm:action="add">
<Order>1</Order>
<PartitionID>1</PartitionID>
<Label>System</Label>
<Format>FAT32</Format>
</ModifyPartition>
<ModifyPartition wcm:action="add">
<Order>2</Order>
<PartitionID>2</PartitionID>
</ModifyPartition>
<ModifyPartition wcm:action="add">
<Order>3</Order>
<PartitionID>3</PartitionID>
<Label>Windows</Label>
<Format>NTFS</Format>
<Letter>C</Letter>
</ModifyPartition>
</ModifyPartitions>
</Disk>
</DiskConfiguration>
<ImageInstall>
<OSImage>
<InstallFrom>
<MetaData wcm:action="add">
<Key>/IMAGE/INDEX</Key>
<Value>1</Value>
</MetaData>
</InstallFrom>
<InstallTo>
<DiskID>0</DiskID>
<PartitionID>3</PartitionID>
</InstallTo>
<WillShowUI>OnError</WillShowUI>
</OSImage>
</ImageInstall>
<UserData>
<AcceptEula>true</AcceptEula>
<FullName>ci</FullName>
<Organization>livesub</Organization>
</UserData>
</component>
</settings>
<settings pass="specialize">
<component name="Microsoft-Windows-Shell-Setup" processorArchitecture="amd64"
publicKeyToken="31bf3856ad364e35" language="neutral" versionScope="nonSxS">
<ComputerName>WIN-CI</ComputerName>
<TimeZone>Korea Standard Time</TimeZone>
</component>
<!-- RDP 로 붙을 수 있게 열어둔다. 호스트에서 127.0.0.1:13389 로 포워딩된다. -->
<component name="Microsoft-Windows-TerminalServices-LocalSessionManager" processorArchitecture="amd64"
publicKeyToken="31bf3856ad364e35" language="neutral" versionScope="nonSxS">
<fDenyTSConnections>false</fDenyTSConnections>
</component>
<component name="Microsoft-Windows-TerminalServices-RDP-WinStationExtensions" processorArchitecture="amd64"
publicKeyToken="31bf3856ad364e35" language="neutral" versionScope="nonSxS">
<UserAuthentication>0</UserAuthentication>
</component>
<component name="Networking-MPSSVC-Svc" processorArchitecture="amd64"
publicKeyToken="31bf3856ad364e35" language="neutral" versionScope="nonSxS">
<FirewallGroups>
<FirewallGroup wcm:action="add" wcm:keyValue="RemoteDesktop">
<Active>true</Active>
<Group>@FirewallAPI.dll,-28752</Group>
<Profile>all</Profile>
</FirewallGroup>
</FirewallGroups>
</component>
</settings>
<settings pass="oobeSystem">
<component name="Microsoft-Windows-Shell-Setup" processorArchitecture="amd64"
publicKeyToken="31bf3856ad364e35" language="neutral" versionScope="nonSxS">
<OOBE>
<HideEULAPage>true</HideEULAPage>
<HideOEMRegistrationScreen>true</HideOEMRegistrationScreen>
<HideOnlineAccountScreens>true</HideOnlineAccountScreens>
<HideWirelessSetupInOOBE>true</HideWirelessSetupInOOBE>
<NetworkLocation>Work</NetworkLocation>
<ProtectYourPC>3</ProtectYourPC>
<SkipMachineOOBE>true</SkipMachineOOBE>
<SkipUserOOBE>true</SkipUserOOBE>
</OOBE>
<UserAccounts>
<LocalAccounts>
<LocalAccount wcm:action="add">
<Name>ci</Name>
<DisplayName>ci</DisplayName>
<Group>Administrators</Group>
<Password>
<Value>LiveSubCI!2026</Value>
<PlainText>true</PlainText>
</Password>
</LocalAccount>
</LocalAccounts>
</UserAccounts>
<AutoLogon>
<Username>ci</Username>
<Enabled>true</Enabled>
<LogonCount>999</LogonCount>
<Password>
<Value>LiveSubCI!2026</Value>
<PlainText>true</PlainText>
</Password>
</AutoLogon>
<FirstLogonCommands>
<SynchronousCommand wcm:action="add">
<Order>1</Order>
<Description>provision windows ci runner</Description>
<RequiresUserInput>false</RequiresUserInput>
<CommandLine>cmd.exe /c "for %d in (D E F G H I J K L M) do @if exist %d:\setup\provision.ps1 start /wait powershell.exe -NoProfile -ExecutionPolicy Bypass -File %d:\setup\provision.ps1"</CommandLine>
</SynchronousCommand>
</FirstLogonCommands>
<TimeZone>Korea Standard Time</TimeZone>
</component>
</settings>
</unattend>

View File

@@ -0,0 +1,96 @@
# Windows CI VM 상주 에이전트
#
# 하는 일은 하나다: 호스트가 주는 stage2.ps1 을 받아서 돌린다.
#
# 왜 이렇게 하나 - ISO 는 한 번 구우면 못 고친다. 설치 스크립트를 ISO 에
# 박아두면 한 줄 고칠 때마다 Windows 를 다시 설치해야 한다. 그래서 ISO 에는
# 이 껍데기만 넣고, 실제 설치 내용은 호스트의 HTTP 로 가져온다.
# QEMU user networking 에서 호스트는 항상 10.0.2.2 다.
#
# 로그는 COM1(호스트 serial.log) + 호스트 HTTP(POST /log) + 로컬 파일.
$ErrorActionPreference = 'Continue'
[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
$Host_ = 'http://10.0.2.2:8099'
$WorkDir = 'C:\win-ci'
$Stage2 = Join-Path $WorkDir 'stage2.ps1'
$HashFile = Join-Path $WorkDir 'stage2.hash'
$LogFile = Join-Path $WorkDir 'agent.log'
New-Item -ItemType Directory -Force -Path $WorkDir | Out-Null
function Say([string]$m) {
$line = "[agent $(Get-Date -Format 'yyyy-MM-dd HH:mm:ss')] $m"
try { Add-Content -Path $LogFile -Value $line -Encoding UTF8 } catch {}
try {
$p = New-Object System.IO.Ports.SerialPort 'COM1', 115200, 'None', 8, 'One'
# 기본 인코딩이 ASCII 라 한글이 전부 '?' 가 된다
$p.Encoding = [Text.Encoding]::UTF8
$p.Open(); $p.WriteLine($line); $p.Close()
} catch {}
try {
Invoke-RestMethod -Uri "$Host_/log" -Method Post -TimeoutSec 5 `
-Body ([Text.Encoding]::UTF8.GetBytes($line)) -ContentType "text/plain; charset=utf-8" | Out-Null
} catch {}
}
# 같은 에이전트가 두 개 돌면 설치가 겹친다. 뮤텍스로 하나만 남긴다.
$mutex = New-Object System.Threading.Mutex($false, 'Global\win-ci-agent')
if (-not $mutex.WaitOne(0)) { Say "이미 다른 agent 가 돌고 있다. 종료."; exit 0 }
Say "agent 시작. user=$env:USERNAME"
# 네트워크가 올라오길 기다린다. virtio-net 드라이버 설치 직후라 몇 초 걸린다.
$netOk = $false
for ($i = 0; $i -lt 60; $i++) {
try {
Invoke-WebRequest -Uri "$Host_/ping" -TimeoutSec 4 -UseBasicParsing | Out-Null
$netOk = $true; break
} catch { Start-Sleep -Seconds 5 }
}
if ($netOk) { Say "호스트 연결 OK ($Host_)" }
else { Say "ERROR 5분 동안 호스트에 못 붙었다. virtio-net 드라이버를 확인해야 한다." }
# 이후로는 계속 돌면서 stage2 가 바뀔 때마다 실행한다.
while ($true) {
try {
# 반드시 바이트로 받아 UTF-8 로 직접 디코딩한다. Invoke-WebRequest 의
# .Content 는 Content-Type 에 charset 이 없으면 ISO-8859-1 로 디코딩해
# 한글 주석이 깨지고, 깨진 바이트가 CP1252 의 스마트따옴표(0x93/0x94)가
# 되어 PowerShell 파서가 그 뒤를 전부 문자열로 삼킨다. 실제로 겪은 사고다.
$wc = New-Object Net.WebClient
$body = [Text.Encoding]::UTF8.GetString($wc.DownloadData("$Host_/stage2.ps1"))
# 받은 쪽에 BOM 이 있으면 문자열 맨 앞에 U+FEFF 가 남는다. 아래에서
# Set-Content -Encoding UTF8 이 BOM 을 또 붙이므로 BOM 이 두 개가 되고,
# PowerShell 은 첫 개만 벗겨내 남은 U+FEFF 때문에 1행이 주석으로
# 인식되지 않는다. 그래서 여기서 반드시 떼어낸다.
$body = $body.TrimStart([char]0xFEFF)
$sha = [BitConverter]::ToString(
[Security.Cryptography.SHA256]::Create().ComputeHash([Text.Encoding]::UTF8.GetBytes($body))
).Replace('-', '')
$prev = if (Test-Path $HashFile) { (Get-Content $HashFile -Raw).Trim() } else { '' }
if ($sha -ne $prev) {
Say "stage2 새 버전 감지 ($($sha.Substring(0,12))) - 실행"
Set-Content -Path $Stage2 -Value $body -Encoding UTF8
try {
& powershell.exe -NoProfile -ExecutionPolicy Bypass -File $Stage2 2>&1 |
ForEach-Object { Say "stage2| $_" }
if ($LASTEXITCODE -eq 0 -or $null -eq $LASTEXITCODE) {
# 성공했을 때만 해시를 저장한다. 중간에 죽었으면 다음 턴에 다시 돈다.
Set-Content -Path $HashFile -Value $sha
Say "stage2 성공"
} else {
Say "stage2 실패 exit=$LASTEXITCODE - 다음 폴링에 재시도"
}
} catch {
Say "stage2 예외: $($_.Exception.Message) - 다음 폴링에 재시도"
}
}
} catch {
Say "폴링 실패: $($_.Exception.Message)"
}
Start-Sleep -Seconds 20
}

View File

@@ -0,0 +1,110 @@
# Windows CI VM 최초 부팅 프로비저닝 (무인)
#
# 이 스크립트는 autounattend.xml 의 FirstLogonCommands 로 한 번만 실행된다.
# 네트워크가 아직 없는 상태에서 시작하므로, 순서가 중요하다.
# 1) virtio 드라이버 설치 (ISO 에서 오프라인) -> 이걸 해야 네트워크가 생긴다
# 2) 화면 꺼짐/잠금 끄기 (GUI 테스트가 잠금화면에서 실패하지 않게)
# 3) agent.ps1 을 C:\win-ci 에 심고 로그온 예약 작업으로 등록
# 4) agent 를 바로 실행 -> 이후 모든 설치는 호스트가 주는 stage2.ps1 이 한다
#
# 로그는 COM1(호스트의 serial.log)과 C:\win-ci\provision.log 양쪽에 남긴다.
# 네트워크가 없는 1) 단계를 호스트에서 관찰할 수 있어야 하기 때문이다.
$ErrorActionPreference = 'Continue'
New-Item -ItemType Directory -Force -Path 'C:\win-ci' | Out-Null
$script:Port = $null
try {
$script:Port = New-Object System.IO.Ports.SerialPort 'COM1', 115200, 'None', 8, 'One'
# 기본 인코딩이 ASCII 라 한글이 전부 '?' 가 된다
$script:Port.Encoding = [Text.Encoding]::UTF8
$script:Port.Open()
} catch { $script:Port = $null }
function Say([string]$m) {
$line = "[provision $(Get-Date -Format 'yyyy-MM-dd HH:mm:ss')] $m"
try { Add-Content -Path 'C:\win-ci\provision.log' -Value $line -Encoding UTF8 } catch {}
if ($script:Port) { try { $script:Port.WriteLine($line) } catch {} }
}
Say "start. user=$env:USERNAME computer=$env:COMPUTERNAME"
# ---------------------------------------------------------------- 1) virtio
# virtio-win ISO 가 몇 번째 드라이브로 잡힐지 모른다. 전부 훑어 찾는다.
$virtio = $null
foreach ($d in [char[]](68..77)) { # D..M
$p = "${d}:\virtio-win-guest-tools.exe"
if (Test-Path $p) { $virtio = "${d}:"; break }
}
if (-not $virtio) {
Say "ERROR virtio-win ISO 를 못 찾음. 네트워크/화면 드라이버 없이 계속한다."
} else {
Say "virtio ISO = $virtio"
# 드라이버 스토어에 직접 넣는 쪽을 먼저 한다. 설치 프로그램보다 확실하다.
$infs = @(
"$virtio\NetKVM\w11\amd64\netkvm.inf", # 네트워크 - 가장 중요
"$virtio\viogpudo\w11\amd64\viogpudo.inf", # 화면 - 모니터 2개에 필요
"$virtio\Balloon\w11\amd64\balloon.inf",
"$virtio\vioserial\w11\amd64\vioser.inf",
"$virtio\viostor\w11\amd64\viostor.inf",
"$virtio\vioscsi\w11\amd64\vioscsi.inf",
"$virtio\vioinput\w11\amd64\vioinput.inf"
)
foreach ($inf in $infs) {
if (Test-Path $inf) {
$out = & pnputil.exe /add-driver $inf /install 2>&1 | Out-String
Say "pnputil $(Split-Path $inf -Leaf) -> $($out.Trim() -replace '\s+', ' ')"
} else {
Say "skip (없음) $inf"
}
}
# guest-tools 는 balloon 서비스/qemu-ga 까지 붙여준다. 실패해도 위에서 드라이버는 들어갔다.
try {
Say "virtio-win-guest-tools 설치 시작"
$p = Start-Process -FilePath "$virtio\virtio-win-guest-tools.exe" `
-ArgumentList '/install', '/quiet', '/norestart' -Wait -PassThru
Say "virtio-win-guest-tools exit=$($p.ExitCode)"
} catch {
Say "virtio-win-guest-tools 실패: $($_.Exception.Message)"
}
}
# ---------------------------------------------------- 2) 화면 꺼짐 / 잠금 끄기
# 러너가 GUI 테스트와 스크린샷을 찍으므로 데스크톱이 살아 있어야 한다.
& powercfg.exe /change monitor-timeout-ac 0
& powercfg.exe /change standby-timeout-ac 0
& powercfg.exe /change disk-timeout-ac 0
& powercfg.exe /change hibernate-timeout-ac 0
Say "powercfg 적용 (모니터/대기/디스크/최대절전 전부 끔)"
# 화면 보호기 + 잠금화면
reg add "HKCU\Control Panel\Desktop" /v ScreenSaveActive /t REG_SZ /d 0 /f | Out-Null
reg add "HKCU\Control Panel\Desktop" /v ScreenSaverIsSecure /t REG_SZ /d 0 /f | Out-Null
reg add "HKLM\SOFTWARE\Policies\Microsoft\Windows\Personalization" /v NoLockScreen /t REG_DWORD /d 1 /f | Out-Null
# 절전 시 암호 요구 끔
reg add "HKLM\SOFTWARE\Policies\Microsoft\Power\PowerSettings\0e796bdb-100d-47d6-a2d5-f7d2daa51f51" /v ACSettingIndex /t REG_DWORD /d 0 /f | Out-Null
Say "잠금화면/화면보호기 끔"
# 첫 부팅 때 뜨는 방해 요소들
reg add "HKLM\SOFTWARE\Microsoft\Windows\CurrentVersion\Policies\Explorer" /v NoDriveTypeAutoRun /t REG_DWORD /d 255 /f | Out-Null
reg add "HKLM\SOFTWARE\Policies\Microsoft\Windows\WindowsUpdate\AU" /v NoAutoRebootWithLoggedOnUsers /t REG_DWORD /d 1 /f | Out-Null
Say "자동실행/업데이트 자동재부팅 끔"
# ------------------------------------------- 3) agent 설치 + 로그온 예약 작업
$src = $PSScriptRoot
Copy-Item -Path (Join-Path $src 'agent.ps1') -Destination 'C:\win-ci\agent.ps1' -Force
Say "agent.ps1 복사 완료"
$taskCmd = 'powershell.exe'
$taskArg = '-NoProfile -ExecutionPolicy Bypass -WindowStyle Hidden -File C:\win-ci\agent.ps1'
& schtasks.exe /Create /TN 'win-ci-agent' /SC ONLOGON /RL HIGHEST /TR "$taskCmd $taskArg" /F 2>&1 | Out-String | ForEach-Object { Say "schtasks -> $($_.Trim())" }
# ------------------------------------------------------------- 4) agent 기동
Say "agent 시작 - 이후 로그는 [agent] 접두어로 나온다"
Start-Process -FilePath $taskCmd -ArgumentList $taskArg -WindowStyle Hidden
Say "provision 끝"
if ($script:Port) { try { $script:Port.Close() } catch {} }

View File

@@ -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",
]

95
src/livesub/models/asr.py Normal file
View File

@@ -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),
)

View File

@@ -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}"

View File

@@ -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

152
src/livesub/models/packs.py Normal file
View File

@@ -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

View File

@@ -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)

209
src/livesub/models/tiers.py Normal file
View File

@@ -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])

View File

@@ -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 <zh>
끝의 `<언어코드>` 태그는 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)