mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 18:01:58 +00:00
ThearchyHelios
538799e70d
为什么：同一类问题被反复问、Agent 反复写出同一个形状的 SQL，就该沉淀成一份口径；以前 from-run 保留 Agent，沉淀出的模板没有固定 SQL，以后永远比对不上。第 2 波的统计在 E2 反馈、E3 认证队列上照实写了「目前没有数据」，这一波补上能补的。
做了什么（新增 ledger/certify.py）：
- GET /api/ledger/certify-queue（搭建者）：确定性重复（问数据运行里被报告引用、没对上口径的查询，按数据源 + normalize_sql 文本分组，≥3 次运行且来自 ≥2 段对话；跨运行变化的字面量是参数候选，展示的形状不带取值）、问题聚类（app/chat/questions 全站唯一的那份）、被反复用的答案（导出 ≥2 次用第 2 波的 report.exported 审计、做成过应用看蓝图来源、在画布里打开或提取过看工作流 origin）。每项给代表问题、轮次、相关口径、是否已固化。
- POST /api/ledger/certify-queue/dismiss：「不是口径」写审计 ledger.certify.dismiss，之后不再进队。
- 固化成口径草稿（from-run 加 freeze、peers 后转调 certify.freeze）：报告引用的查询变成固定 SQL 的取数节点，跨轮变化的字面量换成 '{{ input.pN }}'（缺省值取代表运行的），单行结果的数值列成口径卡指标，报告核对规则按受管要求写、契约 report_from + metrics_from + required（多行加 cells）；过 validate_graph 和受管门禁。新工作流 origin.sources 记来源问题。
- 发布之后补比：api/workflows.commit_publish 提交后调 certify.after_publish，来源运行按新口径补比一次（新的 trigger=certified，界面写「按之后发布的口径回看」，时间线和账本接口同一句）。
- 统计：E2 加对过账的答案数、参数经人核对的、指标由人选的次数（反馈照实写第 4 波）；E3 认证队列的转化（固化、发布、不是口径）。
测试：tests/test_ledger_certify.py（参数推断与模板、三种信号进队、门槛、不是口径、固化草稿过受管门禁、发布后来源运行补比成等价已证明并出队、再问标等价已证明、统计）。
225 lines
12 KiB
Python
225 lines
12 KiB
Python
"""沉淀：可沉淀的问题（认证队列）和「固化成口径草稿」（next-round 02 做法 8，app/ledger/certify.py）。
|
||
|
||
真的跑三次问数据（两段对话、同一个 SQL 形状、日期不同，虚构店铺库），等后台比对做完：
|
||
- 队列：确定性重复进队（3 次运行、2 段对话），变了的字面量是参数候选、不变的是常量，形状里不带取值；同一个问题问了 3 次
|
||
进问题聚类；导出过两次的、做成过应用的答案进「被反复用的答案」；点「不是口径」写审计后不再进队；
|
||
- 固化：报告引用的查询变成固定 SQL 的取数节点，跨轮变化的日期成为入口参数（缺省值取代表运行的），常量 'paid' 照写；单行
|
||
结果的数值列成口径卡指标；出口契约 report_from + metrics_from；过 validate_graph 和受管级别的发布门禁；origin.sources 记
|
||
三个来源问题；
|
||
- 按受管级别发布之后：来源运行在后台按它补比（trigger=certified），查询变成等价已证明、出了队；再问一次同样的形状，标
|
||
「等价已证明」；
|
||
- 统计 E3：认证队列的转化（固化、发布、不是口径）有数了。
|
||
数据是虚构店铺库。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import Conversation, ConversationTurn, Run, Workflow
|
||
from app.engine.runner import run_manager
|
||
from app.engine.schema import GraphSpec, validate_graph
|
||
from app.ledger import certify
|
||
from app.ledger import match as ledger_match
|
||
from app.ledger import store
|
||
from app.main import app
|
||
from tests import ledger_samples as S
|
||
|
||
QUESTION = "九月第一周卖了多少"
|
||
WEEKS = [("2026-09-01", "2026-09-08"), ("2026-09-02", "2026-09-09"), ("2026-09-03", "2026-09-10")]
|
||
|
||
|
||
def weekly(start: str, end: str) -> str:
|
||
return (f"SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '{start}' "
|
||
f"AND paid_at < '{end}'")
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
async def engine_up(tmp_path):
|
||
await run_manager.setup()
|
||
await S.fresh_source(tmp_path / "shop.db")
|
||
yield
|
||
await ledger_match.drain()
|
||
await run_manager.shutdown()
|
||
|
||
|
||
@pytest.fixture
|
||
async def client():
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||
yield c
|
||
|
||
|
||
async def conversation() -> str:
|
||
async with SessionLocal() as session:
|
||
conv = Conversation(title="问销售", kind="chat")
|
||
session.add(conv)
|
||
await session.commit()
|
||
return conv.id
|
||
|
||
|
||
async def ask(monkeypatch, conversation_id: str, sql: str, question: str = QUESTION) -> Run:
|
||
"""问数据的一轮：建轮次、跑一次问数据运行（来源记着对话和轮次）、挂回轮次、等比对做完。"""
|
||
async with SessionLocal() as session:
|
||
turn = ConversationTurn(conversation_id=conversation_id, question=question)
|
||
session.add(turn)
|
||
await session.commit()
|
||
turn_id = turn.id
|
||
S.script(monkeypatch, "这一周销售额 [[v:Q1.r0.gmv]]。")
|
||
run = await run_manager.start(graph=S.explore_graph(sql), input_payload={"question": question}, origin_kind="chat",
|
||
origin={"conversation_id": conversation_id, "turn_id": turn_id})
|
||
row = await S.wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
async with SessionLocal() as session:
|
||
(await session.get(ConversationTurn, turn_id)).run_id = row.id
|
||
await session.commit()
|
||
await ledger_match.drain()
|
||
return row
|
||
|
||
|
||
async def three_asks(monkeypatch) -> list[Run]:
|
||
a, b = await conversation(), await conversation()
|
||
return [await ask(monkeypatch, conv, weekly(*week)) for conv, week in zip((a, a, b), WEEKS)]
|
||
|
||
|
||
def shape_items(body: dict) -> list[dict]:
|
||
return [i for i in body["items"] if i["kind"] == "shape" and i["source"] == S.CURRENT["source"]]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 队列
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def test_varying_literals_are_parameters_and_the_template_puts_them_back():
|
||
shape = certify.Shape("shop", "SELECT SUM(amount) AS gmv FROM orders WHERE status = ? AND paid_at >= ? AND x = ?",
|
||
("'paid'", "'2026-09-01'", "3"))
|
||
assert certify.varying([shape.literals, ("'paid'", "'2026-09-02'", "4")]) == [1, 2]
|
||
template, fields = certify.template_of(shape, [1, 2], "sqlite", start=3)
|
||
assert template == ("SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '{{ input.p3 }}' "
|
||
"AND x = {{ input.p4 }}")
|
||
assert fields == [{"name": "p3", "type": "string", "required": True, "default": "2026-09-01"},
|
||
{"name": "p4", "type": "number", "required": True, "default": 3}]
|
||
|
||
|
||
async def test_repeated_shapes_questions_and_reused_answers_queue_up(client, monkeypatch):
|
||
runs = await three_asks(monkeypatch)
|
||
body = (await client.get("/api/ledger/certify-queue")).json()
|
||
[item] = shape_items(body)
|
||
assert item["runs"] == 3 and item["conversations"] == 2 and item["question"] == QUESTION
|
||
assert item["params"] == [{"index": 1, "distinct": 3}, {"index": 2, "distinct": 3}] and item["constants"] == 1
|
||
assert "2026-09" not in item["shape"] and "'paid'" not in item["shape"] and "?" in item["shape"]
|
||
assert {t["run_id"] for t in item["turns"]} == {r.id for r in runs}
|
||
assert all(t["turn_id"] and t["conversation_id"] for t in item["turns"])
|
||
assert item["representative"]["run_id"] in {r.id for r in runs} and len(item["peers"]) == 2
|
||
assert item["related"] == [] and item["frozen_into"] == []
|
||
clusters = [i for i in body["items"] if i["kind"] == "cluster" and i["question"] == QUESTION]
|
||
assert clusters and clusters[0]["count"] >= 3
|
||
|
||
# 被反复用的答案：导出过两次的、做成过应用的
|
||
from app.db.models_blueprints import Blueprint
|
||
|
||
async with SessionLocal() as session:
|
||
from app.auth.audit import record
|
||
|
||
for _ in range(2):
|
||
await record(session, "report.exported", "run", runs[0].id, data={"format": "html"})
|
||
session.add(Blueprint(name="周销售应用", source={"kind": "run", "id": runs[1].id}, spec={}))
|
||
await session.commit()
|
||
body = (await client.get("/api/ledger/certify-queue")).json()
|
||
reused = {i["representative"]["run_id"]: i for i in body["items"] if i["kind"] == "reused"}
|
||
assert reused[runs[0].id]["uses"]["exports"] == 2 and reused[runs[1].id]["uses"]["app"] is True
|
||
assert runs[2].id not in reused
|
||
|
||
# 「不是口径」：记审计，之后不再进队
|
||
res = await client.post("/api/ledger/certify-queue/dismiss", json={"item_id": item["id"], "kind": "shape"})
|
||
assert res.status_code == 200
|
||
assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == []
|
||
stats = (await client.get("/api/ledger/stats")).json()["e3"]["certify"]
|
||
assert stats["available"] is True and stats["dismissed"] >= 1
|
||
|
||
|
||
async def test_two_runs_or_one_conversation_are_not_enough(client, monkeypatch):
|
||
conv = await conversation()
|
||
for week in WEEKS:
|
||
await ask(monkeypatch, conv, weekly(*week))
|
||
assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == [], "只来自一段对话"
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 固化、发布、再问
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_freezing_makes_a_fixed_sql_draft_that_passes_the_governed_gate(client, monkeypatch):
|
||
from app.engine.governance import publish_issues
|
||
|
||
runs = await three_asks(monkeypatch)
|
||
[item] = shape_items((await client.get("/api/ledger/certify-queue")).json())
|
||
rep = item["representative"]["run_id"]
|
||
res = await client.post("/api/copilot/from-run", json={"run_id": rep, "freeze": True, "peers": item["peers"]})
|
||
assert res.status_code == 200, res.text
|
||
made = res.json()
|
||
assert made["frozen"] is True and made["params"] == ["p1", "p2"] and made["metrics"] == ["m1"]
|
||
async with SessionLocal() as session:
|
||
wf = await session.get(Workflow, made["workflow_id"])
|
||
assert wf.status == "draft" and "certified" in wf.tags
|
||
nodes = {n["id"]: n for n in wf.graph["nodes"]}
|
||
assert set(nodes) == {"start", "fetch", "card", "write", "out"}
|
||
rep_run = next(r for r in runs if r.id == rep)
|
||
start, end = WEEKS[[r.id for r in runs].index(rep)]
|
||
assert nodes["start"]["data"]["config"]["fields"] == [
|
||
{"name": "p1", "type": "string", "required": True, "default": start},
|
||
{"name": "p2", "type": "string", "required": True, "default": end}]
|
||
sql = nodes["fetch"]["data"]["config"]["args"]["sql"]
|
||
assert "'{{ input.p1 }}'" in sql and "'{{ input.p2 }}'" in sql and "'paid'" in sql
|
||
assert nodes["fetch"]["data"]["config"]["tool"] == S.tool()
|
||
assert nodes["card"]["data"]["config"]["metrics"] == [
|
||
{"id": "m1", "name": "gmv", "expression": "cell(nodes.fetch, 0, 'gmv')"}]
|
||
assert nodes["out"]["data"]["config"]["contract"] == {"report_from": "write", "metrics_from": "card",
|
||
"required": ["m1"]}
|
||
assert {k: nodes["write"]["data"]["config"].get(k) for k in ("numbers", "on_violation", "claims")} == {
|
||
"numbers": "strict", "on_violation": "fail", "claims": "require_citation"}
|
||
assert nodes["write"]["data"]["config"]["instructions"] == f"回答：{QUESTION}"
|
||
spec = GraphSpec.model_validate(wf.graph)
|
||
assert not [i for i in validate_graph(spec).issues if i.level == "error"]
|
||
assert not [i for i in publish_issues(spec, level="governed") if i.level == "error"], \
|
||
[i.message for i in publish_issues(spec, level="governed") if i.level == "error"]
|
||
assert [s["run_id"] for s in wf.origin["sources"]][0] == rep and len(wf.origin["sources"]) == 3
|
||
assert all(s["question"] == QUESTION for s in wf.origin["sources"])
|
||
assert wf.origin["run_id"] == rep_run.id and wf.origin["freeze"] is True
|
||
|
||
# 队列上标着已经固化过
|
||
[again] = shape_items((await client.get("/api/ledger/certify-queue")).json())
|
||
assert {w["workflow_id"] for w in again["frozen_into"]} == {wf.id}
|
||
|
||
# 按受管级别发布：来源运行在后台补比，变成等价已证明、出了队
|
||
res = await client.post(f"/api/workflows/{wf.id}/publish", json={"level": "governed"})
|
||
assert res.status_code == 200, res.text
|
||
await ledger_match.drain()
|
||
for run in runs:
|
||
recorded = await store.latest(run.id)
|
||
assert recorded.trigger == "certified" and recorded.retro is True
|
||
[query] = recorded.content["queries"]
|
||
assert query["status"] == "proven" and query["candidates"][0]["caliber"] == f"{wf.id}@1"
|
||
body = (await client.get(f"/api/runs/{runs[0].id}/ledger")).json()
|
||
assert body["retro_note"] == "按之后发布的口径回看"
|
||
assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == []
|
||
|
||
# 再问一次同样的形状：标等价已证明
|
||
later = await ask(monkeypatch, await conversation(), weekly("2026-09-05", "2026-09-12"))
|
||
[label] = (await store.latest(later.id)).content["reports"]["write"]["labels"].values()
|
||
assert label["kind"] == "caliber" and label["how"] == "independent" and label["caliber"] == f"{wf.id}@1"
|
||
assert label["binding"] == {"p1": "2026-09-05", "p2": "2026-09-12"}
|
||
|
||
stats = (await client.get("/api/ledger/stats")).json()["e3"]["certify"]
|
||
assert stats["frozen"] >= 1 and stats["published"] >= 1
|
||
|
||
|
||
async def test_freezing_needs_a_matched_successful_run(client, monkeypatch):
|
||
res = await client.post("/api/copilot/from-run", json={"run_id": "nope", "freeze": True})
|
||
assert res.status_code == 404
|
||
row = await S.explore(monkeypatch, S.ELSEWHERE, text="一共几家店，没有引用。")
|
||
res = await client.post("/api/copilot/from-run", json={"run_id": row.id, "freeze": True})
|
||
assert res.status_code == 409 and res.json()["detail"] == certify.NO_QUERIES
|