mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 03:51:58 +00:00
ThearchyHelios
315 lines
16 KiB
Python
315 lines
16 KiB
Python
"""分析入口接真算子跑一遍（不用替身）：agent 分析工具（含下钻）、分析卡（波动基线 + 变化拆解）、口径卡的可比性挂钩、
|
||
审批恢复时的重放。断言工件、台账、交回模型的那一段、节点产出，并用 analysis.recompute 按封存的工件复算、逐字段相同。
|
||
|
||
虚构零售库 shop：orders(day, region, channel, amount)，2026-03-02 起 12 个整周，每周每个地区、渠道两笔（周三、周日），
|
||
最后一周华东掉三成。模型换成脚本（MockChatModel._decide），不调用任何模型。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import sqlite3
|
||
from datetime import date, timedelta
|
||
|
||
import pytest
|
||
from langchain_core.messages import AIMessage, ToolMessage
|
||
from sqlalchemy import select
|
||
|
||
import app.analysis as analysis_ops
|
||
from app.analysis import contract
|
||
from app.core import artifact_store
|
||
from app.core.artifact_store import canonical_json, content_hash
|
||
from app.data.engine import engines
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import DataSource, Run, RunEvent
|
||
from app.engine.runner import run_manager
|
||
|
||
TOOL = "db_query__shop"
|
||
DECOMPOSE = contract.TOOL_NAMES["decompose"]
|
||
WEEK = "date(day, '-6 days', 'weekday 1')"
|
||
BY_REGION = (f"SELECT region, {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE day >= '2026-05-11' "
|
||
"GROUP BY region, week ORDER BY region, week")
|
||
TOTALS = f"SELECT {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE day >= '2026-05-11' GROUP BY week ORDER BY week"
|
||
EAST = (f"SELECT channel, {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE region = '华东' AND day >= '2026-05-11' "
|
||
"GROUP BY channel, week ORDER BY channel, week")
|
||
WEEKLY = f"SELECT {WEEK} AS week, SUM(amount) AS gmv FROM orders GROUP BY week ORDER BY week"
|
||
TWO_WEEKS = {"period": "week", "base": "2026-05-11", "target": "2026-05-18", "measure": "gmv"}
|
||
A1_ARGS = {"mode": "additive", "query": "Q1", "by": ["region"], **TWO_WEEKS, "label": "销售额",
|
||
"headline": {"base": "Q2.r0.gmv", "target": "Q2.r1.gmv"}, "hypothesis": "下降是不是集中在某个地区"}
|
||
A2_ARGS = {"mode": "additive", "query": "Q3", "by": ["channel"], **TWO_WEEKS, "parent": {"of": "A1", "key": ["华东"]},
|
||
"hypothesis": "华东是哪个渠道掉的"}
|
||
|
||
|
||
def _orders():
|
||
out, i = [], 0
|
||
base = {"华东": 400, "华南": 300, "华北": 250, "西南": 50}
|
||
for w in range(12):
|
||
monday = date(2026, 3, 2) + timedelta(weeks=w)
|
||
for region, amount in base.items():
|
||
for channel, share in (("线上", 3), ("线下", 2)):
|
||
value = amount * share // 5 + (w % 3) * 2
|
||
if w == 11 and region == "华东":
|
||
value = value * 7 // 10
|
||
for offset in (2, 6):
|
||
i += 1
|
||
out.append((i, (monday + timedelta(days=offset)).isoformat(), region, channel, value))
|
||
return out
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
async def engine_up():
|
||
await run_manager.setup()
|
||
yield
|
||
await run_manager.shutdown()
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
async def shop(tmp_path):
|
||
path = tmp_path / "shop.db"
|
||
db = sqlite3.connect(path)
|
||
db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, channel TEXT, amount INTEGER);")
|
||
db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", _orders())
|
||
db.commit()
|
||
db.close()
|
||
async with SessionLocal() as session:
|
||
row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
|
||
if row is None:
|
||
row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
|
||
session.add(row)
|
||
row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {}
|
||
await session.commit()
|
||
source_id = row.id
|
||
await engines.invalidate(source_id)
|
||
yield
|
||
await engines.invalidate(source_id)
|
||
|
||
|
||
def node(nid, ntype, **config):
|
||
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
|
||
|
||
|
||
def chain(*nodes):
|
||
return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
|
||
|
||
|
||
class Script:
|
||
"""Agent 依次发 calls 里的工具调用，再给结论；seen 记下每次带工具的调用看到的消息。"""
|
||
|
||
def __init__(self, monkeypatch, calls):
|
||
from app.providers import mock_model
|
||
|
||
self.seen: list[list] = []
|
||
|
||
def decide(model, messages):
|
||
if not model.tools:
|
||
return AIMessage(content="收尾")
|
||
self.seen.append(list(messages))
|
||
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)
|
||
|
||
def replies(self):
|
||
return [m.content for m in self.seen[-1] if isinstance(m, ToolMessage)]
|
||
|
||
|
||
class Counting:
|
||
"""真算子外面包一层只数次数（重放时不该再算）。"""
|
||
|
||
def __init__(self, monkeypatch):
|
||
self.calls = 0
|
||
real = analysis_ops.run
|
||
|
||
async def run(*args, **kwargs):
|
||
self.calls += 1
|
||
return await real(*args, **kwargs)
|
||
|
||
monkeypatch.setattr(analysis_ops, "run", run)
|
||
|
||
|
||
async def wait(run_id, statuses=("succeeded", "failed")):
|
||
for _ in range(600):
|
||
await asyncio.sleep(0.05)
|
||
async with SessionLocal() as session:
|
||
row = await session.get(Run, run_id)
|
||
if row.status in statuses and (row.status == "interrupted" or row.manifest_seq is not None):
|
||
return row
|
||
raise AssertionError(f"run {run_id} 没有结束：{row.status}")
|
||
|
||
|
||
async def events(run_id, etype=None, node_id=None):
|
||
async with SessionLocal() as session:
|
||
q = select(RunEvent).where(RunEvent.run_id == run_id)
|
||
if etype:
|
||
q = q.where(RunEvent.type == etype)
|
||
if node_id:
|
||
q = q.where(RunEvent.node_id == node_id)
|
||
return [e.data for e in (await session.execute(q.order_by(RunEvent.seq))).scalars()]
|
||
|
||
|
||
def agent_graph(approval="never"):
|
||
return chain(node("start", "input"),
|
||
node("bot", "agent", prompt="销售额为什么下降", tools=[TOOL, DECOMPOSE], approval=approval),
|
||
node("out", "output", fields=[{"name": "answer", "value": "{{ nodes.bot.text }}"}]))
|
||
|
||
|
||
def recomputes(stored):
|
||
"""按封存的工件复算（查询快照、上一层结论都按 sha256 取回），和封存的逐字段相同。"""
|
||
again = analysis_ops.recompute(stored, artifact_store.load, load_result=artifact_store.load)
|
||
assert contract.same(stored, again), {k: (stored.get(k), again.get(k)) for k in stored if stored.get(k) != again.get(k)}
|
||
|
||
|
||
AGENT_CALLS = [(TOOL, {"sql": BY_REGION}), (TOOL, {"sql": TOTALS}), (TOOL, {"sql": EAST}),
|
||
(DECOMPOSE, A1_ARGS), (DECOMPOSE, A2_ARGS)]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# agent 分析工具：拆解、下钻
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_an_agent_decomposes_and_drills_down_with_the_real_operator(monkeypatch):
|
||
script = Script(monkeypatch, AGENT_CALLS)
|
||
run = await run_manager.start(graph=agent_graph(), input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
|
||
ends = await events(row.id, "tool.end", "bot")
|
||
q1, q2, q3 = (ends[i]["query_artifact"] for i in range(3))
|
||
a1_artifact, a2_artifact = ends[3]["analysis_artifact"], ends[4]["analysis_artifact"]
|
||
a1, a2 = artifact_store.load(a1_artifact), artifact_store.load(a2_artifact)
|
||
|
||
# 结论：华东是主要来源，下钻到渠道是线上；总数闭合（headline 是 Q2 的两格）
|
||
assert a1["status"] == "ok" and a1["verdict"] == {"code": "driver", "subjects": [["华东"]], "direction": "down"}
|
||
assert "主要来自华东" in a1["phrase"] and a1["hypothesis"] == "下降是不是集中在某个地区"
|
||
assert {i["ref"] for i in a1["inputs"]} >= {"Q1", "Q2.r0.gmv", "Q2.r1.gmv"}
|
||
assert a2["verdict"]["code"] == "driver" and a2["verdict"]["subjects"] == [["线上"]]
|
||
# 下钻：算子按上一层结论的内容哈希认 parent，就是存下的那份工件
|
||
[parent] = [i for i in a2["inputs"] if i.get("role") == "parent"]
|
||
assert parent == {"ref": "A1", "artifact": a1_artifact, "role": "parent"}
|
||
assert content_hash(canonical_json(a1)) == a1_artifact
|
||
|
||
# 台账：两条 kind analysis，带 call_id，输入是那几份查询快照（下钻的另有上一层结论）
|
||
[finished] = await events(row.id, "node.finished", "bot")
|
||
analyses = [e for e in finished["evidence"] if e["kind"] == "analysis"]
|
||
assert [(e["call_id"], e["artifact"]) for e in analyses] == [("call_3", a1_artifact), ("call_4", a2_artifact)]
|
||
assert set(analyses[0]["inputs"]) == {q1, q2} and set(analyses[1]["inputs"]) == {q3, a1_artifact}
|
||
assert analyses[0] == contract.ledger_entry(a1, node_id="bot", exec_no=1, artifact=a1_artifact, call_id="call_3")
|
||
|
||
# 交回模型的：编号、判定语、关键数
|
||
replies = script.replies()
|
||
assert [r.split("\n", 1)[0] for r in replies] == ["【证据 Q1】", "【证据 Q2】", "【证据 Q3】", "【分析 A1】", "【分析 A2】"]
|
||
body = json.loads(replies[3].split("\n", 1)[1])
|
||
assert body["phrase"] == a1["phrase"] and body["verdict"] == "driver" and body["values"] == a1["values"]
|
||
assert body["table"]["rows"][0][0] == "华东"
|
||
|
||
# 按封存的工件复算，逐字段相同
|
||
recomputes(a1)
|
||
recomputes(a2)
|
||
|
||
|
||
async def test_replaying_after_approvals_does_not_recompute(monkeypatch):
|
||
counted = Counting(monkeypatch)
|
||
script = Script(monkeypatch, AGENT_CALLS)
|
||
run = await run_manager.start(graph=agent_graph(approval="always"), input_payload={})
|
||
for _ in AGENT_CALLS:
|
||
row = await wait(run.id, ("interrupted", "failed", "succeeded"))
|
||
assert row.status == "interrupted", row.error
|
||
await run_manager.resume(run.id, {"approved": True})
|
||
row = await wait(run.id, ("succeeded", "failed", "interrupted"))
|
||
assert row.status == "succeeded", row.error
|
||
|
||
assert counted.calls == 2, "重放时分析又算了一次"
|
||
ends = await events(row.id, "tool.end", "bot")
|
||
assert len(ends) == len(AGENT_CALLS)
|
||
a1_artifact, a2_artifact = ends[3]["analysis_artifact"], ends[4]["analysis_artifact"]
|
||
assert [r.split("\n", 1)[0] for r in script.replies()][-2:] == ["【分析 A1】", "【分析 A2】"]
|
||
[parent] = [i for i in artifact_store.load(a2_artifact)["inputs"] if i.get("role") == "parent"]
|
||
assert parent["artifact"] == a1_artifact
|
||
[finished] = await events(row.id, "node.finished", "bot")
|
||
assert [e["artifact"] for e in finished["evidence"] if e["kind"] == "analysis"] == [a1_artifact, a2_artifact]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 分析卡：波动基线 + 变化拆解
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_an_analysis_card_runs_the_real_operators():
|
||
items = [
|
||
{"id": "wow", "op": "baseline", "from": "weekly", "params": {"period": "week", "measure": "gmv", "label": "销售额"}},
|
||
{"id": "where", "op": "decompose", "mode": "additive", "from": "by_region",
|
||
"params": {"by": ["region"], **TWO_WEEKS, "label": "销售额",
|
||
"headline": {"base": "weekly.r10.gmv", "target": "weekly.r11.gmv"}}},
|
||
]
|
||
graph = chain(node("start", "input"), node("weekly", "tool", tool=TOOL, args={"sql": WEEKLY}),
|
||
node("by_region", "tool", tool=TOOL, args={"sql": BY_REGION}),
|
||
node("card", "analysis", items=items),
|
||
node("out", "output", fields=[{"name": "判定", "value": "{{ nodes.card.results.wow.phrase }}"}]))
|
||
run = await run_manager.start(graph=graph, input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
|
||
[finished] = await events(row.id, "node.finished", "card")
|
||
results = artifact_store.load(finished["artifact"])["results"]
|
||
wow, where = results["wow"], results["where"]
|
||
assert wow["status"] == "ok" and wow["verdict"] in ("outside_band", "far_outside_band") and wow["direction"] == "down"
|
||
assert where["status"] == "ok" and where["verdict"] == "driver" and "主要来自华东" in where["phrase"]
|
||
assert row.output["判定"] == wow["phrase"]
|
||
|
||
baseline = artifact_store.load(wow["artifact"])
|
||
# 本期完不完整按现查的表范围判（analysis.make_prober 按这次运行的数据源探 orders.day）：探到了，不再说核对不了
|
||
[probe] = baseline["probes"]
|
||
assert (probe["source"], probe["table"], probe["column"], probe["basis"]) == ("shop", "orders", "day", "live")
|
||
assert probe["max"] == "2026-05-24"
|
||
assert not [c for c in baseline["caveats"] if c["code"] == "comparability"]
|
||
decomposed = artifact_store.load(where["artifact"])
|
||
assert decomposed["verdict"]["subjects"] == [["华东"]]
|
||
entries = finished["evidence"]
|
||
assert [(e["kind"], e["item"], e["artifact"]) for e in entries] == [
|
||
("analysis", "wow", wow["artifact"]), ("analysis", "where", where["artifact"])]
|
||
recomputes(baseline)
|
||
recomputes(decomposed)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 口径卡的可比性挂钩：真 comparable
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def _week_sql(start: str, end: str) -> str:
|
||
return f"SELECT SUM(amount) AS gmv FROM orders WHERE day BETWEEN '{start}' AND '{end}'"
|
||
|
||
|
||
@pytest.mark.parametrize("this_week, verdict, has_reasons", [
|
||
(("2026-05-18", "2026-05-24"), "comparable", False),
|
||
(("2026-05-18", "2026-05-23"), "conditional", True), # 本期 6 天对上期 7 天：窗口长度、周末天数都不一样
|
||
])
|
||
async def test_a_caliber_card_checks_two_periods_with_the_real_operator(this_week, verdict, has_reasons):
|
||
metrics = [{"id": "wow", "name": "销售额环比", "decimals": 4,
|
||
"expression": "(cell(nodes.this_week, 0, 'gmv') - cell(nodes.last_week, 0, 'gmv'))"
|
||
" / cell(nodes.last_week, 0, 'gmv')"}]
|
||
graph = chain(node("start", "input"), node("this_week", "tool", tool=TOOL, args={"sql": _week_sql(*this_week)}),
|
||
node("last_week", "tool", tool=TOOL, args={"sql": _week_sql("2026-05-11", "2026-05-17")}),
|
||
node("card", "metrics", caliber="周报口径", metrics=metrics),
|
||
node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}]))
|
||
run = await run_manager.start(graph=graph, input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
[finished] = await events(row.id, "node.finished", "card")
|
||
[metric] = artifact_store.load(finished["artifact"])["metrics"]
|
||
found = metric["comparability"]
|
||
stored = artifact_store.load(found["artifact"])
|
||
assert stored["op"] == "comparable" and stored["params"]["base_query"] == "last_week"
|
||
assert stored["params"]["target_query"] == "this_week"
|
||
assert found == {"artifact": found["artifact"], "verdict": verdict, "phrase": stored["phrase"],
|
||
"reasons": analysis_ops.comparable_reasons(stored)}
|
||
assert bool(found["reasons"]) is has_reasons
|
||
# 整期按现查的表范围判
|
||
assert any(p.get("table") == "orders" and p.get("basis") == "live" for p in stored["probes"])
|
||
recomputes(stored)
|
||
assert not [e for e in await events(row.id, "log", "card") if e.get("code") == "metric_comparability_skipped"]
|