jsahn/US2KR
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ed48f4c96baa4eab78adc064fdee6021da4ddb7c
US2KR/tests/test_llm.py
T
jsahnandClaude Fable 5 ed48f4c96b Add an LLM fact-check pass over the generated briefing 
After summarisation the report is verified against the collected
payload (numbers, directions, unsupported claims) by a second LLM call
with a Verification JSON schema. Major issues trigger one revision and
a re-check; minor-only verdicts are normalized to ok. Failures never
block delivery. The verdict is stored in the result JSON and shown on
the dashboard (badge + 검증 메모). Disable with US2KR_VERIFY_ENABLED.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-09-01 10:37:39 +09:00

699 lines
28 KiB
Python
Raw Blame History

"""Tests for us2kr.llm.prompts, us2kr.llm.minimax and us2kr.summarize (no network)."""
from __future__ import annotations
import json
import re
from datetime import date, datetime
from typing import Any
from zoneinfo import ZoneInfo
import httpx2
import openai
import pytest
from openai.types.chat import ChatCompletion
from pydantic import ValidationError
from us2kr.config import Settings
from us2kr.llm.base import LLMError
from us2kr.llm.minimax import (
MiniMaxProvider,
coerce_report_data,
extract_json_text,
parse_report,
report_json_schema,
report_response_format,
)
from us2kr.llm.prompts import SYSTEM_PROMPT, build_user_message
from us2kr.models import (
EarningsItem,
EconEvent,
MarketPayload,
NewsItem,
Quote,
Report,
Verification,
VerificationIssue,
)
from us2kr.summarize import build_provider, summarize_payload, verify_and_revise
ET = ZoneInfo("America/New_York")
KST = ZoneInfo("Asia/Seoul")
GOOD_REPORT: dict[str, Any] = {
"headline": "반도체 강세에 나스닥 상승 마감",
"market_overview": "주요 지수가 상승했습니다. 반도체가 강했습니다.",
"key_drivers": ["엔비디아 실적 기대", "금리 하락"],
"rates_fx_commodities": "10년물 금리가 하락했고 달러/원은 소폭 내렸습니다.",
"korea_implications": ["SOX 상승은 국내 반도체에 긍정적입니다.", "EWY가 올랐습니다."],
"watchlist_today": ["삼성전자 흐름"],
"risk_flags": ["VIX 반등 가능성"],
"data_gaps": ["news: TimeoutError"],
}
# --------------------------------------------------------------------------- fixtures
def make_settings(**overrides: Any) -> Settings:
kwargs: dict[str, Any] = {
"MINIMAX_API_KEY": "test-key-not-real",
"MINIMAX_MODEL": "MiniMax-M3",
"MINIMAX_FALLBACK_MODELS": ["MiniMax-M2.7", "MiniMax-M2.5"],
"MINIMAX_MAX_COMPLETION_TOKENS": 4096,
}
kwargs.update(overrides)
return Settings(_env_file=None, **kwargs)
def make_payload() -> MarketPayload:
session = date(2026, 3, 10)
return MarketPayload(
session_date=session,
us_close_time=datetime(2026, 3, 10, 16, 0, tzinfo=ET),
generated_at=datetime(2026, 3, 11, 6, 15, tzinfo=KST),
krx_date=date(2026, 3, 11),
krx_open_today=True,
quotes=[
Quote(symbol="^GSPC", name="S&P 500", group="indices", close=5912.345, change_pct=0.8123),
Quote(
symbol="^SOX", name="필라델피아 반도체(SOX)", group="indices", close=5321.1, change_pct=2.5
),
Quote(symbol="^TNX", name="미 10년물", group="rates", close=4.2534, change_abs=-0.0512),
Quote(symbol="KRW=X", name="달러/원", group="fx", close=1385.25, change_pct=-0.31),
Quote(
symbol="NVDA",
name="엔비디아",
group="semis",
close=912.5,
change_pct=3.2,
after_hours_price=920.0,
after_hours_change_pct=0.82,
),
Quote(symbol="EWY", name="MSCI Korea ETF(EWY)", group="korea_linked", close=None),
],
news=[
NewsItem(
title="Nvidia rallies ahead of earnings",
summary="Shares rose as investors bet on strong data-center demand. " * 5,
link="https://example.com/1",
source="CNBC",
published=datetime(2026, 3, 10, 15, 30, tzinfo=ET),
),
NewsItem(
title="Treasury yields slip",
link="https://example.com/2",
source="MarketWatch",
published=datetime(2026, 3, 10, 14, 0, tzinfo=ET),
),
],
events=[
EconEvent(
name="CPI (YoY)",
time=datetime(2026, 3, 10, 8, 30, tzinfo=ET),
actual="3.1%",
forecast="3.2%",
previous="3.3%",
importance="high",
kind="released",
),
EconEvent(name="FOMC 의사록", time=None, importance="medium", kind="upcoming"),
],
earnings=[
EarningsItem(
symbol="MU",
name="마이크론",
report_date=session,
timing="amc",
eps_actual=1.23,
eps_estimate=1.10,
revenue_actual=8_100_000_000,
revenue_estimate=7_900_000_000,
)
],
history_notes=["S&P 500 3일 연속 상승"],
data_gaps=["events: HTTPError"],
)
def make_completion(
content: str | None,
*,
model: str = "MiniMax-M3",
finish_reason: str = "stop",
cached_tokens: int | None = 120,
reasoning: str | None = "생각중...",
) -> ChatCompletion:
message: dict[str, Any] = {"role": "assistant", "content": content}
if reasoning is not None:
message["reasoning_content"] = reasoning
usage: dict[str, Any] = {"prompt_tokens": 1500, "completion_tokens": 400, "total_tokens": 1900}
if cached_tokens is not None:
usage["prompt_tokens_details"] = {"cached_tokens": cached_tokens}
return ChatCompletion.model_validate(
{
"id": "chatcmpl-test",
"object": "chat.completion",
"created": 1_700_000_000,
"model": model,
"choices": [{"index": 0, "finish_reason": finish_reason, "message": message}],
"usage": usage,
}
)
def make_status_error(status: int, message: str = "server error") -> openai.APIStatusError:
request = httpx2.Request("POST", "https://api.minimax.io/v1/chat/completions")
response = httpx2.Response(status, request=request, json={"error": message})
if status == 429:
return openai.RateLimitError(message, response=response, body={"error": message})
return openai.APIStatusError(message, response=response, body={"error": message})
class FakeCompletions:
def __init__(self, responses: list[Any]) -> None:
self._responses = list(responses)
self.calls: list[dict[str, Any]] = []
def create(self, **kwargs: Any) -> Any:
self.calls.append(kwargs)
if not self._responses:
raise AssertionError("unexpected extra call to chat.completions.create")
item = self._responses.pop(0)
if isinstance(item, Exception):
raise item
return item
class FakeChat:
def __init__(self, completions: FakeCompletions) -> None:
self.completions = completions
class FakeClient:
def __init__(self, responses: list[Any]) -> None:
self.completions = FakeCompletions(responses)
self.chat = FakeChat(self.completions)
@property
def calls(self) -> list[dict[str, Any]]:
return self.completions.calls
@property
def models_called(self) -> list[str]:
return [c["model"] for c in self.calls]
def make_provider(responses: list[Any], **settings_overrides: Any) -> tuple[MiniMaxProvider, FakeClient]:
client = FakeClient(responses)
return MiniMaxProvider(make_settings(**settings_overrides), client=client), client
# ---------------------------------------------------------------------------- prompts
def test_system_prompt_has_no_dates_or_timestamps() -> None:
assert SYSTEM_PROMPT.strip()
assert not re.search(r"\d{4}-\d{2}-\d{2}", SYSTEM_PROMPT)
assert not re.search(r"\d{4}년", SYSTEM_PROMPT)
assert not re.search(r"\d{1,2}:\d{2}", SYSTEM_PROMPT)
assert not re.search(r"20\d{2}", SYSTEM_PROMPT)
for keyword in ("한국 투자자", "투자 조언", "보도에 따르면", "JSON", "반도체", "환율", "EWY", "습니다"):
assert keyword in SYSTEM_PROMPT
def test_system_prompt_contains_group_glossary() -> None:
for title in ("주요 지수", "미국 국채 금리", "한국 연관(EWY·ADR)", "반도체", "빅테크"):
assert title in SYSTEM_PROMPT
def test_system_prompt_declares_field_types_and_korean_only() -> None:
assert "한국어로만" in SYSTEM_PROMPT
for name in Report.model_fields:
assert f"- {name}: " in SYSTEM_PROMPT
assert f'"{name}":' in SYSTEM_PROMPT # skeleton example
assert "- key_drivers: 문자열 배열" in SYSTEM_PROMPT
assert "- headline: 문자열 -" in SYSTEM_PROMPT
def test_build_user_message_is_deterministic_and_korean() -> None:
payload = make_payload()
first = build_user_message(payload)
second = build_user_message(make_payload())
assert first == second
for title in (
"## 세션 정보",
"## 시세",
"### 주요 지수",
"### 미국 국채 금리",
"### 환율·달러",
"### 반도체",
):
assert title in first
for title in ("## 히스토리 노트", "## 뉴스", "## 경제지표", "## 실적", "## 수집 누락"):
assert title in first
def test_build_user_message_content_and_rounding() -> None:
text = build_user_message(make_payload())
# session block: close time converted to KST (16:00 EDT == 05:00 KST next day)
assert "2026-03-10" in text
assert "2026-03-11 05:00 KST" in text
assert "오늘 한국 장: 개장" in text
# prices rounded, percent vs points
assert "^GSPC S&P 500: 5,912.35 (+0.81%)" in text
assert "^TNX 미 10년물: 4.253 (-0.051%p)" in text
assert "애프터 920.00 (+0.82%)" in text
assert "EWY MSCI Korea ETF(EWY): 자료 없음" in text
# news: title | source | KST time; summary whitespace-collapsed and truncated
assert "Nvidia rallies ahead of earnings | CNBC | 2026-03-11 04:30 KST" in text
assert "요약: Shares rose as investors bet on strong data-center demand." in text
assert "…" in text
# events / earnings / gaps / history
assert "[발표] CPI (YoY)" in text and "실제 3.1% / 예상 3.2% / 이전 3.3%" in text
assert "[예정] FOMC 의사록" in text and "시간 미정" in text
assert "MU 마이크론 (2026-03-10 amc): EPS 실제 1.23 / 예상 1.10" in text
assert "- S&P 500 3일 연속 상승" in text
assert "- events: HTTPError" in text
def test_build_user_message_empty_sections() -> None:
payload = make_payload().model_copy(
update={"quotes": [], "news": [], "events": [], "earnings": [], "history_notes": [], "data_gaps": []}
)
payload = payload.model_copy(update={"krx_open_today": False})
text = build_user_message(payload)
assert "(시세 자료 없음)" in text
assert text.count("(없음)") == 5
assert "오늘 한국 장: 휴장" in text
# ----------------------------------------------------------------------------- schema
def _walk(node: Any):
if isinstance(node, dict):
yield node
for v in node.values():
yield from _walk(v)
elif isinstance(node, list):
for v in node:
yield from _walk(v)
def test_report_json_schema_has_no_titles_and_is_closed() -> None:
schema = report_json_schema()
assert all("title" not in d for d in _walk(schema))
assert schema["additionalProperties"] is False
assert schema["type"] == "object"
assert set(schema["properties"]) == set(Report.model_fields)
fmt = report_response_format()
assert fmt["type"] == "json_schema"
assert fmt["json_schema"]["name"] == "report"
assert fmt["json_schema"]["schema"] == schema
json.dumps(schema) # serialisable
# Report itself is unchanged by the helper (deepcopy)
assert "title" in Report.model_json_schema()
# ------------------------------------------------------------------- extraction utils
@pytest.mark.parametrize(
"raw",
[
json.dumps(GOOD_REPORT, ensure_ascii=False),
"<think>먼저 지수를 본다...</think>\n" + json.dumps(GOOD_REPORT, ensure_ascii=False),
"```json\n" + json.dumps(GOOD_REPORT, ensure_ascii=False) + "\n```",
"<think>a</think><think>b</think>```\n" + json.dumps(GOOD_REPORT, ensure_ascii=False) + "```",
# unterminated <think> with an earlier closed one: take text after the last </think>
"<think>x</think>" + json.dumps(GOOD_REPORT, ensure_ascii=False) + "\n<think>trailing",
# unterminated <think> and no </think>: fall back to first JSON object
"<think>reasoning " + json.dumps(GOOD_REPORT, ensure_ascii=False),
"설명입니다: " + json.dumps(GOOD_REPORT, ensure_ascii=False) + " 끝.",
],
)
def test_extract_and_parse_variants(raw: str) -> None:
text = extract_json_text(raw)
assert json.loads(text) == GOOD_REPORT
report = parse_report(raw)
assert report.headline == GOOD_REPORT["headline"]
def test_parse_report_coerces_string_list_fields(caplog: pytest.LogCaptureFixture) -> None:
prose = {
**GOOD_REPORT,
"key_drivers": "상승 요인: 반도체가 이끌었습니다. 하락 요인: 유가가 내렸습니다.",
"korea_implications": "- SOX 상승은 긍정적입니다.\n- 환율은 안정적입니다.\n- EWY가 올랐습니다.",
"watchlist_today": "1. 삼성전자 흐름\n2. 환율 방향",
"risk_flags": "단일 문장입니다",
"data_gaps": "events: FRED key missing",
}
with caplog.at_level("WARNING", logger="us2kr.llm.minimax"):
report = parse_report("```json\n" + json.dumps(prose, ensure_ascii=False) + "\n```")
assert report.key_drivers == ["상승 요인: 반도체가 이끌었습니다.", "하락 요인: 유가가 내렸습니다."]
assert report.korea_implications == [
"SOX 상승은 긍정적입니다.",
"환율은 안정적입니다.",
"EWY가 올랐습니다.",
]
assert report.watchlist_today == ["삼성전자 흐름", "환율 방향"]
assert report.risk_flags == ["단일 문장입니다"]
assert report.data_gaps == ["events: FRED key missing"]
assert report.headline == GOOD_REPORT["headline"]
assert any("coerced" in r.getMessage() for r in caplog.records)
def test_coerce_report_data_leaves_valid_and_non_dict_alone() -> None:
data, coerced = coerce_report_data(dict(GOOD_REPORT))
assert coerced == [] and data == GOOD_REPORT
assert coerce_report_data(["not", "a", "dict"]) == (["not", "a", "dict"], [])
# wrong non-string types are not coerced -> validation error surfaces
with pytest.raises(ValidationError):
parse_report(json.dumps({**GOOD_REPORT, "key_drivers": 42}))
def test_extract_json_text_raises_without_object() -> None:
with pytest.raises(ValueError):
extract_json_text("<think>only thinking, no answer")
with pytest.raises(ValueError):
extract_json_text("no json here")
# ----------------------------------------------------------------------------- provider
def test_provider_requires_api_key_when_no_client() -> None:
with pytest.raises(LLMError):
MiniMaxProvider(make_settings(MINIMAX_API_KEY=""))
def test_provider_builds_openai_client(monkeypatch: pytest.MonkeyPatch) -> None:
captured: dict[str, Any] = {}
class DummyOpenAI:
def __init__(self, **kwargs: Any) -> None:
captured.update(kwargs)
monkeypatch.setattr(openai, "OpenAI", DummyOpenAI)
settings = make_settings(MINIMAX_TIMEOUT_SECONDS=42.0)
provider = MiniMaxProvider(settings)
assert provider.models == ["MiniMax-M3", "MiniMax-M2.7", "MiniMax-M2.5"]
assert captured["api_key"] == "test-key-not-real"
assert captured["base_url"] == settings.minimax_base_url
assert captured["timeout"] == 42.0
assert captured["max_retries"] == 2
def test_summarize_clean_json(caplog: pytest.LogCaptureFixture) -> None:
provider, client = make_provider([make_completion(json.dumps(GOOD_REPORT, ensure_ascii=False))])
with caplog.at_level("INFO", logger="us2kr.llm.minimax"):
report, model = provider.summarize(make_payload())
assert model == "MiniMax-M3"
assert report == Report.model_validate(GOOD_REPORT)
assert len(client.calls) == 1
call = client.calls[0]
assert call["model"] == "MiniMax-M3"
assert call["max_completion_tokens"] == 4096
assert call["response_format"]["type"] == "json_schema"
assert call["extra_body"]["reasoning_split"] is True
assert call["extra_body"]["thinking"] == {"type": "adaptive"}
assert [m["role"] for m in call["messages"]] == ["system", "user"]
assert call["messages"][0]["content"] == SYSTEM_PROMPT
assert "## 시세" in call["messages"][1]["content"]
usage_logs = [r.getMessage() for r in caplog.records if "prompt_tokens=1500" in r.getMessage()]
assert usage_logs and "cached_tokens=120" in usage_logs[0] and "completion_tokens=400" in usage_logs[0]
# never log the key or the prompt body
assert all("test-key-not-real" not in r.getMessage() for r in caplog.records)
assert all("## 시세" not in r.getMessage() for r in caplog.records)
def test_summarize_think_wrapped_and_code_fence() -> None:
body = json.dumps(GOOD_REPORT, ensure_ascii=False)
provider, _ = make_provider(
[make_completion(f"<think>추론...</think>\n```json\n{body}\n```", reasoning=None)]
)
report, model = provider.summarize(make_payload())
assert model == "MiniMax-M3"
assert report.headline == GOOD_REPORT["headline"]
def test_summarize_no_thinking_param_for_m2_models() -> None:
provider, client = make_provider(
[make_completion(json.dumps(GOOD_REPORT), model="MiniMax-M2.7")], MINIMAX_MODEL="MiniMax-M2.7"
)
provider.summarize(make_payload())
assert "thinking" not in client.calls[0]["extra_body"]
assert client.calls[0]["extra_body"]["reasoning_split"] is True
def test_invalid_json_then_repaired_on_retry() -> None:
broken = '{"headline": "x", "market_overview": 12' # truncated / invalid
provider, client = make_provider(
[make_completion(broken), make_completion(json.dumps(GOOD_REPORT, ensure_ascii=False))]
)
report, model = provider.summarize(make_payload())
assert model == "MiniMax-M3"
assert report.headline == GOOD_REPORT["headline"]
assert client.models_called == ["MiniMax-M3", "MiniMax-M3"]
repair_messages = client.calls[1]["messages"]
assert [m["role"] for m in repair_messages] == ["system", "user", "assistant", "user"]
assert repair_messages[2]["content"] == broken
assert "검증에 실패" in repair_messages[3]["content"]
def test_schema_violation_then_repaired() -> None:
missing_fields = json.dumps({"headline": "only headline"})
provider, client = make_provider(
[make_completion(missing_fields), make_completion(json.dumps(GOOD_REPORT, ensure_ascii=False))]
)
report, _ = provider.summarize(make_payload())
assert report.market_overview == GOOD_REPORT["market_overview"]
assert "market_overview" in client.calls[1]["messages"][3]["content"]
def test_repair_failure_moves_to_next_model() -> None:
provider, client = make_provider(
[
make_completion("not json"),
make_completion("still not json"),
make_completion(json.dumps(GOOD_REPORT, ensure_ascii=False), model="MiniMax-M2.7"),
]
)
_, model = provider.summarize(make_payload())
assert model == "MiniMax-M2.7"
assert client.models_called == ["MiniMax-M3", "MiniMax-M3", "MiniMax-M2.7"]
def test_api_status_error_falls_back_to_second_model() -> None:
provider, client = make_provider(
[
make_status_error(500),
make_completion(json.dumps(GOOD_REPORT, ensure_ascii=False), model="MiniMax-M2.7"),
]
)
report, model = provider.summarize(make_payload())
assert model == "MiniMax-M2.7"
assert client.models_called == ["MiniMax-M3", "MiniMax-M2.7"]
assert report.headline == GOOD_REPORT["headline"]
def test_rate_limit_and_connection_errors_fall_through() -> None:
request = httpx2.Request("POST", "https://api.minimax.io/v1/chat/completions")
provider, client = make_provider(
[
make_status_error(429, "rate limited"),
openai.APITimeoutError(request=request),
make_completion(json.dumps(GOOD_REPORT, ensure_ascii=False), model="MiniMax-M2.5"),
]
)
_, model = provider.summarize(make_payload())
assert model == "MiniMax-M2.5"
assert client.models_called == ["MiniMax-M3", "MiniMax-M2.7", "MiniMax-M2.5"]
def test_finish_reason_length_moves_to_next_model() -> None:
provider, client = make_provider(
[
make_completion('{"headline": "cut off', finish_reason="length"),
make_completion(json.dumps(GOOD_REPORT, ensure_ascii=False), model="MiniMax-M2.7"),
]
)
_, model = provider.summarize(make_payload())
assert model == "MiniMax-M2.7"
assert client.models_called == ["MiniMax-M3", "MiniMax-M2.7"]
def test_all_models_fail_raises_llm_error_with_attempts() -> None:
request = httpx2.Request("POST", "https://api.minimax.io/v1/chat/completions")
provider, client = make_provider(
[
make_status_error(503, "unavailable"),
make_completion(None), # empty content
openai.APIConnectionError(request=request),
]
)
with pytest.raises(LLMError) as excinfo:
provider.summarize(make_payload())
msg = str(excinfo.value)
assert "MiniMax-M3" in msg and "503" in msg
assert "MiniMax-M2.7" in msg and "empty content" in msg
assert "MiniMax-M2.5" in msg and "APIConnectionError" in msg
assert client.models_called == ["MiniMax-M3", "MiniMax-M2.7", "MiniMax-M2.5"]
def test_unexpected_exception_propagates() -> None:
provider, _ = make_provider([RuntimeError("programming error")])
with pytest.raises(RuntimeError):
provider.summarize(make_payload())
# ---------------------------------------------------------------------------- summarize
def test_build_provider_returns_minimax(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(openai, "OpenAI", lambda **kwargs: object())
provider = build_provider(make_settings())
assert isinstance(provider, MiniMaxProvider)
def test_summarize_payload_merges_data_gaps() -> None:
class StubProvider:
def summarize(self, payload: MarketPayload) -> tuple[Report, str]:
report = Report.model_validate(
{**GOOD_REPORT, "data_gaps": ["news: TimeoutError", "events: HTTPError"]}
)
return report, "stub-model"
payload = make_payload().model_copy(update={"data_gaps": ["prices: ^VIX missing", "events: HTTPError"]})
report, model = summarize_payload(payload, StubProvider())
assert model == "stub-model"
assert report.data_gaps == ["prices: ^VIX missing", "events: HTTPError", "news: TimeoutError"]
def test_summarize_payload_no_gaps_keeps_report() -> None:
class StubProvider:
def summarize(self, payload: MarketPayload) -> tuple[Report, str]:
return Report.model_validate({**GOOD_REPORT, "data_gaps": []}), "m"
payload = make_payload().model_copy(update={"data_gaps": []})
report, _ = summarize_payload(payload, StubProvider())
assert report.data_gaps == []
# ---------------------------------------------------------------------- verification
GOOD_VERIFICATION: dict[str, Any] = {
"ok": False,
"issues": [
{
"claim": "S&P 500이 1.5% 하락했습니다.",
"problem": "제공된 데이터는 +0.81% 상승입니다.",
"severity": "major",
}
],
}
def test_verify_sends_payload_and_draft_and_parses_verdict() -> None:
provider, client = make_provider([make_completion(json.dumps({"ok": True, "issues": []}))])
verification, model = provider.verify(make_payload(), Report.model_validate(GOOD_REPORT))
assert verification.ok is True and verification.issues == [] and model == "MiniMax-M3"
call = client.calls[0]
assert call["response_format"]["json_schema"]["name"] == "verification"
assert "팩트체커" in call["messages"][0]["content"]
user = call["messages"][1]["content"]
assert "## 검증 대상 브리핑 초안 (JSON)" in user
assert "반도체 강세에 나스닥 상승 마감" in user # the draft under review
assert "S&P 500" in user # the source data to check against
def test_verify_verdict_has_major_property() -> None:
provider, _ = make_provider([make_completion(json.dumps(GOOD_VERIFICATION, ensure_ascii=False))])
verification, _ = provider.verify(make_payload(), Report.model_validate(GOOD_REPORT))
assert verification.ok is False and verification.has_major is True
def test_verify_exhausts_models_with_task_tag() -> None:
provider, client = make_provider([make_status_error(500)] * 3)
with pytest.raises(LLMError, match=r"\(verify\)"):
provider.verify(make_payload(), Report.model_validate(GOOD_REPORT))
assert client.models_called == ["MiniMax-M3", "MiniMax-M2.7", "MiniMax-M2.5"]
def test_revise_replays_draft_and_issues() -> None:
revised = {**GOOD_REPORT, "headline": "수정된 헤드라인"}
provider, client = make_provider([make_completion(json.dumps(revised, ensure_ascii=False))])
issues = [
VerificationIssue(claim="S&P 500이 1.5% 하락", problem="데이터는 +0.81%", severity="major")
]
report, model = provider.revise(make_payload(), Report.model_validate(GOOD_REPORT), issues)
assert report.headline == "수정된 헤드라인" and model == "MiniMax-M3"
call = client.calls[0]
assert call["response_format"]["json_schema"]["name"] == "report"
messages = call["messages"]
assert messages[0]["content"] == SYSTEM_PROMPT # same author persona as summarize
assert messages[2]["role"] == "assistant" # the draft being corrected
assert "검증 이슈" in messages[3]["content"] and "[major]" in messages[3]["content"]
# ------------------------------------------------------------------ verify_and_revise
class GlueProvider:
"""LLMProvider stub with scripted verify verdicts and one revise result."""
def __init__(
self,
verdicts: list[Verification],
revised: Report | None = None,
revise_fails: bool = False,
) -> None:
self._verdicts = list(verdicts)
self._revised = revised
self._revise_fails = revise_fails
self.verify_calls = 0
self.revise_calls = 0
def summarize(self, payload: MarketPayload) -> tuple[Report, str]:
raise AssertionError("summarize must not be called")
def verify(self, payload: MarketPayload, report: Report) -> tuple[Verification, str]:
self.verify_calls += 1
return self._verdicts.pop(0), "stub"
def revise(self, payload, report, issues) -> tuple[Report, str]:
self.revise_calls += 1
if self._revise_fails:
raise LLMError("revise down")
assert self._revised is not None
return self._revised, "stub"
def _issue(severity: str) -> VerificationIssue:
return VerificationIssue(claim="c", problem="p", severity=severity) # type: ignore[arg-type]
def test_verify_and_revise_clean_verdict_short_circuits() -> None:
provider = GlueProvider([Verification(ok=True)])
report = Report.model_validate(GOOD_REPORT)
out, verification = verify_and_revise(make_payload(), report, provider)
assert out is report and verification.ok is True
assert provider.verify_calls == 1 and provider.revise_calls == 0
def test_verify_and_revise_minor_issues_do_not_revise_and_normalize_ok() -> None:
verdict = Verification(ok=False, issues=[_issue("minor")]) # verifiers over-report minor as not-ok
provider = GlueProvider([verdict])
report = Report.model_validate(GOOD_REPORT)
out, verification = verify_and_revise(make_payload(), report, provider)
assert out is report
assert verification.ok is True and len(verification.issues) == 1 # ok normalized to "no major"
assert provider.revise_calls == 0
def test_verify_and_revise_major_issue_revises_and_rechecks() -> None:
revised = Report.model_validate({**GOOD_REPORT, "headline": "고친 헤드라인"})
provider = GlueProvider(
[Verification(ok=False, issues=[_issue("major")]), Verification(ok=True)], revised=revised
)
out, verification = verify_and_revise(make_payload(), Report.model_validate(GOOD_REPORT), provider)
assert out is revised and verification.ok is True
assert provider.verify_calls == 2 and provider.revise_calls == 1
def test_verify_and_revise_revision_failure_keeps_original() -> None:
verdict = Verification(ok=False, issues=[_issue("major")])
provider = GlueProvider([verdict], revise_fails=True)
report = Report.model_validate(GOOD_REPORT)
out, verification = verify_and_revise(make_payload(), report, provider)
assert out is report and verification is verdict
assert provider.verify_calls == 1 and provider.revise_calls == 1
Reference in New Issue View Git Blame Copy Permalink