From 72be600ea223cf176a1303cf070326ea56a04f7a Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 21 Oct 2026 14:08:56 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BD=AC=E6=AD=A3=20W1-b=20=E7=9A=84=E7=AB=AF?= =?UTF-8?q?=E5=88=B0=E7=AB=AF=E6=B5=8B=E8=AF=95=EF=BC=88tests/test=5Fanaly?= =?UTF-8?q?sis=5Fe2e=5Fevidence.py=EF=BC=89=EF=BC=9A=E5=88=86=E6=9E=90?= =?UTF-8?q?=E5=8D=A1=E3=80=81agent=20=E5=88=86=E6=9E=90=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=E3=80=81=E5=8F=A3=E5=BE=84=E5=8D=A1=E6=8C=87=E6=A0=87=E4=B8=8A?= =?UTF-8?q?=E7=9A=84=E4=B8=A4=E6=9C=9F=E5=8F=AF=E6=AF=94=E6=80=A7=E5=90=84?= =?UTF-8?q?=E5=8F=91=E8=B5=B7=E4=B8=80=E6=AC=A1=E7=9C=9F=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=EF=BC=8C=E7=89=87=E6=AE=B5=E6=8E=A5=E5=8F=A3=E7=9A=84=E5=87=BA?= =?UTF-8?q?=E5=A4=84=E9=93=BE=E3=80=81=E8=AF=81=E6=8D=AE=E5=9B=BE=E3=80=81?= =?UTF-8?q?=E5=AE=A1=E8=AE=A1=E3=80=81=E6=8A=A5=E5=91=8A=E5=AF=BC=E5=87=BA?= =?UTF-8?q?=E3=80=81=E5=8F=AF=E9=AA=8C=E8=AF=81=E6=8A=A5=E5=91=8A=E9=83=BD?= =?UTF-8?q?=E6=8C=89=E5=B0=81=E5=AD=98=E7=9A=84=E8=BE=93=E5=85=A5=E7=9C=9F?= =?UTF-8?q?=E5=A4=8D=E7=AE=97=E3=80=81=E9=80=90=E5=AD=97=E6=AE=B5=E7=9B=B8?= =?UTF-8?q?=E5=90=8C=EF=BC=9B=E6=89=93=E5=8D=B0=E6=94=B9=E6=88=90=E6=96=AD?= =?UTF-8?q?=E8=A8=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/tests/test_analysis_e2e_evidence.py | 170 ++++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 backend/tests/test_analysis_e2e_evidence.py diff --git a/backend/tests/test_analysis_e2e_evidence.py b/backend/tests/test_analysis_e2e_evidence.py new file mode 100644 index 00000000..c5d70e54 --- /dev/null +++ b/backend/tests/test_analysis_e2e_evidence.py @@ -0,0 +1,170 @@ +"""分析结论从入口到证据的整条链路，接真算子、真复算（不用替身）：分析卡（W1-c）、agent 分析工具（W1-c）、口径卡指标上的 +两期可比性（W1-c）各发起一次运行，算子是 W1-a 的，证据接入（W1-b）的片段接口、证据图、审计、报告导出、可验证报告都按 +封存的输入复算一遍、逐字段相同。 + +库是虚构的零售库 shop（tests/test_analysis_card：orders(region, week, amount)，华东、华南、华北两周）。写作模型换成脚本 +（MockChatModel._decide），不调用任何模型。 +""" +from __future__ import annotations + +import json +import re + +from httpx import ASGITransport, AsyncClient +from langchain_core.messages import AIMessage, ToolMessage + +from app.analysis import contract +from app.core import artifact_store +from app.engine.evidence import iter_segments +from app.main import app +from tests.test_analysis_card import REGION, WEEKLY, edge, engine_up, node, run_graph, shop # noqa: F401 - 夹具按名字引入 +from tests.test_analysis_tools import DECOMPOSE, DECOMPOSE_ARGS, SQL, TOOL + +TEXT = "## 异动\n\n[[a:A1]]。华东占 [[a:A1.top_share|pct]]，变化 [[v:A1.r0.delta]]。[[see:A1]]" +WHERE = {"id": "where", "op": "decompose", "mode": "additive", "from": "by_region", + "params": {"by": ["region"], "period": "week", "base": "2026-W41", "target": "2026-W42", "measure": "gmv", + "headline": {"base": "weekly.r0.gmv", "target": "weekly.r1.gmv"}, "label": "销售额"}, + "hypothesis": "下降是不是集中在某个地区"} +#: 判定语的出处链：判定 → 方法 → 计算 → 输入（两格总数 + 拆解的那份快照，后面接两份快照的查询步骤）→ 复算 → 封存 +CHAIN = ["analysis", "analysis_method", "analysis_calc", "analysis_input", "analysis_input", "analysis_input", "query", + "query", "analysis_recompute", "analysis_seal"] +#: 片段：判定语（系统渲染）、具名标量、结果表的一格 +SEGMENTS = {"a:A1": ("analysis", "销售额下降 194（-9.2%），主要来自华东（-120，占 61.9%）"), + "a:A1.top_share|pct": ("number", "61.9%"), "v:A1.r0.delta": ("number", "-120")} + + +def _report(*before, contract_=None): + """报告撰写 → 成果（带契约，报告来自 write）；before 是报告前面的节点，按顺序连成一条。""" + nodes = [node("start", "input"), *before, node("write", "report", instructions="写周报", on_violation="flag"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}], + contract=contract_ or {"report_from": "write"})] + ids = [n["id"] for n in nodes] + return {"nodes": nodes, "edges": [edge(a, b) for a, b in zip(ids, ids[1:])]} + + +def _segments(row): + doc = artifact_store.load(row.output["_evidence"]["doc_artifact"]) + return {s["ref"]: s for s in iter_segments(doc) if s.get("ref")} + + +def _embedded(html: str, element: str) -> dict: + return json.loads(re.search(rf'', html, re.S).group(1)) + + +async def test_a_card_conclusion_opens_and_rechecks_everywhere(monkeypatch, shop): + from app.providers import mock_model + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=TEXT)) + graph = _report(node("weekly", "tool", tool="db_query__shop", args={"sql": WEEKLY}), + node("by_region", "tool", tool="db_query__shop", args={"sql": REGION}), + node("card", "analysis", items=[WHERE])) + row = await run_graph(graph) + assert row.status == "succeeded", row.error + segs = _segments(row) + assert {ref: (s["kind"], s["text"]) for ref, s in segs.items()} == SEGMENTS + assert not [s for s in segs.values() if s.get("issue")] + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{segs['a:A1']['id']}")).json() + assert [s["step"] for s in body["chain"]] == CHAIN + judged = body["chain"][0] + assert (judged["origin"], judged["item"]) == ("card", "where") + again = next(s for s in body["chain"] if s["step"] == "analysis_recompute") + assert again["status"] == "same" and again["method"] == "decompose.additive@1", again + inputs = [s for s in body["chain"] if s["step"] == "analysis_input"] + assert [s["role"] for s in inputs] == ["headline_base", "headline_target", "data"] + assert all(s["sealed"] for s in inputs) + assert [s["closure"]["closed"] for s in inputs if s.get("closure")] == [True, True] + assert body["seal"]["covered"] is True, body["seal"] + + graph_view = (await client.get(f"/api/runs/{row.id}/evidence")).json() + a1 = next(e for e in graph_view["evidence"] if e["alias"] == "A1") + assert {k: a1.get(k) for k in ("kind", "op", "verdict", "item", "sealed")} == { + "kind": "analysis", "op": "decompose", "verdict": "driver", "item": "where", "sealed": True} + + # 报告导出：判定语这一处的出处链同样复算一致 + data = _embedded((await client.get(f"/api/runs/{row.id}/report.html")).text, "report-data") + chain = data["chains"][data["refs"][segs["a:A1"]["id"]]] + assert next(s for s in chain if s["step"] == "analysis_recompute")["status"] == "same" + + # 可验证报告：结论按原字节嵌入、导出时服务端复算一致；分析卡那一项的 analysis.end 带规范字符串 + bundle = await client.get(f"/api/runs/{row.id}/bundle.html") + assert bundle.status_code == 200, bundle.text[:300] + b = _embedded(bundle.text, "bundle-data") + items = [a for a in b["artifacts"] if a["kind"] == contract.ARTIFACT_KIND] + assert items and all("b64" in a and a["recheck"]["status"] == "same" for a in items) + ends = [e for e in b["events"] if e["type"] == "analysis.end"] + assert ends and all("canonical" in e for e in ends) + + audit = (await client.get(f"/api/runs/{row.id}/evidence/audit")).json() + rows = [r for g in audit["groups"] for r in g["rows"] if r.get("ref") == "a:A1"] + assert [(r["kind"], r["evidence_kind"], r["sealed"]) for r in rows] == [("analysis", "analysis", True)] + + +async def test_an_agent_conclusion_opens_and_rechecks_everywhere(monkeypatch, shop): + from app.providers import mock_model + + calls = [(TOOL, {"sql": SQL}), (DECOMPOSE, DECOMPOSE_ARGS)] + + def decide(model, messages): + if not model.tools: + return AIMessage(content=TEXT) + done = sum(isinstance(m, ToolMessage) for m in messages) + if done < len(calls): + name, args = calls[done] + return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": f"call_{done}"}]) + return AIMessage(content="销售额下降，主要来自华东。") + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide) + row = await run_graph(_report(node("bot", "agent", prompt="销售额为什么下降", tools=[TOOL, DECOMPOSE], + approval="never"))) + assert row.status == "succeeded", row.error + segs = _segments(row) + assert {ref: (s["kind"], s["text"]) for ref, s in segs.items()} == SEGMENTS + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{segs['a:A1']['id']}")).json() + judged = body["chain"][0] + assert (judged["origin"], judged["call_id"]) == ("agent", "call_1") + again = next(s for s in body["chain"] if s["step"] == "analysis_recompute") + assert again["status"] == "same", again + + b = _embedded((await client.get(f"/api/runs/{row.id}/bundle.html")).text, "bundle-data") + # 分析工具的 tool.end（预览里是结论的前一截）和结论同一个认法：结论随文件附带得了才带内容 + ends = [e for e in b["events"] if e["type"] == "tool.end" and e["node_id"] == "bot"] + assert len(ends) == 2 and all("canonical" in e for e in ends) + assert [a["recheck"]["status"] for a in b["artifacts"] if a["kind"] == contract.ARTIFACT_KIND] == ["same"] + + +async def test_a_metric_comparability_opens_and_rechecks_everywhere(monkeypatch, shop): + from app.providers import mock_model + + this = "SELECT SUM(amount) AS gmv FROM orders WHERE week = '2026-W42'" + last = "SELECT SUM(amount) AS gmv FROM orders WHERE week = '2026-W41'" + wow = "(cell(nodes.this_week, 0, 'gmv') - cell(nodes.last_week, 0, 'gmv')) / cell(nodes.last_week, 0, 'gmv')" + monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content="环比 [[m:wow]]。")) + graph = _report(node("this_week", "tool", tool="db_query__shop", args={"sql": this}), + node("last_week", "tool", tool="db_query__shop", args={"sql": last}), + node("card", "metrics", caliber="周报口径", + metrics=[{"id": "wow", "name": "环比", "expression": wow, "format": "percent_of_ratio", + "decimals": 3}])) + row = await run_graph(graph) + assert row.status == "succeeded", row.error + segs = _segments(row) + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{segs['m:wow']['id']}")).json() + view = body["chain"][0]["comparability"] + # 表里只有周次这一列文字、没有业务日期：窗口长度、整期、星期构成核对不了，不判不可比 + assert (view["verdict"], view["phrase"], view["reasons"]) == ( + "comparable", "两期可比，窗口长度、整期、星期构成核对不了", []) + assert view["sealed"] is True and view["hash_ok"] is True and view["method"] == "comparable@1" + steps = view["steps"] + assert [s["step"] for s in steps] == ["analysis", "analysis_method", "analysis_calc", "analysis_input", + "analysis_input", "analysis_recompute", "analysis_seal"] + assert steps[-2]["status"] == "same" + assert (steps[-1]["via"], steps[-1]["sealed"]) == ("metric_set", True) + + b = _embedded((await client.get(f"/api/runs/{row.id}/bundle.html")).text, "bundle-data") + found = [a for a in b["artifacts"] if a["kind"] == contract.ARTIFACT_KIND] + assert len(found) == 1 and "b64" in found[0] and found[0]["recheck"]["status"] == "same"