byeongwook.choiandClaude Fable 5.1
5df8595fe2
기존엔 프로그램의 호출을 전부 받은 뒤 한꺼번에 저장해서, 마지막 호출이 느리면 앞의 것도 잃었다 (고객사 실측: [8/9] 까지 ok 인 상태에서 9번째 대기). 저장은 여전히 메인 스레드에서만 한다. Ctrl+C 는 남은 호출을 취소하고 올라간다. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
230 lines
11 KiB
Python
230 lines
11 KiB
Python
"""summarize.runner — 프로그램 요약 + unit 조각 추출의 상태·실패 사유·실행로그 라이프사이클.
|
|
|
|
- 프로그램 요약이 먼저 생성되고 program.summary_json 에 저장된다
|
|
- 실패 unit → summary_error 에 예외명+메시지, 재실행 시 실패분만 재시도
|
|
- 조각은 logic_chunk / chunk_fts 에 저장되고 unit.chunk_count 에 반영된다
|
|
- --fake 는 FakeLLM 으로 파이프라인 전체(조각 1개/unit)를 통과한다
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from config.settings import settings
|
|
from index.db import connect, loads
|
|
import summarize.runner as runner
|
|
|
|
PAYLOAD = {
|
|
"MAIN_PROGRAM": "ZRUNNER_T1",
|
|
"DESCRIPTION": "러너 테스트",
|
|
"TEXT_SYMBOL": [{"SYMBOL": "001", "TEXT": "회사코드 조회"}],
|
|
"INCLUDE_PROGRAM": [
|
|
{
|
|
"INCLUDE": "ZRUNNER_T1",
|
|
"SOURCE_CODE": (
|
|
"REPORT zrunner_t1.\n"
|
|
"START-OF-SELECTION.\n"
|
|
" PERFORM ok_one.\n"
|
|
" PERFORM fail_me.\n"
|
|
"FORM ok_one.\n"
|
|
" SELECT * FROM t001 INTO TABLE gt_t001.\n"
|
|
" SORT gt_t001 BY bukrs.\n"
|
|
"ENDFORM.\n"
|
|
"FORM fail_me.\n"
|
|
" WRITE 2.\n"
|
|
"ENDFORM."
|
|
),
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
class StubLLM:
|
|
"""FAIL_ME unit 에서만 예외. ok_one 에는 SELECT 조각 하나를 (줄 번호를 틀리게) 돌려준다."""
|
|
|
|
def __init__(self, fail_names: set[str]):
|
|
self.fail_names = fail_names
|
|
self.usage = {"calls": 0, "prompt_tokens": 0, "completion_tokens": 0, "cost_usd": 0.0}
|
|
self.prompts: list[str] = []
|
|
|
|
def complete_json(self, system: str, user: str) -> dict:
|
|
self.prompts.append(user)
|
|
for name in self.fail_names:
|
|
if f"name: {name}" in user:
|
|
raise RuntimeError("HTTP Error 429: rate limited")
|
|
self.usage["calls"] += 1
|
|
self.usage["prompt_tokens"] += 10
|
|
self.usage["completion_tokens"] += 20
|
|
if "[작업] program_summary" in user:
|
|
return {"program": "x", "business_purpose_ko": "회사코드 마스터 조회 테스트",
|
|
"main_flow": ["회사코드 조회", "출력"], "business_tags": ["마스터관리"], "confidence": 0.7}
|
|
if "name: OK_ONE" in user:
|
|
return {"unit_purpose_ko": "회사코드 마스터를 읽는다",
|
|
"chunks": [{"line_start": 7, "line_end": 8, # 실제는 6~7 — 앵커로 보정되어야 함
|
|
"first_line": "SELECT * FROM t001 INTO TABLE gt_t001.",
|
|
"kind": "sql_select", "purpose_ko": "회사코드 마스터(T001) 전체 조회",
|
|
"purpose_en": "Read company code master", "keywords_ko": ["회사코드"],
|
|
"keywords_en": ["company code"], "sap_objects": ["T001"], "confidence": 0.9}]}
|
|
return {"unit_purpose_ko": "요약", "chunks": []}
|
|
|
|
|
|
@pytest.fixture()
|
|
def ingested(tmp_path, monkeypatch):
|
|
import query.api as api
|
|
|
|
monkeypatch.setattr(settings, "database_url", f"sqlite:///{tmp_path / 'index.db'}")
|
|
monkeypatch.setattr(settings, "data_normalized", tmp_path / "normalized")
|
|
monkeypatch.setattr(settings, "data_parsed", tmp_path / "parsed")
|
|
monkeypatch.setattr(settings, "llm_base_url", "") # ingest 의 자동 요약은 끔
|
|
c = TestClient(api.app)
|
|
assert c.post("/ingest", json=PAYLOAD).json()["status"] == "loaded"
|
|
return c
|
|
|
|
|
|
def _unit(con, name: str):
|
|
return con.execute(
|
|
"SELECT summary_status, summary_error, chunk_count, summary_json FROM unit "
|
|
"WHERE program='ZRUNNER_T1' AND name=?", (name,),
|
|
).fetchone()
|
|
|
|
|
|
def test_program_summary_chunks_failure_and_retry(ingested, monkeypatch):
|
|
stub = StubLLM({"FAIL_ME"})
|
|
monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: stub)
|
|
stats = runner.summarize_units("ZRUNNER_T1", None, fake=False, dry_run=False, trigger="test")
|
|
assert stats["program_summaries"] == 1
|
|
assert stats["failed"] == 1
|
|
assert stats["done"] == 2 # OK_ONE + START-OF-SELECTION
|
|
assert stats["chunks"] == 1
|
|
|
|
# 프롬프트에 프로그램 요약 문맥과 텍스트 심볼·파서 힌트가 들어갔는가
|
|
unit_prompt = next(p for p in stub.prompts if "name: OK_ONE" in p)
|
|
assert "회사코드 마스터 조회 테스트" in unit_prompt # 프로그램 요약 문맥
|
|
assert "L6 SELECT" in unit_prompt # 파서 힌트
|
|
assert " 6| " in unit_prompt # 줄번호 붙은 코드
|
|
|
|
con = connect()
|
|
try:
|
|
p = con.execute("SELECT summary_status, summary_json FROM program WHERE name='ZRUNNER_T1'").fetchone()
|
|
assert p["summary_status"] == "done"
|
|
assert loads(p["summary_json"])["prompt_version"] == runner.PROMPT_VERSION
|
|
|
|
fail = _unit(con, "FAIL_ME")
|
|
assert fail["summary_status"] == "failed"
|
|
assert "RuntimeError: HTTP Error 429" in fail["summary_error"]
|
|
|
|
ok = _unit(con, "OK_ONE")
|
|
assert ok["summary_status"] == "done" and ok["summary_error"] is None
|
|
assert ok["chunk_count"] == 1
|
|
thin = loads(ok["summary_json"])
|
|
assert thin["purpose_ko"] == "회사코드 마스터를 읽는다" and thin["chunk_count"] == 1
|
|
|
|
ch = con.execute("SELECT * FROM logic_chunk WHERE program='ZRUNNER_T1'").fetchone()
|
|
assert ch["chunk_id"].endswith("#FORM#OK_ONE#C1")
|
|
assert (ch["line_start"], ch["line_end"]) == (6, 7) # 앵커로 보정됨
|
|
assert loads(ch["tables_read"]) == ["T001"] # 파서가 채움
|
|
assert ch["kind"] == "sql_select"
|
|
# 검색 색인에 들어갔는가
|
|
hit = con.execute("SELECT chunk_id FROM chunk_fts WHERE chunk_fts MATCH '\"회사코드\"'").fetchone()
|
|
assert hit and hit["chunk_id"] == ch["chunk_id"]
|
|
# 얇은 unit 색인도 한 줄 요약으로 갱신
|
|
u = con.execute("SELECT purpose FROM unit_fts WHERE unit_id=?", (ch["unit_id"],)).fetchone()
|
|
assert u["purpose"] == "회사코드 마스터를 읽는다"
|
|
|
|
log = con.execute("SELECT * FROM llm_usage_log ORDER BY id DESC").fetchall()
|
|
assert len(log) == 1
|
|
assert log[0]["status"] == "done"
|
|
assert log[0]["total"] == stats["done"] + stats["failed"]
|
|
assert log[0]["failed"] == 1 and log[0]["chunks"] == 1
|
|
assert log[0]["trigger_by"] == "test"
|
|
finally:
|
|
con.close()
|
|
|
|
# 재실행(전부 성공) → 실패분만 재시도, 프로그램 요약은 스킵
|
|
stub2 = StubLLM(set())
|
|
monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: stub2)
|
|
stats2 = runner.summarize_units("ZRUNNER_T1", None, fake=False, dry_run=False, trigger="test")
|
|
assert stats2["failed"] == 0 and stats2["done"] == 1 and stats2["program_summaries"] == 0
|
|
assert stats2["skipped"] == stats["done"]
|
|
assert not any("[작업] program_summary" in p for p in stub2.prompts)
|
|
|
|
con = connect()
|
|
try:
|
|
fail = _unit(con, "FAIL_ME")
|
|
assert fail["summary_status"] == "done" and fail["summary_error"] is None and fail["chunk_count"] == 0
|
|
assert con.execute("SELECT COUNT(*) FROM llm_usage_log").fetchone()[0] == 2
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def test_all_skipped_writes_no_log(ingested, monkeypatch):
|
|
monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: StubLLM(set()))
|
|
runner.summarize_units("ZRUNNER_T1", None, fake=False, dry_run=False, trigger="test")
|
|
stats = runner.summarize_units("ZRUNNER_T1", None, fake=False, dry_run=False, trigger="test")
|
|
assert stats["done"] == 0 and stats["failed"] == 0 and stats["program_summaries"] == 0
|
|
con = connect()
|
|
try: # 두 번째 실행은 전부 스킵 — 로그 행이 추가되지 않는다
|
|
assert con.execute("SELECT COUNT(*) FROM llm_usage_log").fetchone()[0] == 1
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def test_fake_llm_pipeline(ingested):
|
|
"""--fake: FakeLLM 이 unit 마다 첫 실행문을 조각으로 만들어 파이프라인 전체를 통과한다."""
|
|
stats = runner.summarize_units("ZRUNNER_T1", None, fake=True, dry_run=False, trigger="test")
|
|
assert stats["failed"] == 0 and stats["program_summaries"] == 1
|
|
assert stats["chunks"] >= 1
|
|
con = connect()
|
|
try:
|
|
n = con.execute("SELECT COUNT(*) FROM logic_chunk WHERE program='ZRUNNER_T1'").fetchone()[0]
|
|
assert n == stats["chunks"]
|
|
for r in con.execute("SELECT c.*, u.line_start us, u.line_end ue FROM logic_chunk c "
|
|
"JOIN unit u ON u.unit_id=c.unit_id"):
|
|
assert r["us"] <= r["line_start"] <= r["line_end"] <= r["ue"]
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def test_reload_keeps_chunks_for_unchanged_units(ingested, monkeypatch):
|
|
"""소스가 바뀐 unit 의 조각만 버리고, 같은 unit 의 조각은 재적재 후에도 남는다."""
|
|
monkeypatch.setattr(runner, "create_llm", lambda fake=False, **kw: StubLLM(set()))
|
|
runner.summarize_units("ZRUNNER_T1", None, fake=False, dry_run=False, trigger="test")
|
|
|
|
changed = dict(PAYLOAD)
|
|
changed["INCLUDE_PROGRAM"] = [dict(PAYLOAD["INCLUDE_PROGRAM"][0])]
|
|
changed["INCLUDE_PROGRAM"][0]["SOURCE_CODE"] = PAYLOAD["INCLUDE_PROGRAM"][0]["SOURCE_CODE"].replace(
|
|
"WRITE 2.", "WRITE 3.") # FAIL_ME 만 바뀜
|
|
assert ingested.post("/ingest", json=changed).json()["status"] == "loaded"
|
|
|
|
con = connect()
|
|
try:
|
|
assert con.execute("SELECT COUNT(*) FROM logic_chunk WHERE unit_id LIKE '%#OK_ONE'").fetchone()[0] == 1
|
|
assert _unit(con, "OK_ONE")["summary_status"] == "done"
|
|
assert _unit(con, "FAIL_ME")["summary_status"] == "none" # 재추출 대상
|
|
assert con.execute("SELECT summary_status FROM program WHERE name='ZRUNNER_T1'").fetchone()[0] == "stale"
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
|
|
def test_run_calls_parallel_saves_each_call_on_main_thread_as_it_finishes():
|
|
"""on_done 은 호출이 끝나는 대로, 메인 스레드에서 불린다 (중간 저장의 전제)."""
|
|
import threading
|
|
import time as _t
|
|
from summarize import runner
|
|
|
|
class SlowLLM:
|
|
usage = {}
|
|
def complete_json(self, system, user):
|
|
_t.sleep(0.3 if user == "slow" else 0.01)
|
|
return {"unit_purpose_ko": user, "chunks": []}
|
|
|
|
calls = [{"prompt": "slow"}, {"prompt": "a"}, {"prompt": "b"}]
|
|
seen = []
|
|
main = threading.get_ident()
|
|
runner._run_calls_parallel(SlowLLM(), calls, 3, on_done=lambda c: seen.append((c["prompt"], threading.get_ident())))
|
|
assert [p for p, _ in seen][-1] == "slow" # 느린 호출이 끝나기 전에 빠른 것들이 먼저 저장된다
|
|
assert {p for p, _ in seen} == {"slow", "a", "b"}
|
|
assert all(tid == main for _, tid in seen)
|
|
assert all("raw" in c for c in calls)
|