mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 22:51:58 +00:00
ThearchyHelios
263 lines
13 KiB
Python
263 lines
13 KiB
Python
"""口径卡钉住另一个已发布工作流某个版本里的口径卡（caliber_from），版本纳入升版处置。
|
||
|
||
用户拍板：口径卡的版本和子工作流一样纳入升版处置，不新造实体。
|
||
|
||
- caliber_from: {workflow_id, workflow_version, node_id}：运行时从那个 WorkflowVersion 的图里
|
||
取出口径卡的 caliber、caliber_version、metrics 定义，metric_set 里记下来源
|
||
- 引用的工作流、版本或节点不存在：节点失败，报错写清楚
|
||
- 上游有更新的版本时，正式运行前必须声明 upgrade_policy：没声明就挡住，声明了记
|
||
caliber.upgrade 事件（node_id、pinned、latest、policy、policy_label）
|
||
- 证据面板的指标步骤带上来源工作流和版本，以及升版处置
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
from langchain_core.messages import AIMessage
|
||
from sqlalchemy import select
|
||
|
||
from app.core import artifact_store
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import Run, RunEvent, Workflow, WorkflowVersion
|
||
from app.engine.evidence import iter_segments
|
||
from app.engine.runner import run_manager
|
||
from app.main import app
|
||
|
||
KPI = "{'gmv': 45678.5, 'orders': 1234}"
|
||
STANDARD = [
|
||
{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "format": "thousands", "expression": "vars.kpi.gmv"},
|
||
{"id": "orders", "name": "订单数", "unit": "单", "format": "thousands", "expression": "vars.kpi.orders"},
|
||
]
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
async def engine_up():
|
||
await run_manager.setup()
|
||
yield
|
||
await run_manager.shutdown()
|
||
|
||
|
||
@pytest.fixture
|
||
async def client():
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||
yield c
|
||
|
||
|
||
def node(nid, ntype, label=None, **config):
|
||
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or 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:])]}
|
||
|
||
|
||
def library(metrics=STANDARD, version="v3"):
|
||
"""上游：一张只放口径卡的工作流（口径库）。"""
|
||
return chain(node("start", "input"),
|
||
node("caliber", "metrics", "周报口径卡", caliber="周报口径", caliber_version=version, metrics=metrics),
|
||
node("out", "output", fields=[{"name": "r", "value": "{{ nodes.caliber.text }}"}]))
|
||
|
||
|
||
async def publish(graph, *, name="口径库", versions=1, status="published", later=None) -> str:
|
||
async with SessionLocal() as session:
|
||
wf = Workflow(name=name, graph=graph, status=status, published_version=versions, version=versions)
|
||
session.add(wf)
|
||
await session.flush()
|
||
for v in range(1, versions + 1):
|
||
# 每一版都发布过（level 记着）：口径卡固定的版本须「已发布且未退役」（D-q）
|
||
session.add(WorkflowVersion(workflow_id=wf.id, version=v, graph=later if later and v > 1 else graph,
|
||
level="published"))
|
||
await session.commit()
|
||
return wf.id
|
||
|
||
|
||
def weekly(pinned: dict, *, report=False, **card):
|
||
nodes = [
|
||
node("start", "input"),
|
||
node("fetch", "transform", mode="expression", expression=KPI, assign_to="kpi"),
|
||
node("card", "metrics", "本地口径卡", caliber_from=pinned, **card),
|
||
]
|
||
if report:
|
||
nodes.append(node("write", "report", instructions="写周报", on_violation="flag"))
|
||
nodes.append(node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]))
|
||
else:
|
||
nodes.append(node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}]))
|
||
return chain(*nodes)
|
||
|
||
|
||
async def wait(run_id: str) -> Run:
|
||
for _ in range(400):
|
||
await asyncio.sleep(0.05)
|
||
async with SessionLocal() as session:
|
||
row = await session.get(Run, run_id)
|
||
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
|
||
return row
|
||
raise AssertionError(f"run {run_id} 没跑完：{row.status}")
|
||
|
||
|
||
async def events(run_id: str, etype: str, node_id: str | None = None) -> list[RunEvent]:
|
||
async with SessionLocal() as session:
|
||
q = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == etype)
|
||
if node_id:
|
||
q = q.where(RunEvent.node_id == node_id)
|
||
return list((await session.execute(q.order_by(RunEvent.seq))).scalars())
|
||
|
||
|
||
async def explore(graph) -> Run:
|
||
return await wait((await run_manager.start(graph=graph, input_payload={})).id)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 取定义、记来源
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_definitions_come_from_the_pinned_version():
|
||
wf_id = await publish(library())
|
||
pinned = {"workflow_id": wf_id, "workflow_version": 1, "node_id": "caliber"}
|
||
row = await explore(weekly(pinned))
|
||
assert row.status == "succeeded", row.error
|
||
|
||
[finished] = await events(row.id, "node.finished", "card")
|
||
card = artifact_store.load(finished.data["artifact"])
|
||
assert (card["caliber"], card["caliber_version"]) == ("周报口径", "v3")
|
||
assert [(m["id"], m["value"], m["rendered"]) for m in card["metrics"]] == [
|
||
("gmv", 45678.5, "45,678.5元"), ("orders", 1234, "1,234单")]
|
||
assert card["source"] == pinned
|
||
# 落进工件库的指标集里也记着来源：证据接口从这里取
|
||
[entry] = finished.data["evidence"]
|
||
stored = artifact_store.load(entry["artifact"])
|
||
assert stored["source"] == pinned and entry["source"] == pinned
|
||
|
||
|
||
async def test_later_versions_do_not_leak_into_a_pinned_card():
|
||
"""钉住 v1 就用 v1 的定义：上游 v2 改了算法，这次运行照旧按 v1 算。"""
|
||
changed = [{**STANDARD[0], "expression": "vars.kpi.gmv * 2"}, STANDARD[1]]
|
||
wf_id = await publish(library(), versions=2, later=library(changed, version="v4"))
|
||
row = await explore(weekly({"workflow_id": wf_id, "workflow_version": 1, "node_id": "caliber"}))
|
||
[finished] = await events(row.id, "node.finished", "card")
|
||
card = artifact_store.load(finished.data["artifact"])
|
||
assert card["caliber_version"] == "v3" and card["metrics"][0]["value"] == 45678.5
|
||
|
||
|
||
async def test_graphs_without_caliber_from_are_unchanged():
|
||
row = await explore(chain(node("start", "input"),
|
||
node("fetch", "transform", mode="expression", expression=KPI, assign_to="kpi"),
|
||
node("card", "metrics", caliber="周报口径", metrics=STANDARD),
|
||
node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}])))
|
||
[finished] = await events(row.id, "node.finished", "card")
|
||
card = artifact_store.load(finished.data["artifact"])
|
||
assert "source" not in card and "source" not in finished.data["evidence"][0]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 引用不存在
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize("pin, words", [
|
||
(lambda wf: {"workflow_id": "no-such-workflow", "workflow_version": 1, "node_id": "caliber"},
|
||
["no-such-workflow", "不存在"]),
|
||
(lambda wf: {"workflow_id": wf, "workflow_version": 7, "node_id": "caliber"}, ["口径库", "v7"]),
|
||
(lambda wf: {"workflow_id": wf, "workflow_version": 1, "node_id": "nope"}, ["口径库", "v1", "nope"]),
|
||
(lambda wf: {"workflow_id": wf, "workflow_version": 1, "node_id": "out"}, ["不是口径卡"]),
|
||
])
|
||
async def test_missing_references_fail_the_node_clearly(pin, words):
|
||
wf_id = await publish(library())
|
||
row = await explore(weekly(pin(wf_id)))
|
||
assert row.status == "failed"
|
||
[failed] = await events(row.id, "node.failed", "card")
|
||
assert all(w in failed.data["error"] for w in words), failed.data["error"]
|
||
|
||
|
||
async def test_pinning_a_card_that_is_itself_pinned_is_refused():
|
||
base = await publish(library(), name="源头口径")
|
||
middle = await publish(chain(node("start", "input"),
|
||
node("caliber", "metrics", caliber_from={"workflow_id": base, "workflow_version": 1,
|
||
"node_id": "caliber"}),
|
||
node("out", "output", fields=[{"name": "r", "value": "x"}])), name="转手口径")
|
||
row = await explore(weekly({"workflow_id": middle, "workflow_version": 1, "node_id": "caliber"}))
|
||
[failed] = await events(row.id, "node.failed", "card")
|
||
assert "源头" in failed.data["error"] and "源头口径" in failed.data["error"], failed.data["error"]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 升版处置：正式运行
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def formal(client, graph):
|
||
wf_id = await publish(graph, name="周报")
|
||
return await client.post("/api/runs", json={"workflow_id": wf_id, "run_class": "formal"})
|
||
|
||
|
||
async def test_a_newer_upstream_blocks_formal_runs_without_a_policy(client):
|
||
lib = await publish(library(), versions=2, later=library(version="v4"))
|
||
res = await formal(client, weekly({"workflow_id": lib, "workflow_version": 1, "node_id": "caliber"}))
|
||
assert res.status_code == 409, res.text
|
||
detail = res.json()["detail"]
|
||
assert "「本地口径卡」" in detail and "v1" in detail and "v2" in detail and "「上游发布新版本时」" in detail
|
||
|
||
|
||
async def test_a_declared_policy_is_recorded(client):
|
||
lib = await publish(library(), versions=2, later=library(version="v4"))
|
||
res = await formal(client, weekly({"workflow_id": lib, "workflow_version": 1, "node_id": "caliber"},
|
||
upgrade_policy="dual"))
|
||
assert res.status_code == 201, res.text
|
||
row = await wait(res.json()["id"])
|
||
assert row.status == "succeeded", row.error
|
||
[event] = await events(row.id, "caliber.upgrade")
|
||
assert event.node_id == "card"
|
||
assert {k: event.data.get(k) for k in ("workflow_id", "pinned", "latest", "policy", "policy_label",
|
||
"caliber_node")} == {
|
||
"workflow_id": lib, "pinned": 1, "latest": 2, "policy": "dual", "policy_label": "新旧口径并列展示",
|
||
"caliber_node": "caliber"}
|
||
|
||
|
||
async def test_pinned_to_the_latest_needs_no_policy(client):
|
||
lib = await publish(library())
|
||
res = await formal(client, weekly({"workflow_id": lib, "workflow_version": 1, "node_id": "caliber"}))
|
||
assert res.status_code == 201, res.text
|
||
row = await wait(res.json()["id"])
|
||
assert await events(row.id, "caliber.upgrade") == []
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 证据面板的指标步骤
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_metric_step_shows_the_source_and_the_upgrade(client, monkeypatch):
|
||
from app.providers import mock_model
|
||
|
||
monkeypatch.setattr(mock_model.MockChatModel, "_decide",
|
||
lambda self, messages: AIMessage(content="销售额 [[m:gmv]]。"))
|
||
lib = await publish(library(), versions=2, later=library(version="v4"))
|
||
pinned = {"workflow_id": lib, "workflow_version": 1, "node_id": "caliber"}
|
||
res = await formal(client, weekly(pinned, report=True, upgrade_policy="incomparable"))
|
||
row = await wait(res.json()["id"])
|
||
assert row.status == "succeeded", row.error
|
||
doc_id = row.output["_evidence"]["doc_artifact"]
|
||
doc = artifact_store.load(doc_id)
|
||
seg = next(s for s in iter_segments(doc) if s["text"] == "45,678.5元")
|
||
body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}")).json()
|
||
metric = body["chain"][0]
|
||
assert metric["source"] == {**pinned, "workflow_name": "口径库"}
|
||
assert (metric["caliber"], metric["version"]) == ("周报口径", "v3")
|
||
assert metric["caliber_upgrade"] == {"workflow_id": lib, "caliber_node": "caliber", "pinned": 1, "latest": 2,
|
||
"policy": "incomparable", "policy_label": "标注与历史不可比"}
|
||
assert json.dumps(metric, ensure_ascii=False)
|
||
|
||
|
||
async def test_subgraphs_keep_the_same_upgrade_rule(client):
|
||
"""子工作流的升版处置原样不变：口径卡是并进同一套规则，不是另起一套。"""
|
||
lib = await publish(library(), versions=2, later=library(version="v4"))
|
||
graph = chain(node("start", "input"),
|
||
node("sub", "subgraph", "方法卡", workflow_id=lib, workflow_version=1),
|
||
node("out", "output", fields=[{"name": "r", "value": "{{ nodes.sub.text }}"}]))
|
||
res = await formal(client, graph)
|
||
assert res.status_code == 409 and "「方法卡」（子工作流）固定在 v1，但上游已有 v2" in res.json()["detail"]
|