Files
luke_scribe/scripts/llm_correct.py
T
lukehemmin 8f6f8969fd feat(api): sync test API (serve) + opt-in LLM correction + cloudflared tunnel
- api/: FastAPI app, X-API-Key 인증(미설정 시 임시키), 엔진 load-once 풀
  (+transcribe lock), POST /v1/transcribe(multipart, 동기), /health, /v1/system,
  /v1/models. 업로드 임시파일 finally 삭제(프라이버시).
- postprocess/: llm.correct(scripts/llm_correct.py 승격; opt-in·allowlist·감사로그·재시도)
  + rules.normalize(EmbeddingGemma 등 정규화).
- results/formats.py: txt/srt/vtt. connectivity/tunnel.py: cloudflared quick tunnel(Colab).
- cli serve: uvicorn 단일워커 + --tunnel cloudflare; config llm_* 필드;
  pyproject api/queue extra 분리(+python-multipart, dev httpx).

검증: 22 단위테스트(API TestClient·formats·postprocess) + 실서버 e2e
(/health·auth 401·실제 전사(JFK)·SRT·임시파일 삭제). KO 품질은 turbo/large-v3 필요(tiny는 한국어 degenerate).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-08 23:20:01 +09:00

83 lines
3.4 KiB
Python

#!/usr/bin/env python3
"""STT 후처리 PoC — 음차된 영문 기술용어를 사내 LLM(OpenAI 호환)으로 복원.
게이트가 닿는 환경에서 실행:
export SCRIBE_LLM_BASE_URL=http://localhost:8080/v1
export SCRIBE_LLM_API_KEY=<사내 키>
export SCRIBE_LLM_MODEL=copilot-gpt-4o
python3 scripts/llm_correct.py # 내장 샘플로 데모
python3 scripts/llm_correct.py < my.txt # 임의 전사 교정
외부 의존성 없음(urllib). 향후 postprocess/llm.py(confidence-gated, 청크/러닝글로서리)로 발전.
"""
from __future__ import annotations
import json
import os
import sys
import time
import urllib.error
import urllib.request
SYSTEM = (
"너는 한국어 STT 전사 후처리기다. 한국어 음성에 섞여 나온 영어 기술용어·고유명사가 "
"발음대로 한글로 음차되어 잘못 적힌 부분을 문맥과 지식으로 원래 영어 표기로 복원하라. "
"일반 한국어는 그대로 두고, 확실하지 않으면 바꾸지 마라. 설명 없이 교정된 전사문만 출력하라."
)
# turbo가 망친 실제 전사(EmbeddingGemma 강연) — 내장 데모용
SAMPLE = (
"그래서 오늘 준비한 내용은 기본적으로 인베딩 점마에 대해서 설명을 드릴 텐데요. "
"여러분들이 알고 계시는 랭기징 모델이 정말 사람이 생각하는 것처럼 하는데 "
"그 다음에 구글에 런칭한 오픈모델입니다. 인베딩 점마 라는 것을 소개를 해드릴 예정입니다. "
"그리고 어 재미나이 하고 이제 점마하고 두 가지가 있는데요. "
"구글 포 디벨로퍼스 사이트에 가시면 제가 올린 포스트도 보실 수 있는데."
)
def correct(text: str) -> str:
base = os.environ.get("SCRIBE_LLM_BASE_URL", "http://localhost:8080/v1").rstrip("/")
key = os.environ.get("SCRIBE_LLM_API_KEY", "")
model = os.environ.get("SCRIBE_LLM_MODEL", "copilot-gpt-4o")
payload = {
"model": model,
"temperature": 0,
"messages": [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": text},
],
}
req = urllib.request.Request(
base + "/chat/completions",
data=json.dumps(payload).encode(),
headers={"Content-Type": "application/json", "Authorization": "Bearer " + key},
)
retries = 4
for attempt in range(1, retries + 1):
try:
with urllib.request.urlopen(req, timeout=90) as resp:
return json.loads(resp.read())["choices"][0]["message"]["content"]
except urllib.error.HTTPError:
raise # 실제 HTTP 응답(401/400 등) — 재시도 무의미
except (urllib.error.URLError, OSError) as exc: # 연결 reset/timeout 등 transient
if attempt == retries:
raise
print(f" [retry {attempt}/{retries - 1}] {type(exc).__name__} → 재시도", file=sys.stderr)
time.sleep(1.5 * attempt)
raise RuntimeError("unreachable")
def main() -> None:
src = (sys.stdin.read().strip() if not sys.stdin.isatty() else "") or SAMPLE
print("=== 원본 ===\n" + src + "\n\n=== 교정 ===")
try:
print(correct(src))
except urllib.error.HTTPError as exc:
sys.exit(f"HTTP {exc.code}: {exc.read().decode()[:300]}")
except Exception as exc: # noqa: BLE001
sys.exit(f"{type(exc).__name__}: {exc}")
if __name__ == "__main__":
main()