byeongwook.choiandClaude Fable 5.1
d8718c858f
고객사 사내 LLM 실측: "Expecting ',' delimiter" — 설명 문장 안의 따옴표를 이스케이프하지 않았다. _repair_json 이 문자열 안의 따옴표를 닫는 따옴표(뒤에 , : } ])와 구분해 이스케이프하고, 끝내 못 읽으면 data/llm_jobs/_badjson/ 에 원문을 남긴다. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
306 lines
13 KiB
Python
306 lines
13 KiB
Python
"""LLM 백엔드 검증 — file 큐(키 불필요 입구)와 재시도/백오프."""
|
|
import json
|
|
import urllib.error
|
|
|
|
import pytest
|
|
|
|
from summarize import jobs
|
|
from summarize.llm_client import (
|
|
FileQueueLLM,
|
|
PendingResponse,
|
|
_extract_json,
|
|
_meta_from_prompt,
|
|
create_llm,
|
|
job_id,
|
|
)
|
|
|
|
UNIT_PROMPT_SAMPLE = """[작업] extract_chunks
|
|
[프로그램] ZFIR10070 — 총계정원장 조회
|
|
[단위] unit_id: ZFIR10070#ZFIR10070_F01#FORM#SELECT_DATA
|
|
unit_type: FORM, name: SELECT_DATA
|
|
[창] 이 단위는 길어서 3개 창으로 나눠 보여준다. 지금은 2번째 창 (L10-L20).
|
|
"""
|
|
|
|
|
|
def test_job_id_is_stable_and_content_addressed():
|
|
a = job_id("sys", "user")
|
|
assert a == job_id("sys", "user")
|
|
assert a != job_id("sys", "user2")
|
|
|
|
|
|
def test_meta_parsed_from_prompt():
|
|
m = _meta_from_prompt(UNIT_PROMPT_SAMPLE)
|
|
assert m["task"] == "extract_chunks"
|
|
assert m["program"] == "ZFIR10070"
|
|
assert m["unit_id"] == "ZFIR10070#ZFIR10070_F01#FORM#SELECT_DATA"
|
|
assert m["unit_type"] == "FORM"
|
|
assert m["window"] == "2"
|
|
|
|
|
|
def test_file_backend_emits_prompt_then_reads_answer(tmp_path):
|
|
llm = FileQueueLLM(tmp_path)
|
|
|
|
# 1) 응답이 없으면 프롬프트를 내놓고 PendingResponse
|
|
with pytest.raises(PendingResponse) as ei:
|
|
llm.complete_json("sys", UNIT_PROMPT_SAMPLE)
|
|
jid = ei.value.job_id
|
|
assert llm.prompt_path(jid).exists()
|
|
prompt_text = llm.prompt_path(jid).read_text(encoding="utf-8")
|
|
assert "## SYSTEM" in prompt_text and "extract_chunks" in prompt_text
|
|
# 인덱스에 메타가 기록된다
|
|
rows = [json.loads(l) for l in (tmp_path / "index.jsonl").read_text(encoding="utf-8").splitlines()]
|
|
assert rows[0]["job_id"] == jid and rows[0]["program"] == "ZFIR10070"
|
|
|
|
# 2) 응답을 채우면 그걸 돌려준다
|
|
llm.response_path(jid).write_text('{"unit_purpose_ko": "x", "chunks": []}', encoding="utf-8")
|
|
out = llm.complete_json("sys", UNIT_PROMPT_SAMPLE)
|
|
assert out == {"unit_purpose_ko": "x", "chunks": []}
|
|
assert llm.usage["calls"] == 1
|
|
|
|
|
|
def test_file_backend_does_not_duplicate_index_rows(tmp_path):
|
|
llm = FileQueueLLM(tmp_path)
|
|
for _ in range(3):
|
|
with pytest.raises(PendingResponse):
|
|
llm.complete_json("sys", UNIT_PROMPT_SAMPLE)
|
|
lines = (tmp_path / "index.jsonl").read_text(encoding="utf-8").strip().splitlines()
|
|
assert len(lines) == 1
|
|
|
|
|
|
def test_create_llm_backends(tmp_path):
|
|
from summarize.llm_client import FakeLLM
|
|
|
|
assert isinstance(create_llm(backend="fake"), FakeLLM)
|
|
assert isinstance(create_llm(fake=True), FakeLLM) # 하위호환
|
|
assert isinstance(create_llm(backend="file", jobs_dir=tmp_path), FileQueueLLM)
|
|
with pytest.raises(ValueError):
|
|
create_llm(backend="없는백엔드")
|
|
|
|
|
|
def test_extract_json_survives_code_fence():
|
|
assert _extract_json('```json\n{"a": 1}\n```') == {"a": 1}
|
|
assert _extract_json('설명입니다\n{"a": 2}\n끝') == {"a": 2}
|
|
|
|
|
|
def test_api_backend_retries_on_429(monkeypatch):
|
|
from config.settings import settings
|
|
from summarize import llm_client
|
|
|
|
monkeypatch.setattr(settings, "llm_base_url", "http://x")
|
|
monkeypatch.setattr(settings, "llm_api_key", "k")
|
|
monkeypatch.setattr(settings, "llm_backoff_base", 0.0)
|
|
monkeypatch.setattr(settings, "llm_backoff_max", 0.0)
|
|
monkeypatch.setattr(llm_client.time, "sleep", lambda *_: None)
|
|
|
|
client = llm_client.OpenAICompatClient()
|
|
calls = {"n": 0}
|
|
|
|
def flaky(system, user):
|
|
calls["n"] += 1
|
|
if calls["n"] < 3:
|
|
raise urllib.error.HTTPError("u", 429, "Too Many Requests", {}, None)
|
|
return {"choices": [{"message": {"content": '{"ok": true}'}}], "usage": {"prompt_tokens": 5}}
|
|
|
|
monkeypatch.setattr(client, "_post_once", flaky)
|
|
assert client.complete_json("s", "u") == {"ok": True}
|
|
assert calls["n"] == 3
|
|
assert client.usage["retries"] == 2
|
|
assert client.usage["calls"] == 1
|
|
|
|
|
|
def test_api_backend_does_not_retry_on_400(monkeypatch):
|
|
from config.settings import settings
|
|
from summarize import llm_client
|
|
|
|
monkeypatch.setattr(settings, "llm_base_url", "http://x")
|
|
monkeypatch.setattr(settings, "llm_api_key", "k")
|
|
client = llm_client.OpenAICompatClient()
|
|
calls = {"n": 0}
|
|
|
|
def bad(system, user):
|
|
calls["n"] += 1
|
|
raise urllib.error.HTTPError("u", 400, "Bad Request", {}, None)
|
|
|
|
monkeypatch.setattr(client, "_post_once", bad)
|
|
with pytest.raises(urllib.error.HTTPError):
|
|
client.complete_json("s", "u")
|
|
assert calls["n"] == 1, "4xx(429 제외)는 재시도하지 않아야 한다"
|
|
|
|
|
|
# ------------------------------------------------------------------ jobs CLI
|
|
|
|
def test_jobs_answer_validates_schema(tmp_path, capsys):
|
|
llm = FileQueueLLM(tmp_path)
|
|
with pytest.raises(PendingResponse) as ei:
|
|
llm.complete_json("sys", UNIT_PROMPT_SAMPLE)
|
|
jid = ei.value.job_id
|
|
|
|
bad = tmp_path / "bad.json"
|
|
bad.write_text('{"unit_purpose_ko": 1, "chunks": [{"line_start": "x"}]}', encoding="utf-8")
|
|
args = type("A", (), {"dir": str(tmp_path), "job_id": jid, "file": str(bad),
|
|
"stdin": False, "no_validate": False})()
|
|
assert jobs.cmd_answer(args) == 1
|
|
assert not llm.response_path(jid).exists()
|
|
|
|
good = tmp_path / "good.json"
|
|
good.write_text('{"unit_purpose_ko": "정상", "chunks": []}', encoding="utf-8")
|
|
args.file = str(good)
|
|
assert jobs.cmd_answer(args) == 0
|
|
assert json.loads(llm.response_path(jid).read_text(encoding="utf-8"))["unit_purpose_ko"] == "정상"
|
|
|
|
|
|
def test_jobs_answer_reports_missing_file(tmp_path):
|
|
args = type("A", (), {"dir": str(tmp_path), "job_id": "deadbeef", "file": "nope.json",
|
|
"stdin": False, "no_validate": False})()
|
|
assert jobs.cmd_answer(args) == 1 # 프롬프트도 없으므로 1
|
|
|
|
|
|
# ------------------------------------------------- /ingest 의 백엔드 선택
|
|
|
|
INGEST_PAYLOAD = {
|
|
"MAIN_PROGRAM": "ZBK_T1",
|
|
"INCLUDE_PROGRAM": [{"INCLUDE": "ZBK_T1", "SOURCE_CODE":
|
|
"REPORT zbk_t1.\nSTART-OF-SELECTION.\n PERFORM go.\n"
|
|
"FORM go.\n SELECT * FROM t001 INTO TABLE gt.\nENDFORM."}],
|
|
}
|
|
|
|
|
|
@pytest.fixture()
|
|
def api_client(tmp_path, monkeypatch):
|
|
from fastapi.testclient import TestClient
|
|
|
|
import query.api as api
|
|
from config.settings import settings
|
|
|
|
monkeypatch.setattr(settings, "database_url", f"sqlite:///{tmp_path / 'i.db'}")
|
|
monkeypatch.setattr(settings, "data_normalized", tmp_path / "n")
|
|
monkeypatch.setattr(settings, "data_parsed", tmp_path / "p")
|
|
monkeypatch.setattr(settings, "data_llm_jobs", tmp_path / "jobs")
|
|
# /ingest 의 백그라운드 작업이 요약 후 위키를 쓴다 — 실제 wiki/ 로 새지 않게 격리
|
|
monkeypatch.setattr(settings, "wiki_dir", tmp_path / "wiki")
|
|
return TestClient(api.app), settings, tmp_path
|
|
|
|
|
|
def test_ingest_backend_off_skips_summary(api_client, monkeypatch):
|
|
c, settings, _ = api_client
|
|
monkeypatch.setattr(settings, "summarize_backend", "off")
|
|
monkeypatch.setattr(settings, "llm_base_url", "http://x")
|
|
monkeypatch.setattr(settings, "llm_api_key", "k")
|
|
r = c.post("/ingest", json=INGEST_PAYLOAD).json()
|
|
assert r["summarize"] == "disabled" and r["summarize_backend"] == "off"
|
|
|
|
|
|
def test_ingest_backend_file_queues_without_key(api_client, monkeypatch):
|
|
"""키가 없어도 file 백엔드면 프롬프트가 큐에 쌓인다."""
|
|
c, settings, tmp_path = api_client
|
|
monkeypatch.setattr(settings, "summarize_backend", "file")
|
|
monkeypatch.setattr(settings, "llm_base_url", "") # 키 없음
|
|
monkeypatch.setattr(settings, "llm_api_key", "")
|
|
r = c.post("/ingest", json=INGEST_PAYLOAD).json()
|
|
assert r["summarize"] == "scheduled", r
|
|
q = c.get("/summaries/jobs").json()
|
|
assert q["pending"] >= 1, q
|
|
assert any(n["task"] == "program_summary" for n in q["next"]), q
|
|
|
|
|
|
# ------------------------------------------------------------------ 호출 규격 (LLM_PROVIDER)
|
|
|
|
def _configure(monkeypatch, **kw):
|
|
from config.settings import settings
|
|
base = {"llm_base_url": "http://x", "llm_api_key": "", "llm_provider": "openai",
|
|
"llm_chat_path": "/chat/completions", "llm_body_model": "", "llm_json_mode": True,
|
|
"llm_max_tokens": 0, "llm_model": "m", "fabrix_client_key": "", "fabrix_openapi_token": "",
|
|
"fabrix_user_email": ""}
|
|
base.update(kw)
|
|
for k, v in base.items():
|
|
monkeypatch.setattr(settings, k, v)
|
|
|
|
|
|
def test_openai_provider_headers_and_body(monkeypatch):
|
|
from summarize import llm_client
|
|
_configure(monkeypatch, llm_api_key="k", llm_max_tokens=800)
|
|
c = llm_client.OpenAICompatClient()
|
|
assert c.url == "http://x/chat/completions"
|
|
assert c.headers() == {"Content-Type": "application/json", "Authorization": "Bearer k"}
|
|
b = c.body("s", "u")
|
|
assert b["model"] == "m" and b["response_format"] == {"type": "json_object"} and b["max_tokens"] == 800
|
|
|
|
|
|
def test_fabrix_provider_headers_and_body(monkeypatch):
|
|
from summarize import llm_client
|
|
_configure(monkeypatch, llm_provider="fabrix", llm_base_url="https://h/dxhq/prod/api-llm",
|
|
llm_chat_path="", llm_model="581", llm_body_model="/mnt/models", llm_json_mode=False,
|
|
fabrix_client_key="CK", fabrix_openapi_token="TOK", fabrix_user_email="me@x.com")
|
|
c = llm_client.OpenAICompatClient()
|
|
assert c.url == "https://h/dxhq/prod/api-llm" # 완성 주소면 경로를 안 붙인다
|
|
h = c.headers()
|
|
assert h["x-openapi-token"] == "Bearer TOK" # Bearer 접두를 코드가 붙인다
|
|
assert h["x-generative-ai-client"] == "CK"
|
|
assert h["x-llm-model-id"] == "581"
|
|
assert h["x-generative-ai-user-email"] == "me@x.com"
|
|
assert "Authorization" not in h
|
|
b = c.body("s", "u")
|
|
assert b["model"] == "/mnt/models" and "response_format" not in b and "max_tokens" not in b
|
|
|
|
|
|
def test_fabrix_keeps_existing_bearer_prefix(monkeypatch):
|
|
from summarize import llm_client
|
|
_configure(monkeypatch, llm_provider="fabrix", fabrix_client_key="CK", fabrix_openapi_token="Bearer TOK")
|
|
assert llm_client.OpenAICompatClient().headers()["x-openapi-token"] == "Bearer TOK"
|
|
|
|
|
|
def test_provider_requires_its_credentials(monkeypatch):
|
|
from config.settings import settings
|
|
from summarize import llm_client
|
|
_configure(monkeypatch, llm_provider="fabrix") # 키 없음
|
|
with pytest.raises(RuntimeError):
|
|
llm_client.OpenAICompatClient()
|
|
assert settings.llm_enabled() is False
|
|
_configure(monkeypatch, llm_provider="fabrix", fabrix_client_key="a", fabrix_openapi_token="b")
|
|
assert settings.llm_enabled() is True
|
|
_configure(monkeypatch, llm_provider="openai") # api key 없음
|
|
assert settings.llm_enabled() is False
|
|
_configure(monkeypatch, llm_provider="bogus", llm_api_key="k")
|
|
with pytest.raises(RuntimeError):
|
|
llm_client.OpenAICompatClient()
|
|
|
|
|
|
def test_merge_system_folds_system_into_user(monkeypatch):
|
|
from config.settings import settings
|
|
from summarize import llm_client
|
|
_configure(monkeypatch, llm_api_key="k")
|
|
monkeypatch.setattr(settings, "llm_merge_system", True)
|
|
msgs = llm_client.OpenAICompatClient().body("SYS", "USR")["messages"]
|
|
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")
|