mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
ThearchyHelios
574 lines
26 KiB
Python
574 lines
26 KiB
Python
"""跑完之后的复核层。
|
||
|
||
核心不变量有两条，其余都是围着它们转的：
|
||
|
||
1. 干净的运行**一次模型都不调**。复核是补救手段，不是每次问数的固定开销。
|
||
2. 复核模型看不到原始数据，所以它写出来的任何新数字都是现编的，必须机械拦住。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
|
||
from app.core.events import EventType
|
||
from app.engine import review as rv
|
||
from app.main import app
|
||
|
||
|
||
@pytest.fixture
|
||
async def client():
|
||
transport = ASGITransport(app=app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
||
yield c
|
||
|
||
|
||
def ev(etype, **data):
|
||
return {"type": str(etype), "node_id": data.pop("node_id", ""), "data": data}
|
||
|
||
|
||
def warn(message, code=""):
|
||
return ev(EventType.LOG, level="warn", message=message, code=code)
|
||
|
||
|
||
OK_OUTPUT = {"结果": "管理员有 6 个。"}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# scan：规则层
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def test_tool_error_is_broken():
|
||
signals = rv.scan([ev(EventType.TOOL_ERROR, tool="db_query__x", error="no such table")],
|
||
OK_OUTPUT)
|
||
assert [s.kind for s in signals] == ["tool_error"]
|
||
assert signals[0].severity == rv.BROKEN
|
||
assert "no such table" in signals[0].detail
|
||
|
||
|
||
def test_tool_error_then_success_is_only_degraded():
|
||
"""自愈接住了就不该报"答案不可信"。
|
||
|
||
参数自动纠正那条路每修一次就发一条 tool.error，不区分的话每次自愈
|
||
都会触发一次复核，还告诉用户答案不可信——而它明明跑通了。
|
||
"""
|
||
signals = rv.scan([
|
||
ev(EventType.TOOL_ERROR, tool="db_query__x", error="unexpected keyword 'query'"),
|
||
ev(EventType.TOOL_END, tool="db_query__x"),
|
||
], OK_OUTPUT)
|
||
assert [s.kind for s in signals] == ["tool_args_fixed"]
|
||
assert signals[0].severity == rv.DEGRADED
|
||
|
||
|
||
def test_retrieve_degraded_and_empty():
|
||
degraded = rv.scan([ev(EventType.RETRIEVE_END, count=3, degraded=True)], OK_OUTPUT)
|
||
assert degraded[0].kind == "retrieve_degraded"
|
||
assert degraded[0].severity == rv.DEGRADED
|
||
|
||
empty = rv.scan([ev(EventType.RETRIEVE_END, count=0, query="商家账户")], OK_OUTPUT)
|
||
assert empty[0].kind == "retrieve_empty"
|
||
assert empty[0].severity == rv.BROKEN
|
||
assert "商家账户" in empty[0].detail
|
||
|
||
|
||
def test_specific_retrieve_reason_wins_over_the_generic_one():
|
||
"""检索降级会发两条：knowledge.py 先发具体原因的 warn，再发 retrieve.end。
|
||
|
||
两边都收就是同一件事报两遍，用户读到一条含糊的加一条具体的，还得自己
|
||
判断是不是两个问题。
|
||
"""
|
||
signals = rv.scan([
|
||
warn("default 里有 32 条向量由「local-hashing」建立，和当前的对不上", "retrieve_degraded"),
|
||
ev(EventType.RETRIEVE_END, count=5, degraded=True),
|
||
], OK_OUTPUT)
|
||
assert len(signals) == 1
|
||
assert "32 条向量" in signals[0].detail
|
||
|
||
|
||
def test_node_and_run_failures_are_broken():
|
||
signals = rv.scan([
|
||
ev(EventType.NODE_FAILED, node_id="agent1", error="boom"),
|
||
ev(EventType.RUN_FAILED, error="超时"),
|
||
], OK_OUTPUT)
|
||
assert {s.kind for s in signals} == {"node_failed", "run_failed"}
|
||
assert all(s.severity == rv.BROKEN for s in signals)
|
||
|
||
|
||
def test_a_rejected_node_is_a_human_decision_not_a_fault():
|
||
"""被驳回的那一步节点层照旧发 node.failed，数据里带 reason=rejected（总方案 X9、05 (a)）：复核不能把它说成「执行失败」，
|
||
更不能当成「重跑一次大概率能好」——重跑只会把同一件事再问一遍人。没有 reason 的老运行照旧按失败算。"""
|
||
signals = rv.scan([
|
||
ev(EventType.NODE_STARTED, node_id="gate", label="把关"),
|
||
ev(EventType.NODE_FAILED, node_id="gate", error="人工驳回：不对", reason="rejected", kind="approve",
|
||
decision="reject"),
|
||
], OK_OUTPUT)
|
||
[signal] = signals
|
||
assert signal.kind == "node_rejected" and signal.severity == rv.DEGRADED
|
||
assert signal.detail == "节点「把关」被驳回：人工驳回：不对" and "失败" not in signal.detail
|
||
assert not rv.should_retry(signals)
|
||
legacy = rv.scan([ev(EventType.NODE_FAILED, node_id="gate", error="人工驳回：不对")], OK_OUTPUT)
|
||
assert [s.kind for s in legacy] == ["node_failed"] and rv.should_retry(legacy)
|
||
|
||
|
||
def test_warn_classified_by_code_not_by_message():
|
||
"""分类挂在 code 上。消息是给人读的，随时会改。"""
|
||
assert rv.scan([warn("agent 用满了 8 步还没给出结论。", "step_limit")], OK_OUTPUT)[0] \
|
||
.severity == rv.BROKEN
|
||
assert rv.scan([warn("第 1 次校验失败：不是合法 JSON", "validate_retry")], OK_OUTPUT)[0] \
|
||
.severity == rv.DEGRADED
|
||
|
||
|
||
def test_benign_warns_do_not_trigger_review():
|
||
"""流式回退、隔离降级对答案内容零影响。
|
||
|
||
为它们叫一次模型，等于每次网络抖一下就多花一次调用，还要用户读一句
|
||
与结论无关的告警。轨迹里有记录就够了。
|
||
"""
|
||
events = [
|
||
warn("流式输出失败，已改用非流式调用：connection reset", "stream_fallback"),
|
||
warn("节点要求 strict 隔离，但 microVM 未就绪，实际使用了 seatbelt", "isolation_fallback"),
|
||
]
|
||
assert rv.scan(events, OK_OUTPUT) == []
|
||
|
||
|
||
def test_warn_without_code_defaults_to_degraded():
|
||
"""库里的老事件没有 code。宁可多说一句，也好过把真问题当噪音吞掉。"""
|
||
signals = rv.scan([warn("某个没带 code 的告警")], OK_OUTPUT)
|
||
assert [s.kind for s in signals] == ["warn"]
|
||
assert signals[0].severity == rv.DEGRADED
|
||
|
||
|
||
def test_empty_and_placeholder_output_are_broken():
|
||
assert rv.scan([], None)[0].kind == "empty_output"
|
||
assert rv.scan([], {"结果": ""})[0].kind == "empty_output"
|
||
|
||
# run dd9927e6 真的把这句内部管道文案当答案交出去过
|
||
leaked = {"结果": "本轮只执行了第一个工具。看到它的结果之后，再决定这一个还要不要调。"}
|
||
assert rv.scan([], leaked)[0].kind == "placeholder_output"
|
||
|
||
|
||
def test_duplicate_signals_collapse():
|
||
"""一次运行里同一条告警发五遍，用户只需要看到一次。"""
|
||
events = [warn("第 1 次校验失败：不是合法 JSON", "validate_retry")] * 5
|
||
assert len(rv.scan(events, OK_OUTPUT)) == 1
|
||
|
||
|
||
def test_scan_accepts_orm_like_rows():
|
||
"""事件既可能是 dict 也可能是 ORM 行，两种都要吃得下。"""
|
||
|
||
class Row:
|
||
type = str(EventType.TOOL_ERROR)
|
||
node_id = "n1"
|
||
data = {"tool": "t", "error": "e"}
|
||
|
||
assert rv.scan([Row()], OK_OUTPUT)[0].kind == "tool_error"
|
||
|
||
|
||
def test_worst_and_should_retry():
|
||
broken = rv.Signal("step_limit", "步数用满", rv.BROKEN)
|
||
degraded = rv.Signal("retrieve_degraded", "检索降级", rv.DEGRADED)
|
||
assert rv.worst([degraded, broken]) == rv.BROKEN
|
||
assert rv.worst([degraded]) == rv.DEGRADED
|
||
assert rv.worst([]) == ""
|
||
|
||
assert rv.should_retry([broken]) is True
|
||
# 检索什么都没查到，重跑一遍还是查不到——那不是重跑能解决的
|
||
assert rv.should_retry([rv.Signal("retrieve_empty", "没命中", rv.BROKEN)]) is False
|
||
assert rv.should_retry([degraded]) is False
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# check_numbers：不得新增数字
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def test_check_numbers_allows_reordering():
|
||
original = "管理员有 6 个，商家有 5 个，占比 45.5%。"
|
||
revised = "占比 45.5%。其中商家 5 个、管理员 6 个。"
|
||
assert rv.check_numbers(original, revised) == []
|
||
|
||
|
||
def test_check_numbers_catches_invented():
|
||
invented = rv.check_numbers("管理员有 6 个。", "管理员有 6 个，商家有 5 个。")
|
||
assert invented == ["5"]
|
||
|
||
|
||
def test_check_numbers_ignores_prose_without_numbers():
|
||
assert rv.check_numbers("管理员有 6 个。", "这次没能查到有效数据，建议重试。") == []
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# review：调模型那一段
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class FakeModel:
|
||
"""最小的 with_structured_output 替身。calls 记录被调了几次。"""
|
||
|
||
def __init__(self, reply):
|
||
self.reply = reply
|
||
self.calls = 0
|
||
|
||
def with_structured_output(self, schema):
|
||
return self
|
||
|
||
async def ainvoke(self, messages):
|
||
self.calls += 1
|
||
if isinstance(self.reply, Exception):
|
||
raise self.reply
|
||
return self.reply
|
||
|
||
|
||
SIGNALS = [rv.Signal("step_limit", "agent 用满了 8 步还没给出结论", rv.BROKEN)]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_review_annotates_without_rewriting():
|
||
model = FakeModel({"note": "这次没跑完就收尾了。", "answer": "", "retry": True})
|
||
result = await rv.review(model, question="有几个管理员", answer="管理员有 6 个。",
|
||
signals=SIGNALS)
|
||
assert result.verdict == "annotated"
|
||
assert result.answer is None # 沿用原答案
|
||
assert result.note == "这次没跑完就收尾了。"
|
||
assert result.retry is True
|
||
assert result.severity == rv.BROKEN
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_review_adopts_clean_rewrite():
|
||
model = FakeModel({
|
||
"note": "步数用满了。", "answer": "管理员有 6 个（这个数可能不完整）。", "retry": False,
|
||
})
|
||
result = await rv.review(model, question="q", answer="管理员有 6 个。", signals=SIGNALS)
|
||
assert result.verdict == "rewritten"
|
||
assert result.answer == "管理员有 6 个（这个数可能不完整）。"
|
||
# 规则层管资格、模型有否决权：step_limit 本来够格重跑，模型说不必就不跑
|
||
assert result.retry is False
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_model_cannot_force_an_unhelpful_retry():
|
||
"""检索什么都没查到，重跑一遍还是查不到——模型说要重跑也不能听。
|
||
|
||
否则就是白烧一次建图加一次跑图，而用户等的时间翻倍。
|
||
"""
|
||
model = FakeModel({"note": "知识库里没有这个。", "answer": "", "retry": True})
|
||
result = await rv.review(
|
||
model, question="q", answer="没查到。",
|
||
signals=[rv.Signal("retrieve_empty", "知识库里没有检索到相关内容", rv.BROKEN)],
|
||
)
|
||
assert result.severity == rv.BROKEN
|
||
assert result.retry is False
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_review_reverts_rewrite_with_invented_numbers():
|
||
"""复核手上只有原答案，它写出来的新数字只能是凭空来的。"""
|
||
model = FakeModel({
|
||
"note": "步数用满了。", "answer": "管理员有 6 个，商家有 5 个。", "retry": False,
|
||
})
|
||
result = await rv.review(model, question="q", answer="管理员有 6 个。", signals=SIGNALS)
|
||
assert result.verdict == "reverted"
|
||
assert result.answer is None # 那份正文一个字都不能用
|
||
assert "5" in result.note # 但要说清是为什么退的
|
||
assert "步数用满了。" in result.note # 说明本身还是要给
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_review_survives_model_failure():
|
||
"""复核只是加一层说明，它自己挂了不能把已经到手的答案也带走。"""
|
||
model = FakeModel(RuntimeError("502"))
|
||
result = await rv.review(model, question="q", answer="管理员有 6 个。", signals=SIGNALS)
|
||
assert result.verdict == "annotated"
|
||
assert result.answer is None
|
||
assert result.note == SIGNALS[0].detail # 退回规则层自己的措辞
|
||
assert result.retry is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_review_survives_garbage_reply():
|
||
model = FakeModel("这不是 JSON，也不是对象")
|
||
result = await rv.review(model, question="q", answer="管理员有 6 个。", signals=SIGNALS)
|
||
assert result.verdict == "annotated"
|
||
assert result.note
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 接口
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def _make_run(client, output, events):
|
||
"""直接落一条运行和它的事件，省掉真跑一遍图。"""
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import Run, RunEvent
|
||
|
||
async with SessionLocal() as session:
|
||
run = Run(workflow_id=None, status="succeeded", output=output, graph={})
|
||
session.add(run)
|
||
await session.flush()
|
||
for i, e in enumerate(events):
|
||
session.add(RunEvent(
|
||
run_id=run.id, seq=i + 1, type=e["type"],
|
||
node_id=e["node_id"] or None, ts=0.0, data=e["data"],
|
||
))
|
||
await session.commit()
|
||
return run.id
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_clean_run_never_touches_the_model(client, monkeypatch):
|
||
"""干净运行零成本——这条是整层设计的前提，要用「模型根本没被碰过」来断言。"""
|
||
async def explode(*a, **k):
|
||
raise AssertionError("干净的运行不该调模型")
|
||
|
||
monkeypatch.setattr("app.providers.factory.get_chat_model", explode)
|
||
monkeypatch.setattr("app.api.copilot.get_chat_model", explode)
|
||
|
||
run_id = await _make_run(client, OK_OUTPUT, [
|
||
ev(EventType.TOOL_START, tool="t"), ev(EventType.TOOL_END, tool="t"),
|
||
])
|
||
resp = await client.post("/api/copilot/review", json={"run_id": run_id, "question": "q"})
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["verdict"] == "ok"
|
||
assert body["signals"] == []
|
||
assert body["answer"] is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_review_endpoint_reports_signals(client, monkeypatch):
|
||
"""没配模型时也得把规则层扫到的东西说出来，而不是静默放行。"""
|
||
from app.providers.factory import ProviderNotConfigured
|
||
|
||
async def unconfigured(*a, **k):
|
||
raise ProviderNotConfigured("没有可用的 provider")
|
||
|
||
monkeypatch.setattr("app.api.copilot.get_chat_model", unconfigured)
|
||
|
||
run_id = await _make_run(client, OK_OUTPUT, [
|
||
ev(EventType.TOOL_ERROR, tool="db_query__x", error="no such table: users"),
|
||
])
|
||
body = (await client.post(
|
||
"/api/copilot/review", json={"run_id": run_id, "question": "q"}
|
||
)).json()
|
||
assert body["verdict"] == "annotated"
|
||
assert body["severity"] == rv.BROKEN
|
||
assert [s["kind"] for s in body["signals"]] == ["tool_error"]
|
||
assert "no such table" in body["note"]
|
||
assert body["retry"] is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_review_endpoint_404(client):
|
||
resp = await client.post("/api/copilot/review", json={"run_id": "没这个", "question": "q"})
|
||
assert resp.status_code == 404
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 带逐段证据的答案：只加说明，不改写
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class RecordingModel(FakeModel):
|
||
def __init__(self, reply):
|
||
super().__init__(reply)
|
||
self.seen: list = []
|
||
|
||
async def ainvoke(self, messages):
|
||
self.seen.append(messages)
|
||
return await super().ainvoke(messages)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_review_never_rewrites_an_answer_that_carries_evidence():
|
||
"""报告节点产出的答案，每个数字都按位置对应着证据文档。复核把它整段改写，
|
||
文档和答案就对不上了——界面上点开的出处指向一段已经不存在的文字。"""
|
||
model = RecordingModel({"note": "报告被要求重写过一次。", "answer": "改写后的周报，销售额 45,678.5元。",
|
||
"retry": False})
|
||
result = await rv.review(model, question="q", answer="周报：销售额 45,678.5元。", signals=SIGNALS,
|
||
locked=True)
|
||
assert result.verdict == "annotated"
|
||
assert result.answer is None and result.original is None
|
||
assert result.note == "报告被要求重写过一次。"
|
||
prompt = "\n".join(str(text) for _, text in model.seen[0])
|
||
assert "不能改写" in prompt
|
||
|
||
|
||
def test_report_repair_is_a_degraded_signal():
|
||
signals = rv.scan([warn("报告里有 1 处没通过核对（「45678」），已要求写作者重写（第 1 次）",
|
||
"report_repair")], OK_OUTPUT)
|
||
assert [(s.kind, s.severity) for s in signals] == [("report_repair", rv.DEGRADED)]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_review_endpoint_keeps_answers_with_evidence(client, monkeypatch):
|
||
model = FakeModel({"note": "报告被要求重写过一次。", "answer": "整段改写过的周报。", "retry": False})
|
||
|
||
async def fake_model(*a, **k):
|
||
return model, "fake"
|
||
|
||
monkeypatch.setattr("app.api.copilot.get_chat_model", fake_model)
|
||
output = {"周报": "周报：销售额 45,678.5元。",
|
||
"_evidence": {"report_node": "write", "doc_artifact": "a" * 64, "fields": ["周报"]}}
|
||
run_id = await _make_run(client, output, [warn("报告里有 1 处没通过核对，已要求写作者重写", "report_repair")])
|
||
body = (await client.post("/api/copilot/review", json={"run_id": run_id, "question": "q"})).json()
|
||
assert model.calls == 1
|
||
assert body["verdict"] == "annotated" and body["answer"] is None
|
||
assert body["note"] == "报告被要求重写过一次。"
|
||
|
||
|
||
def test_agent_field_warnings_are_degraded_signals():
|
||
"""agent 的结构化字段和快照对不上（以快照为准）、或者核对不了（记成空值）：答案能用，但有缺口要说明。
|
||
按自己的 kind 记下，界面才说得清是哪一种，而不是一个笼统的「警告」。"""
|
||
signals = rv.scan([
|
||
warn("字段 orders：模型报 30，快照是 3，已按快照取值", "agent_field_mismatch"),
|
||
warn("字段 region 核对不了出处，记为空值", "agent_field_unverified"),
|
||
], OK_OUTPUT)
|
||
assert [(s.kind, s.severity) for s in signals] == [
|
||
("agent_field_mismatch", rv.DEGRADED), ("agent_field_unverified", rv.DEGRADED)]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 报告「改过一次」的信号按最终结果定档
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def _repaired(final_ok: bool) -> list[dict]:
|
||
violations = [] if final_ok else [{"code": "bare_number", "message": "裸数字 12"}]
|
||
return [
|
||
ev(EventType.NODE_STARTED, node_id="write", label="写周报"),
|
||
ev(EventType.LOG, node_id="write", level="warn", code="report_repair", count=2, attempt=1,
|
||
message="报告有 2 处未通过核对（裸数字 12、裸数字 30），已要求模型重写（第 1 次）"),
|
||
ev(EventType.REPORT_CHECKED, node_id="write", ok=final_ok, violations=violations, repairs=1),
|
||
]
|
||
|
||
|
||
def test_a_repair_that_finally_passed_is_not_a_gap():
|
||
"""首稿没过核对、重写一次之后通过（最后一次 report.checked ok、没有违规）：不进异常清单，复核不为它叫模型；
|
||
复核被别的信号叫起来时，说明里写「首稿有 N 处没过核对，重写后已全部通过」。"""
|
||
rows = _repaired(True)
|
||
assert rv.scan(rows, OK_OUTPUT) == []
|
||
[note] = rv.notes(rows)
|
||
assert note == "报告「写周报」首稿有 2 处没过核对，重写后已全部通过"
|
||
prompt = rv.review_prompt("上周产量", "答案", [rv.Signal("tool_error", "工具 x 调用失败", rv.BROKEN)],
|
||
settled=rv.notes(rows))
|
||
assert "重写后已全部通过" in prompt and "不要写成缺口" in prompt
|
||
assert prompt.index("检测到的异常") < prompt.index("已经处理好的")
|
||
assert "首稿有 2 处" not in prompt.split("已经处理好的")[0], "处理好了的不在异常清单里"
|
||
|
||
|
||
def test_a_repair_that_still_failed_stays_a_gap():
|
||
rows = _repaired(False)
|
||
[signal] = rv.scan(rows, OK_OUTPUT)
|
||
assert signal.kind == "report_repair" and signal.severity == rv.DEGRADED
|
||
assert rv.notes(rows) == []
|
||
|
||
|
||
def test_a_rewrite_counts_by_the_final_judgement():
|
||
base = [ev(EventType.NODE_STARTED, node_id="write", label="写周报"),
|
||
ev(EventType.LOG, node_id="write", level="warn", code="report_rewrite", count=1,
|
||
message="裁判认为 1 句结论证据不支持（……），已退回写作模型仅修改这几句")]
|
||
good = [*base, ev(EventType.REPORT_CHECKED, node_id="write", ok=True, violations=[],
|
||
judge={"counts": {"supported": 3, "contradicted": 0, "insufficient": 0}})]
|
||
bad = [*base, ev(EventType.REPORT_CHECKED, node_id="write", ok=True, violations=[],
|
||
judge={"counts": {"supported": 2, "contradicted": 1, "insufficient": 0}})]
|
||
assert rv.scan(good, OK_OUTPUT) == [] and rv.notes(good) == ["报告「写周报」有 1 句结论裁判认为证据不支持，改写后都有了依据"]
|
||
assert [s.kind for s in rv.scan(bad, OK_OUTPUT)] == ["report_rewrite"] and rv.notes(bad) == []
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 分析能力一轮的信号：只写说明、不自动重来（00-总方案 §3.7）
|
||
# --------------------------------------------------------------------------
|
||
|
||
#: check 下判断词重写一次后仍没挂对分析结论：报告照常产出，标在报告里
|
||
CLAIMS_LEFT = "报告有 1 处分析说法未通过核对（1 处判断没挂分析结论：「主要来自」），已要求模型重写 1 次仍未改掉：…这几处已标注"
|
||
REFUSED = "「周报分析」的波动基线没有做成：历史只有 5 期，至少要 8 期"
|
||
SKIPPED = "指标「销售额环比」用到的两次查询没做成可比性检查：探查超时。指标照常计算，两期能不能直接比请自行核对"
|
||
|
||
|
||
def test_analysis_signals_are_registered_as_gaps_that_only_need_a_note():
|
||
signals = rv.scan([warn(CLAIMS_LEFT, "report_analysis_claims"), warn(REFUSED, "analysis_refused"),
|
||
warn(SKIPPED, "metric_comparability_skipped")], OK_OUTPUT)
|
||
assert [(s.kind, s.severity) for s in signals] == [
|
||
("report_analysis_claims", rv.DEGRADED), ("analysis_refused", rv.DEGRADED),
|
||
("metric_comparability_skipped", rv.DEGRADED)]
|
||
# NOTE_ONLY 另有一种不是日志代码的：分析调用写错了一直没改对（analysis_call_failed，见 tests/test_review_analysis.py）
|
||
assert {s.kind for s in signals} == rv.NOTE_ONLY - {rv.ANALYSIS_CALL_FAILED}
|
||
assert not rv.should_retry(signals) and not rv.NOTE_ONLY & (rv.RETRYABLE | rv.REBUILD)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_a_run_with_only_note_only_signals_is_annotated_without_calling_the_model():
|
||
model = FakeModel(AssertionError("不该叫模型"))
|
||
signals = rv.scan([warn(CLAIMS_LEFT, "report_analysis_claims"), warn(REFUSED, "analysis_refused")], OK_OUTPUT)
|
||
result = await rv.review(model, question="q", answer="答案", signals=signals, locked=True)
|
||
assert model.calls == 0
|
||
assert result.verdict == "annotated" and result.retry is False and result.severity == rv.DEGRADED
|
||
assert result.note == f"{CLAIMS_LEFT}；{REFUSED}" and result.answer is None
|
||
# 和别的信号一起出现时照常叫模型
|
||
mixed = [*signals, rv.Signal("tool_error", "工具 x 调用失败", rv.BROKEN)]
|
||
model = FakeModel({"note": "取数失败了一次。", "answer": "", "retry": True})
|
||
assert (await rv.review(model, question="q", answer="答案", signals=mixed)).note == "取数失败了一次。"
|
||
assert model.calls == 1
|
||
|
||
|
||
def _left_after_rewrite(mode: str, violations: list[dict], count: int = 2) -> list[dict]:
|
||
return [
|
||
ev(EventType.NODE_STARTED, node_id="write", label="写周报"),
|
||
ev(EventType.LOG, node_id="write", level="warn", code="report_repair", count=count, attempt=1,
|
||
message=f"报告有 {count} 处未通过核对（「12」、「主要来自」），已要求模型重写（第 1 次）"),
|
||
ev(EventType.REPORT_CHECKED, node_id="write", ok=not violations, violations=violations, repairs=1,
|
||
analysis_claims=mode),
|
||
]
|
||
|
||
|
||
CLAIM = {"code": "analysis_claim_uncited", "text": "主要来自", "message": "「主要来自」是分析判断"}
|
||
|
||
|
||
def test_check_mode_leftovers_do_not_make_the_rewrite_count_as_failed():
|
||
"""check 下重写一次后仍有的说法违规是标注（同缺失断言，各有自己的信号）：「改过一次」算改好了，不再另报 report_repair；
|
||
复核被叫起来时照实说剩下的是哪几处。"""
|
||
rows = [*_left_after_rewrite("check", [CLAIM]), warn(CLAIMS_LEFT, "report_analysis_claims")]
|
||
assert [s.kind for s in rv.scan(rows, OK_OUTPUT)] == ["report_analysis_claims"]
|
||
assert rv.notes(rows) == ["报告「写周报」首稿有 2 处没过核对，重写后只剩已标出的 1 处分析说法"]
|
||
# 一处都没改掉：不说「改好了」，剩下的由它自己的信号说
|
||
assert rv.notes(_left_after_rewrite("check", [CLAIM], count=1)) == []
|
||
|
||
|
||
def test_absence_claim_leftovers_are_annotations_too():
|
||
absent = {"code": "absence_claim", "text": "没有查到", "message": "缺失断言"}
|
||
rows = _left_after_rewrite("check", [absent, CLAIM], count=3)
|
||
assert rv.scan(rows, OK_OUTPUT) == []
|
||
assert rv.notes(rows) == ["报告「写周报」首稿有 3 处没过核对，重写后只剩已标出的 1 处缺失断言、1 处分析说法"]
|
||
|
||
|
||
@pytest.mark.parametrize("mode, violations", [
|
||
("strict", [CLAIM]), # strict 下是出具缺口，不是标注
|
||
("check", [CLAIM, {"code": "uncited_number", "text": "12"}]), # 还有别的违规
|
||
("off", [CLAIM]),
|
||
("check", [CLAIM] * 20), # 带满了 20 条：后面可能还有
|
||
])
|
||
def test_other_leftovers_still_count_as_a_failed_rewrite(mode, violations):
|
||
rows = _left_after_rewrite(mode, violations)
|
||
assert [s.kind for s in rv.scan(rows, OK_OUTPUT)] == ["report_repair"] and rv.notes(rows) == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_review_endpoint_does_not_call_the_model_for_a_check_mode_leftover(client, monkeypatch):
|
||
model = FakeModel(AssertionError("不该叫模型"))
|
||
|
||
async def fake_model(*a, **k):
|
||
return model, "fake"
|
||
|
||
monkeypatch.setattr("app.api.copilot.get_chat_model", fake_model)
|
||
output = {"周报": "周报：销售额 45,678.5元，下降主要来自华东。",
|
||
"_evidence": {"report_node": "write", "doc_artifact": "a" * 64, "fields": ["周报"]}}
|
||
run_id = await _make_run(client, output, [*_left_after_rewrite("check", [CLAIM]),
|
||
warn(CLAIMS_LEFT, "report_analysis_claims")])
|
||
body = (await client.post("/api/copilot/review", json={"run_id": run_id, "question": "q"})).json()
|
||
assert model.calls == 0
|
||
assert body["verdict"] == "annotated" and body["note"] == CLAIMS_LEFT and body["retry_instruction"] is None
|
||
assert [s["kind"] for s in body["signals"]] == ["report_analysis_claims"]
|