From b721ca64191db2e67a9f587595b1261cf0484897 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=B0=95=EC=83=81=ED=98=B8=20Sangho=20Park?= Date: Tue, 9 Jun 2026 07:09:51 +0900 Subject: [PATCH] feat(api): chunk LLM correction for small context windows (+running glossary) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 사내 GPT-4o 컨텍스트(<30k)에 맞춰 긴 전사를 문장 경계로 청크 분할하고, 각 청크 보정의 영문 용어를 '러닝 글로서리'로 다음 청크 system에 전달 → 큰 창 없이 강연 전체 용어 일관성 유지. config.llm_max_chars(기본 3000; ~8k창→1500/~16k→3000/~30k→6000). 과대 단일문장은 글자단위 강제 분할 안전망. 23 tests pass(청크 분할/글로서리 주입 포함), ruff clean. Co-Authored-By: Claude Opus 4.8 --- src/luke_scribe/api/routes/transcribe.py | 1 + src/luke_scribe/config.py | 2 + src/luke_scribe/postprocess/llm.py | 132 +++++++++++++++++------ tests/test_postprocess.py | 18 ++++ 4 files changed, 123 insertions(+), 30 deletions(-) diff --git a/src/luke_scribe/api/routes/transcribe.py b/src/luke_scribe/api/routes/transcribe.py index d9e8f7b..5b707ed 100644 --- a/src/luke_scribe/api/routes/transcribe.py +++ b/src/luke_scribe/api/routes/transcribe.py @@ -93,6 +93,7 @@ def transcribe_ep( # noqa: PLR0913 — 요청 옵션 다수(스펙 options 스 base_url=settings.llm_base_url, api_key=settings.llm_api_key, model=settings.llm_model, + max_chars=settings.llm_max_chars, ) ) corrected = True diff --git a/src/luke_scribe/config.py b/src/luke_scribe/config.py index f9761c3..652ce97 100644 --- a/src/luke_scribe/config.py +++ b/src/luke_scribe/config.py @@ -43,6 +43,8 @@ class Settings(BaseSettings): llm_base_url: str | None = None # 예: http://192.168.0.123:8080/v1 (allowlist=이 endpoint만) llm_api_key: str | None = None # env SCRIBE_LLM_API_KEY 로만 주입 llm_model: str = "copilot-gpt-4o" + # 보정 청크 크기(글자) — 사내 LLM 컨텍스트 창에 맞춰 조정 (예: ~8k창→1500, ~16k→3000, ~30k→6000) + llm_max_chars: int = 3000 settings = Settings() diff --git a/src/luke_scribe/postprocess/llm.py b/src/luke_scribe/postprocess/llm.py index 0d191c4..e0c4e1f 100644 --- a/src/luke_scribe/postprocess/llm.py +++ b/src/luke_scribe/postprocess/llm.py @@ -1,13 +1,17 @@ """LLM 보정 (스펙 §7 stage 3 / §3.8) — 음차된 영문 용어를 문맥+지식으로 복원. -OpenAI 호환 백엔드(사내/로컬). **opt-in**(요청 correct=true에서만 호출), **allowlist**(설정된 -base_url만), **감사로그**(호출 1줄). transient(연결 reset/timeout) 재시도. -긴 입력 청크/러닝글로서리는 TODO — MVP는 단일 호출(짧은 클립엔 충분). +작은 컨텍스트 창 대응(사내 GPT-4o < 30k 토큰): 긴 전사는 **문장 경계로 청크 분할**, +각 청크를 순차 보정하며 **이미 확정된 영문 표기(러닝 글로서리)** 를 다음 청크로 전달 → +큰 창 없이도 강연 전체 용어 일관성 유지. + +OpenAI 호환 백엔드(사내/로컬). **opt-in**(요청 correct=true) · **allowlist**(설정 base_url만) · +**감사로그**(호출 요약 1줄). transient(연결 reset/timeout) 재시도. """ from __future__ import annotations import json import logging +import re import time import urllib.error import urllib.request @@ -20,47 +24,115 @@ SYSTEM = ( "일반 한국어는 그대로 두고, 확실하지 않으면 바꾸지 마라. 설명 없이 교정된 전사문만 출력하라." ) +_SENT_RE = re.compile(r"(?<=[.!?。…\n])\s+") # 문장 경계 +_TERM_RE = re.compile(r"[A-Za-z][A-Za-z0-9.+/#-]{1,}") # 러닝 글로서리용 영문 토큰 +_GLOSSARY_CAP = 60 + class LLMNotConfigured(RuntimeError): """llm_base_url / llm_api_key 미설정.""" +def _chunk(text: str, max_chars: int) -> list[str]: + """문장 경계로 max_chars 이하 청크 패킹. 한 문장이 과대하면 글자 단위 강제 분할.""" + if len(text) <= max_chars: + return [text] + packed: list[str] = [] + cur = "" + for part in _SENT_RE.split(text): + if not part: + continue + if cur and len(cur) + len(part) + 1 > max_chars: + packed.append(cur) + cur = part + else: + cur = f"{cur} {part}" if cur else part + if cur: + packed.append(cur) + out: list[str] = [] + for c in packed: # 안전망: 단일 문장이 너무 길면 글자 단위 강제 분할 + if len(c) > max_chars: + out.extend(c[i : i + max_chars] for i in range(0, len(c), max_chars)) + else: + out.append(c) + return out + + +def _terms(text: str) -> list[str]: + seen: dict[str, None] = {} + for m in _TERM_RE.finditer(text): + seen.setdefault(m.group(0), None) + return list(seen) + + +def _request( + messages: list[dict], + *, + url: str, + api_key: str, + model: str, + retries: int, + timeout: float, +) -> str: + payload = {"model": model, "temperature": 0, "messages": messages} + req = urllib.request.Request( + url, + data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json", "Authorization": "Bearer " + api_key}, + ) + for attempt in range(1, retries + 1): + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read())["choices"][0]["message"]["content"] + except urllib.error.HTTPError: + raise # 실제 HTTP 응답(401/4xx) — 재시도 무의미 + except (urllib.error.URLError, OSError): # transient + if attempt == retries: + raise + time.sleep(1.0 * attempt) + raise RuntimeError("unreachable") + + def correct( text: str, *, base_url: str | None, api_key: str | None, model: str = "copilot-gpt-4o", + max_chars: int = 3000, retries: int = 4, timeout: float = 90.0, ) -> str: + """음차 영문 용어 복원. max_chars로 청크 분할(작은 컨텍스트 창 대응).""" if not base_url or not api_key: - raise LLMNotConfigured("llm_base_url/llm_api_key 미설정 — correct를 쓰려면 SCRIBE_LLM_* 설정 필요") + raise LLMNotConfigured("llm_base_url/llm_api_key 미설정 — correct에 SCRIBE_LLM_* 필요") url = base_url.rstrip("/") + "/chat/completions" - payload = { - "model": model, - "temperature": 0, - "messages": [ - {"role": "system", "content": SYSTEM}, - {"role": "user", "content": text}, - ], - } - req = urllib.request.Request( - url, - data=json.dumps(payload).encode(), - headers={"Content-Type": "application/json", "Authorization": "Bearer " + api_key}, + chunks = _chunk(text, max_chars) + logger.info( + "llm-correct egress endpoint=%s model=%s chars=%d chunks=%d", + url, model, len(text), len(chunks), ) - # 감사로그 (allowlist=설정 endpoint, 호출 1줄) - logger.info("llm-correct egress endpoint=%s model=%s chars=%d", url, model, len(text)) - for attempt in range(1, retries + 1): - try: - with urllib.request.urlopen(req, timeout=timeout) as resp: - data = json.loads(resp.read()) - return data["choices"][0]["message"]["content"] - except urllib.error.HTTPError: - raise # 실제 HTTP 응답(401/4xx) — 재시도 무의미 - except (urllib.error.URLError, OSError): # 연결 reset/timeout 등 transient - if attempt == retries: - raise - time.sleep(1.0 * attempt) - raise RuntimeError("unreachable") + glossary: dict[str, None] = {} + out: list[str] = [] + for chunk in chunks: + system = SYSTEM + if glossary: + system += ( + "\n이미 이 전사에서 확정된 영문 표기: " + + ", ".join(glossary) + + ". 같은/유사 용어는 이 표기로 통일하라." + ) + corrected = _request( + [{"role": "system", "content": system}, {"role": "user", "content": chunk}], + url=url, + api_key=api_key, + model=model, + retries=retries, + timeout=timeout, + ) + out.append(corrected) + for term in _terms(corrected): + glossary.setdefault(term, None) + if len(glossary) > _GLOSSARY_CAP: + glossary = dict(list(glossary.items())[-_GLOSSARY_CAP:]) + return " ".join(out).strip() diff --git a/tests/test_postprocess.py b/tests/test_postprocess.py index 225ee19..a397079 100644 --- a/tests/test_postprocess.py +++ b/tests/test_postprocess.py @@ -39,3 +39,21 @@ def test_llm_correct_monkeypatched(monkeypatch): monkeypatch.setattr(llm.urllib.request, "urlopen", fake_urlopen) out = llm.correct("인베딩 점마", base_url="http://x/v1", api_key="k", model="m") assert out == "EmbeddingGemma 복원됨" + + +def test_llm_chunking_and_glossary(monkeypatch): + """긴 입력 → 청크 분할 + 러닝 글로서리(작은 컨텍스트 창 대응).""" + calls: list[list[dict]] = [] + + def fake_request(messages, **_kw): + calls.append(messages) + return messages[1]["content"] # 청크 그대로 echo + + monkeypatch.setattr(llm, "_request", fake_request) + long_text = ". ".join(f"문장{i} EmbeddingGemma 설명" for i in range(400)) + out = llm.correct(long_text, base_url="http://x/v1", api_key="k", max_chars=200) + + assert len(calls) > 1 # 분할됨 + assert "EmbeddingGemma" in out # 재조립됨 + # 2번째 청크부터 이전에 확정된 영문 표기가 system에 주입됨 + assert any("확정된 영문 표기" in m[0]["content"] for m in calls[1:])