diff --git a/summarize/llm_client.py b/summarize/llm_client.py index da2d2e1..b164f56 100644 --- a/summarize/llm_client.py +++ b/summarize/llm_client.py @@ -43,15 +43,98 @@ class PendingResponse(Exception): self.response_path = response_path +def _repair_json(text: str) -> str: + """모델이 흔히 내는 JSON 문법 오류를 고친다 (고객사 사내 LLM 실측: 문자열 안의 따옴표 미이스케이프). + + - 문자열 안의 `"` 가 닫는 따옴표가 아니면(뒤에 , : } ] 가 안 오면) `\\"` 로 바꾼다 + - 문자열 안의 실제 줄바꿈·탭은 `\\n` `\\t` 로 + - 닫는 괄호 앞의 꼬리 콤마 제거 + - 출력이 잘린 경우: 열린 문자열·괄호를 닫아 준다 (부분 결과라도 스키마 검증에 맡긴다) + """ + out: list[str] = [] + stack: list[str] = [] + in_str = False + i, n = 0, len(text) + while i < n: + ch = text[i] + if in_str: + if ch == "\\": + out.append(text[i : i + 2]); i += 2; continue + if ch == '"': + j = i + 1 + while j < n and text[j] in " \t\r\n": + j += 1 + if j >= n or text[j] in ",:}]": + in_str = False + out.append(ch) + else: + out.append('\\"') # 값 안의 따옴표 + elif ch == "\n": + out.append("\\n") + elif ch == "\t": + out.append("\\t") + elif ch == "\r": + pass + else: + out.append(ch) + else: + if ch == '"': + in_str = True + out.append(ch) + elif ch in "{[": + stack.append("}" if ch == "{" else "]") + out.append(ch) + elif ch in "}]": + # 꼬리 콤마 제거 + k = len(out) - 1 + while k >= 0 and out[k].strip() == "": + k -= 1 + if k >= 0 and out[k] == ",": + del out[k] + if stack: + stack.pop() + out.append(ch) + else: + out.append(ch) + i += 1 + if in_str: + out.append('"') + while stack: + k = len(out) - 1 + while k >= 0 and out[k].strip() == "": + k -= 1 + if k >= 0 and out[k] == ",": + del out[k] + out.append(stack.pop()) + return "".join(out) + + def _extract_json(content: str) -> dict: - """모델이 코드펜스/서문을 붙이는 경우까지 감안해 JSON 을 뽑아낸다.""" + """모델이 코드펜스/서문을 붙이거나 문법이 약간 틀린 JSON 을 내는 경우까지 감안해 뽑아낸다. + + 끝내 못 읽으면 원문을 data/llm_jobs/_badjson/ 에 남기고 JSONDecodeError 를 올린다. + """ try: return json.loads(content) - except json.JSONDecodeError: + except json.JSONDecodeError as first: s, e = content.find("{"), content.rfind("}") - if s >= 0 and e > s: - return json.loads(content[s : e + 1]) - raise + candidate = content[s : e + 1] if (s >= 0 and e > s) else content[s:] if s >= 0 else content + for attempt in (candidate, _repair_json(candidate)): + try: + data = json.loads(attempt) + if isinstance(data, dict): + return data + except json.JSONDecodeError: + continue + try: + d = settings.data_llm_jobs / "_badjson" + d.mkdir(parents=True, exist_ok=True) + name = hashlib.sha256(content.encode("utf-8")).hexdigest()[:12] + (d / f"{name}.txt").write_text(content, encoding="utf-8") + print(f"[BADJSON] 모델 응답을 JSON 으로 읽지 못함 — 원문: {d / (name + '.txt')}") + except OSError: + pass + raise first class OpenAICompatClient(LLMClient): diff --git a/tests/test_llm_backends.py b/tests/test_llm_backends.py index 40e8b0a..b086206 100644 --- a/tests/test_llm_backends.py +++ b/tests/test_llm_backends.py @@ -274,3 +274,32 @@ def test_merge_system_folds_system_into_user(monkeypatch): assert msgs == [{"role": "user", "content": "SYS\n\nUSR"}] monkeypatch.setattr(settings, "llm_merge_system", False) assert [m["role"] for m in llm_client.OpenAICompatClient().body("SYS", "USR")["messages"]] == ["system", "user"] + + +# ------------------------------------------------------------------ 문법이 틀린 JSON 복구 + +def test_extract_json_repairs_common_model_mistakes(tmp_path, monkeypatch): + from config.settings import settings + monkeypatch.setattr(settings, "data_llm_jobs", tmp_path) + # 값 안의 따옴표 미이스케이프 (고객사 실측: Expecting ',' delimiter) + bad = '{"unit_purpose_ko": "전표를 "확정" 상태로 바꾼다", "chunks": [{"kind": "db_write", "purpose_ko": "BKPF 갱신"}]}' + assert _extract_json(bad)["unit_purpose_ko"] == '전표를 "확정" 상태로 바꾼다' + # 문자열 안의 실제 줄바꿈 + 꼬리 콤마 + bad2 = '{"a": "줄1\n줄2", "b": [1, 2,],}' + assert _extract_json(bad2) == {"a": "줄1\n줄2", "b": [1, 2]} + # 출력이 잘린 경우 — 열린 것을 닫아 부분 결과라도 돌려준다 + cut = '{"unit_purpose_ko": "요약", "chunks": [{"kind": "select", "purpose_ko": "조회' + assert _extract_json(cut) == {"unit_purpose_ko": "요약", "chunks": [{"kind": "select", "purpose_ko": "조회"}]} + # 코드펜스 + 서문 + 따옴표 문제가 같이 있는 경우 + assert _extract_json('답입니다:\n```json\n{"x": "a "b" c"}\n```')["x"] == 'a "b" c' + # 정상 JSON 은 그대로 + assert _extract_json('{"ok": true}') == {"ok": True} + + +def test_extract_json_dumps_unreadable_content(tmp_path, monkeypatch): + from config.settings import settings + monkeypatch.setattr(settings, "data_llm_jobs", tmp_path) + with pytest.raises(json.JSONDecodeError): + _extract_json("안녕하세요! 반갑습니다.") + dumped = list((tmp_path / "_badjson").glob("*.txt")) + assert len(dumped) == 1 and "안녕하세요" in dumped[0].read_text(encoding="utf-8")