mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:01:58 +00:00
内置模板 ⑪「异动分析（可追溯）」：两个调用工具节点取数（整体按周 26 周、按地区两周，期次在 SQL 里按每周第一天的日期算出来、整体按周倒序排），分析卡固定做两项（本周的变化放回历史看的波动基线、按地区的可加变化拆解，总数取整体按周的第 0、1 行核对闭合；阈值不写死、取设置），报告撰写节点按受管级别写明最严的几项（含分析说法核对严格），出口按报告来自 + 单元格引用的契约出具。拿来就能按受管级别发布。界面文案白名单加一条：模板 SQL 里中文列别名按 SQLite 加的双引号是 SQL 语法。测试：画布校验、受管发布检查没有错误（模板 ⑨ 一并核）、接一份虚构的四地区 26 周零售库从头运行，两项都做成（明显超出常规波动、主要来自华东）、完整出具
This commit is contained in:
ThearchyHelios
@@ -338,6 +338,64 @@ TEMPLATES: list[dict[str, Any]] = [
|
||||
],
|
||||
"edges": [_e("start", "review"), _e("review", "approved", "approved"), _e("review", "rejected", "rejected")],
|
||||
},
|
||||
{
|
||||
# 分析能力一轮（docs/design/analysis-round/00-总方案.md §3.7）：模板定骨架——分析卡固定做哪几项分析写在图里，判定由
|
||||
# 系统按平台的方法和阈值算、系统写判定语，报告撰写节点只引用。照模板 ⑨ 的写法，拿来就能按受管级别发布
|
||||
"name": "⑪ 异动分析（可追溯）",
|
||||
"description": "周销售额的异动分析，每个判断都可追溯：两个「调用工具」节点取数（整体按周、按地区两周），分析卡固定做"
|
||||
"两项分析——这一周的变化在不在常规波动范围内、变化主要来自哪个地区（拆开后各段之和和总数核对闭合），报告撰写节点只引用"
|
||||
"判定语和数，判断词逐句核对。使用前请接入名为 shop 的数据源（含 orders 表的 order_date、region、amount 列），或上传"
|
||||
"同名表格。SQL 按 SQLite 的写法，按周汇总时期次取每周第一天（周一）的日期；接别的库时换成那个库的写法。报告撰写节点已按"
|
||||
"受管级别设置核对规则。",
|
||||
"tags": ["分析", "出具", "周报", "证据"],
|
||||
"nodes": [
|
||||
_n("start", "input", "周期", fields=[
|
||||
{"name": "week", "default": "2026-10-12", "description": "本周的第一天（周一）"},
|
||||
{"name": "prev_week", "default": "2026-10-05", "description": "上周的第一天（周一）"}]),
|
||||
# 期次在 SQL 里由日期列算出来（每周第一天的日期）：分析卡据此认出每一期覆盖哪几天，本期过没过完核对得了。
|
||||
# 倒序排：第 0 行是本周、第 1 行是上周，变化拆解按这两格核对闭合
|
||||
_n("weekly", "tool", "整体按周（26 周）", tool="db_query__shop", args={"sql": (
|
||||
"SELECT date(order_date, '-6 days', 'weekday 1') AS \"周\", SUM(amount) AS \"销售额\"\n"
|
||||
"FROM orders\n"
|
||||
"WHERE order_date >= date('{{ input.week }}', '-175 days')"
|
||||
" AND order_date < date('{{ input.week }}', '+7 days')\n"
|
||||
"GROUP BY date(order_date, '-6 days', 'weekday 1')\n"
|
||||
"ORDER BY \"周\" DESC")}),
|
||||
_n("by_region", "tool", "按地区两周", tool="db_query__shop", args={"sql": (
|
||||
"SELECT region AS \"地区\", date(order_date, '-6 days', 'weekday 1') AS \"周\", SUM(amount) AS \"销售额\"\n"
|
||||
"FROM orders\n"
|
||||
"WHERE order_date >= '{{ input.prev_week }}' AND order_date < date('{{ input.week }}', '+7 days')\n"
|
||||
"GROUP BY region, date(order_date, '-6 days', 'weekday 1')\n"
|
||||
"ORDER BY \"地区\", \"周\"")}),
|
||||
# 阈值不写在图里：取设置里的分析一组（发起时快照）；要固定阈值时在这张卡上写「阈值覆盖」
|
||||
_n("check", "analysis", "异动分析卡", items=[
|
||||
{"id": "wow", "op": "baseline", "from": "weekly",
|
||||
"params": {"period": "周", "measure": "销售额", "target": "{{ input.week }}", "grain": "week",
|
||||
"label": "销售额"},
|
||||
"hypothesis": "本周销售额的变化在不在常规波动范围内"},
|
||||
{"id": "where", "op": "decompose", "mode": "additive", "from": "by_region",
|
||||
"params": {"by": ["地区"], "period": "周", "base": "{{ input.prev_week }}", "target": "{{ input.week }}",
|
||||
"measure": "销售额", "headline": {"base": "weekly.r1.销售额", "target": "weekly.r0.销售额"},
|
||||
"label": "销售额"},
|
||||
"hypothesis": "本周销售额的变化主要来自哪个地区"},
|
||||
]),
|
||||
# 受管出具要求报告撰写节点写明最严的几项（发布门禁 G3，含分析说法核对设为严格），模板照着写
|
||||
_n("write", "report", "报告撰写", numbers="strict", on_violation="fail", claims="require_citation",
|
||||
analysis_claims="strict",
|
||||
system="你是经营分析撰写人：结论先行，判断照分析结论写，只写有数据支撑的内容。",
|
||||
instructions="为 {{ input.week }} 开始的这一周写一份销售额异动分析：先用一句话说本周销售额和上周比变了多少，"
|
||||
"再按分析卡的两项结论说这个变化在不在常规波动范围内、主要来自哪个地区，最后用一张表列出各地区的"
|
||||
"变化。判断照分析结论写、直接引用判定语，没有分析结论的判断不下；有一项没有做成的，照实写原因。"),
|
||||
# 图里没有口径卡：报告直接引用查询和分析结论的格，契约开「单元格引用」（受管级别要写明）
|
||||
_n("done", "output", "出具",
|
||||
fields=[{"name": "异动分析", "value": "{{ nodes.write.text }}"}],
|
||||
contract={"report_from": "write", "cells": True, "strict": True}),
|
||||
],
|
||||
"edges": [
|
||||
_e("start", "weekly"), _e("weekly", "by_region"), _e("by_region", "check"),
|
||||
_e("check", "write"), _e("write", "done"),
|
||||
],
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
"""内置模板「⑪ 异动分析（可追溯）」：取数（整体按周 26 周、按地区两周）→ 分析卡（波动基线 + 带总数的变化拆解）→ 报告撰写
|
||||
（分析说法严格核对）→ 出具（引用模式，单元格引用）。
|
||||
|
||||
模板既是示范也是活文档（同 tests/test_template_traceable.py 守模板 ⑨）：要通过画布校验、能直接按受管级别发布（发布检查
|
||||
没有错误），接上一份虚构的零售库 shop（orders(order_date, region, amount)：华东、华南、华北、西南四个地区，26 周每天一笔）
|
||||
要能从头到尾运行、两项分析都做成、完整出具。期次是每周第一天的日期，分析卡认得出每一期覆盖哪几天，本期过没过完核对得了。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import copy
|
||||
import sqlite3
|
||||
from datetime import date, timedelta
|
||||
|
||||
import pytest
|
||||
from langchain_core.messages import AIMessage
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import Run, RunEvent
|
||||
from app.engine.governance import lint_for_publish, publish_issues
|
||||
from app.engine.runner import run_manager
|
||||
from app.engine.schema import GraphSpec, validate_graph
|
||||
from app.seed import TEMPLATES
|
||||
|
||||
NAME = "⑪ 异动分析（可追溯）"
|
||||
REGIONS = {"华东": 400, "华南": 300, "华北": 200, "西南": 100}
|
||||
|
||||
|
||||
def template() -> dict:
|
||||
return next(t for t in TEMPLATES if t["name"] == NAME)
|
||||
|
||||
|
||||
def graph() -> dict:
|
||||
t = template()
|
||||
return copy.deepcopy({"nodes": t["nodes"], "edges": t["edges"]})
|
||||
|
||||
|
||||
def test_the_template_is_listed_once_after_the_others():
|
||||
names = [t["name"] for t in TEMPLATES]
|
||||
assert names.count(NAME) == 1 and names[-1] == NAME
|
||||
assert any(n.startswith("⑨ 可追溯周报") for n in names)
|
||||
|
||||
|
||||
def test_the_template_passes_validation():
|
||||
report = validate_graph(GraphSpec.model_validate(graph()))
|
||||
assert report.ok, [i.message for i in report.issues if i.level == "error"]
|
||||
types = {n["id"]: n["type"] for n in template()["nodes"]}
|
||||
assert list(types.values()).count("analysis") == 1 and "metrics" not in types.values()
|
||||
card = next(n for n in template()["nodes"] if n["type"] == "analysis")["data"]["config"]
|
||||
assert [(i["id"], i["op"], i.get("mode")) for i in card["items"]] == [("wow", "baseline", None),
|
||||
("where", "decompose", "additive")]
|
||||
assert card["items"][1]["params"]["headline"] == {"base": "weekly.r1.销售额", "target": "weekly.r0.销售额"}
|
||||
assert "overrides" not in card, "阈值取设置，模板不写死"
|
||||
|
||||
|
||||
def test_the_template_publishes_as_governed():
|
||||
"""拿来就能按受管级别发布：发布检查（和真正发布同一套）没有错误，证据门禁 G1–G5 和分析卡的受管规则都过。"""
|
||||
spec = GraphSpec.model_validate(graph())
|
||||
errors = [i for i in publish_issues(spec, level="governed") if i.level == "error"]
|
||||
assert errors == [], [i.message for i in errors]
|
||||
report = next(n for n in template()["nodes"] if n["type"] == "report")["data"]["config"]
|
||||
assert {k: report[k] for k in ("numbers", "on_violation", "claims", "analysis_claims")} == {
|
||||
"numbers": "strict", "on_violation": "fail", "claims": "require_citation", "analysis_claims": "strict"}
|
||||
out = next(n for n in template()["nodes"] if n["type"] == "output")["data"]["config"]
|
||||
assert out["contract"] == {"report_from": "write", "cells": True, "strict": True}
|
||||
assert out["fields"] == [{"name": "异动分析", "value": "{{ nodes.write.text }}"}]
|
||||
|
||||
|
||||
def test_template_nine_also_publishes_as_governed():
|
||||
"""模板 ⑨ 在第 1 波补了分析说法核对设为严格：同样直接按受管级别发布，没有错误。"""
|
||||
nine = next(t for t in TEMPLATES if t["name"].startswith("⑨"))
|
||||
spec = GraphSpec.model_validate(copy.deepcopy({"nodes": nine["nodes"], "edges": nine["edges"]}))
|
||||
errors = [i for i in publish_issues(spec, level="governed") if i.level == "error"]
|
||||
assert errors == [], [i.message for i in errors]
|
||||
assert not [i for i in lint_for_publish(spec, level="governed").issues if i.code == "governed.report_policy"]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 接一份虚构的零售库从头到尾运行
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _last_week() -> date:
|
||||
"""最近一个已经过完的整周的周一（运行当天往前，本周没过完）。"""
|
||||
today = date.today()
|
||||
return today - timedelta(days=today.weekday() + 7)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def shop(tmp_path, scratch_source):
|
||||
"""26 周、每个地区每天一笔：各周略有起伏；最后一周（最近一个过完的整周）华东掉了四成，别的地区照常。"""
|
||||
monday = _last_week()
|
||||
first = monday - timedelta(weeks=25)
|
||||
path = tmp_path / "shop.db"
|
||||
db = sqlite3.connect(path)
|
||||
db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, order_date TEXT, region TEXT, amount REAL);")
|
||||
rows, n = [], 0
|
||||
for week in range(26):
|
||||
for day in range(7):
|
||||
when = first + timedelta(weeks=week, days=day)
|
||||
for k, (region, base) in enumerate(REGIONS.items()):
|
||||
amount = base + (week * 7 + day * 3 + k * 5) % 11
|
||||
if week == 25 and region == "华东":
|
||||
amount = round(amount * 0.6, 2)
|
||||
n += 1
|
||||
rows.append((n, when.isoformat(), region, float(amount)))
|
||||
db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", rows)
|
||||
db.commit()
|
||||
db.close()
|
||||
await scratch_source("shop", path)
|
||||
return monday
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def engine_up():
|
||||
await run_manager.setup()
|
||||
yield
|
||||
await run_manager.shutdown()
|
||||
|
||||
|
||||
async def _wait(run_id: str) -> Run:
|
||||
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 ("succeeded", "failed") and row.manifest_seq is not None:
|
||||
return row
|
||||
raise AssertionError(f"run {run_id} 没有结束：{row.status}")
|
||||
|
||||
|
||||
async def test_the_template_runs_end_to_end_and_issues_formally(engine_up, shop, monkeypatch):
|
||||
from app.providers import mock_model
|
||||
|
||||
text = ("## 本周异动\n\n本周销售额 [[v:Q1.r0.销售额]]，上周 [[v:Q1.r1.销售额]]。[[see:Q1]]\n\n"
|
||||
"[[a:A1]]。[[see:A1]]\n\n[[a:A2]]。[[see:A2]]\n\n[[table:A2 cols=地区,base,target,delta]]")
|
||||
monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text))
|
||||
monday = shop
|
||||
run = await run_manager.start(graph=graph(), input_payload={
|
||||
"week": monday.isoformat(), "prev_week": (monday - timedelta(weeks=1)).isoformat()})
|
||||
row = await _wait(run.id)
|
||||
assert row.status == "succeeded", row.error
|
||||
async with SessionLocal() as session:
|
||||
ends = [e.data for e in (await session.execute(
|
||||
select(RunEvent).where(RunEvent.run_id == run.id, RunEvent.type == "analysis.end")
|
||||
.order_by(RunEvent.seq))).scalars()]
|
||||
by_item = {e["item"]: e for e in ends}
|
||||
# 两项都做成了：本周的变化放回 25 周的历史里看（期次认得出、本期过完了），华东主导了这次下降（各段之和和总数闭合）
|
||||
assert by_item["wow"]["status"] == "ok", by_item["wow"]
|
||||
assert by_item["wow"]["verdict"] in ("outside_band", "far_outside_band"), by_item["wow"]
|
||||
assert (by_item["where"]["status"], by_item["where"]["verdict"]) == ("ok", "driver"), by_item["where"]
|
||||
assert "主要来自华东" in by_item["where"]["phrase"]
|
||||
issuance = row.output["_issuance"]
|
||||
assert issuance["mode"] == "citations" and issuance["tier"] == "formal", issuance
|
||||
report = row.output["异动分析"]
|
||||
assert by_item["where"]["phrase"] in report and by_item["wow"]["phrase"] in report
|
||||
@@ -55,6 +55,12 @@
|
||||
"rule": "半角标点",
|
||||
"reason": "模板里「写代码」节点的系统提示（写给模型），print(json.dumps(...)) 是代码示例，半角括号和省略号是 Python 语法"
|
||||
},
|
||||
{
|
||||
"file": "backend/app/seed.py",
|
||||
"text": "AS \"周\"",
|
||||
"rule": "半角标点",
|
||||
"reason": "模板 ⑪「异动分析（可追溯）」两个取数节点的 SQL：结果列按节点手册的规矩起中文别名、SQLite 的写法加双引号（AS \"周\"），引号和逗号是 SQL 语法；中文列名让分析卡的判定语写「分散在各地区」、报告的表头读得懂"
|
||||
},
|
||||
{
|
||||
"file": "backend/app/api/copilot.py",
|
||||
"text": "报告按 [[i:字段名]]",
|
||||
|
||||
Reference in New Issue
Block a user