mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:21:58 +00:00
固化成口径草稿：口径卡用传入的名字、报告指令不写死代表那一轮的日期、指标名取目录中文名、已固化去重
验收报告里 certify.freeze 的四处：
- 口径卡的 caliber 用的是代表问题的原文，没用传入的 name。改成 name（去首尾空白）优先，没传才用代表问题。
- 报告的 instructions 写死成「回答：<代表问题>」，问题里带着那一轮的日期（「上周（10月5日到11日）」「九月第一周」），
参数化之后这句就错了。有入口参数时改写运行时渲染的参数（「这次的参数：p1 = {{ input.p1 }}，…；时间范围、筛选的
取值以参数为准」），不写问题原文；没有参数时取数的字面量都是口径的常量，照旧写问题原文。
- 指标名直接用列名。现在能从数据目录取到中文名的用中文名：先看查询快照冻结的目录（schema_artifact，和口径比对用的
同一份，ledger.match._frozen），老快照没冻结的看数据源现在的目录；只认列名本身就是目录里的一列（SELECT amount、
SUM(amount) AS amount），gmv 这类算出来起的别名取不到，照旧用列名。
- 认证队列里 cluster 项的 frozen_into 把同一个工作流列了好几次：固化时同组的几次运行都记进 origin.sources，一项里
的几次运行就各指一遍同一个工作流。shape 项同样会重。加 _frozen_of 按工作流去重（保持先后），三种项都用它。
测试（test_ledger_certify）：原来的固化用例改成断言指令里没有代表问题和日期、有 {{ input.p1 }} {{ input.p2 }}，
没传名字时口径卡照旧用代表问题，shape 和 cluster 两项的 frozen_into 都只有一个工作流；新增一条：传了名字的口径卡
用名字，目录里有中文名的列（amount → 金额）用中文名、取不到的照旧（n），没有参数时指令照旧写问题原文。去掉修复
回跑两条都挂。
This commit is contained in:
ThearchyHelios
@@ -11,13 +11,16 @@
|
||||
个问题的类；
|
||||
- 被反复用的答案（reused）：导出过 ≥ 2 次（report.exported 审计，第 2 波）、做成过应用（蓝图的来源是这次运行）、在画布里
|
||||
打开过或提取过（工作流的 origin.run_id 是这次运行）的轮次。
|
||||
每项给代表问题、轮次链接、（shape 的）SQL 形状和参数候选、有没有相关口径（可以对账的那份）、是不是已经固化过。点「不是口径」
|
||||
每项给代表问题、轮次链接、（shape 的）SQL 形状和参数候选、有没有相关口径（可以对账的那份）、是不是已经固化过（frozen_into，
|
||||
同一个工作流只列一次）。点「不是口径」
|
||||
写一条审计（ledger.certify.dismiss，subject 是这一项的 id），之后不再进队。门槛同 ledger/stats（E3 的重复度）。
|
||||
|
||||
**固化成口径草稿**（freeze，from-run 加 freeze 选项后转调它）：代表运行里报告引用了的查询，各变成一个调用工具节点——SQL 用
|
||||
规范化后的文本，跨轮变化的字面量换成 '{{ input.p1 }}'（同组的别的运行 peers 用来认哪些位置变了），入口字段的缺省值取
|
||||
代表运行里的值；只有一行的结果每个数值列建一项口径卡指标（cell(nodes.取数, 0, '列')）；报告节点保留，出口契约写
|
||||
report_from + metrics_from（有多行结果的另加 cells: true），过 validate_graph。新工作流记 origin.sources：
|
||||
代表运行里的值；只有一行的结果每个数值列建一项口径卡指标（cell(nodes.取数, 0, '列')，指标名能从数据目录取到中文名的
|
||||
用中文名，取不到用列名）；口径卡的名字用传入的 name，没传用代表问题；报告节点保留，指令在有入口参数时不写代表那一轮的问题
|
||||
原文（里面的日期参数化之后就不对了），改写运行时渲染的参数，没有参数时照旧写问题原文；出口契约写 report_from +
|
||||
metrics_from（有多行结果的另加 cells: true），过 validate_graph。新工作流记 origin.sources：
|
||||
[{run_id, conversation_id, turn_id, question}]，这就是「口径 → 来源问题」的记录。之后由人在画布上审改、按受管级别发布——
|
||||
提取器是作者，不是发布者。
|
||||
|
||||
@@ -64,6 +67,9 @@ DRAFT_NAME = "{question}（固化的口径草稿）"
|
||||
DRAFT_DESCRIPTION = ("从问数据的回答固化：报告引用了的 {n} 条查询变成固定的取数，跨轮变化的 {p} 个字面量成为入口参数。"
|
||||
"经人工审核、按受管级别发布后，再问同样的问题就能认出等价、按口径出具")
|
||||
REPORT_INSTRUCTIONS = "回答：{question}"
|
||||
#: 有入口参数时的报告指令：不写代表那一轮的问题原文——里面的日期（「上周（10月5日到11日）」「九月第一周」）参数化之后
|
||||
#: 就不对了；改写这次运行时渲染出来的参数。{params} 是「p1 = {{ input.p1 }}，…」
|
||||
REPORT_INSTRUCTIONS_PARAMS = "按取到的数回答这个口径的问题（这次的参数：{params}；时间范围、筛选的取值以参数为准）"
|
||||
|
||||
|
||||
class CertifyRefused(Exception):
|
||||
@@ -228,6 +234,16 @@ async def _frozen(session: Any) -> dict[str, list[dict[str, Any]]]:
|
||||
return out
|
||||
|
||||
|
||||
def _frozen_of(run_ids: Sequence[str | None], frozen: dict[str, list[dict[str, Any]]]) -> list[dict[str, Any]]:
|
||||
"""这几次运行固化进了哪些工作流：同一个工作流只列一次（固化时同组的运行都记进 origin.sources，一项里的几次运行常常
|
||||
指着同一个工作流），按先后。"""
|
||||
out: dict[str, dict[str, Any]] = {}
|
||||
for run_id in run_ids:
|
||||
for w in frozen.get(run_id or "", []):
|
||||
out.setdefault(str(w["workflow_id"]), w)
|
||||
return list(out.values())
|
||||
|
||||
|
||||
def _turn_view(member: Any) -> dict[str, Any]:
|
||||
return {"run_id": member.run_id, "conversation_id": member.conversation_id, "turn_id": member.turn_id,
|
||||
"question": member.question}
|
||||
@@ -308,7 +324,7 @@ async def queue(session: Any, *, principal: Any) -> dict[str, Any]:
|
||||
"constants": len(rep.literals) - len(params), "related": related,
|
||||
"representative": {"run_id": rep.run_id, "query": rep.artifact},
|
||||
"peers": [m.run_id for m in once[1:]],
|
||||
"frozen_into": [w for m in once for w in frozen.get(m.run_id, [])],
|
||||
"frozen_into": _frozen_of([m.run_id for m in once], frozen),
|
||||
})
|
||||
# 问题聚类
|
||||
questions = await recent_questions(session, principal, limit=QUESTION_LIMIT)
|
||||
@@ -322,7 +338,7 @@ async def queue(session: Any, *, principal: Any) -> dict[str, Any]:
|
||||
items.append({"id": iid, "kind": "cluster", "question": seed.text, "count": cluster.count,
|
||||
"turns": [{"run_id": q.run_id, "turn_id": q.turn_id, "question": q.text}
|
||||
for q in cluster.members[:TURNS_LIMIT]],
|
||||
"frozen_into": [w for q in cluster.members if q.run_id for w in frozen.get(q.run_id, [])]})
|
||||
"frozen_into": _frozen_of([q.run_id for q in cluster.members], frozen)})
|
||||
items += await _reused(session, runs, turns, dismissed, frozen)
|
||||
return {"items": items, "considered": {"runs": len(runs), "matched": len(latest), "questions": len(questions)},
|
||||
"threshold": {"runs": REPEAT_RUNS, "conversations": REPEAT_CONVERSATIONS, "exports": EXPORTS},
|
||||
@@ -367,7 +383,7 @@ async def _reused(session: Any, runs: list[Run], turns: dict[str, dict[str, Any]
|
||||
out.append({"id": iid, "kind": "reused", "question": info.get("question") or "", "uses": uses,
|
||||
"turns": [{"run_id": run.id, "conversation_id": info.get("conversation_id"),
|
||||
"turn_id": info.get("turn_id"), "question": info.get("question") or ""}],
|
||||
"representative": {"run_id": run.id}, "frozen_into": frozen.get(run.id, [])})
|
||||
"representative": {"run_id": run.id}, "frozen_into": _frozen_of([run.id], frozen)})
|
||||
return out
|
||||
|
||||
|
||||
@@ -385,6 +401,24 @@ async def dismiss(session: Any, iid: str, *, kind: str | None = None, note: str
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
async def _labels(session: Any, snap: dict[str, Any], source: str) -> Any:
|
||||
"""结果列的中文名：先看查询当时冻结的数据目录（快照的 schema_artifact，同口径比对），没冻结的（老快照）看数据源现在
|
||||
的目录。只认列名本身就是目录里的一列（SELECT amount、SUM(amount) AS amount）；算出来起的别名（gmv）取不到，调用方
|
||||
照旧用列名。返回 _labeler：(表, 列) → 中文名或 None。"""
|
||||
from app.data import catalog
|
||||
from app.data.sql_semantics import _labeler, catalog_labels
|
||||
from app.db.models import DataSource
|
||||
from app.ledger.match import _frozen as frozen_schema
|
||||
|
||||
_, names = frozen_schema(snap.get("schema_artifact"))
|
||||
if not names:
|
||||
source_id = (await session.execute(select(DataSource.id).where(DataSource.name == source))).scalar()
|
||||
if source_id:
|
||||
entries = await catalog.read_catalog(session, source_id)
|
||||
names = catalog_labels({table: entry.notes for table, entry in entries.items()})
|
||||
return _labeler(names)
|
||||
|
||||
|
||||
async def freeze(session: Any, run_id: str, *, peers: Sequence[str] = (), name: str = "",
|
||||
conversation_id: str | None = None, turn_id: str | None = None) -> dict[str, Any]:
|
||||
"""把一次问数据运行固化成口径草稿（见模块说明），落库成新的草稿工作流，返回 {workflow_id, name, nodes, edges, params,
|
||||
@@ -460,9 +494,10 @@ async def freeze(session: Any, run_id: str, *, peers: Sequence[str] = (), name:
|
||||
rows = snap.get("rows") if isinstance(snap.get("rows"), list) else []
|
||||
columns = [str(c) for c in snap.get("columns") or []]
|
||||
if len(rows) == 1 and isinstance(rows[0], (list, tuple)):
|
||||
label = await _labels(session, snap, shape.source)
|
||||
for col, value in zip(columns, rows[0]):
|
||||
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
||||
metrics.append({"id": f"m{len(metrics) + 1}", "name": col,
|
||||
metrics.append({"id": f"m{len(metrics) + 1}", "name": label((None, col)) or col,
|
||||
"expression": f"cell(nodes.{nid(i)}, 0, {col!r})"})
|
||||
else:
|
||||
multi = True
|
||||
@@ -473,15 +508,20 @@ async def freeze(session: Any, run_id: str, *, peers: Sequence[str] = (), name:
|
||||
found = next((n for n in (run.graph or {}).get("nodes") or [] if n.get("id") == reports[0].node_id), None)
|
||||
cfg = ((found or {}).get("data") or {}).get("config") or {}
|
||||
report_cfg = {k: cfg[k] for k in ("provider", "model", "entities") if k in cfg}
|
||||
report_cfg["instructions"] = REPORT_INSTRUCTIONS.format(question=question or "（问题）")
|
||||
if fields:
|
||||
report_cfg["instructions"] = REPORT_INSTRUCTIONS_PARAMS.format(
|
||||
params="，".join(f"{f['name']} = {{{{ input.{f['name']} }}}}" for f in fields))
|
||||
else:
|
||||
# 没有入口参数：取数的字面量都是口径的常量，问题原文里的日期和它们对得上
|
||||
report_cfg["instructions"] = REPORT_INSTRUCTIONS.format(question=question or "（问题）")
|
||||
# 报告的核对规则一律按受管级别的要求写（engine/governance.REPORT_POLICY，和发布门禁同一份）：草稿要能按受管发布
|
||||
report_cfg.update({key: value for key, value, _ in REPORT_POLICY})
|
||||
contract: dict[str, Any] = {"report_from": "write"}
|
||||
tail: list[dict[str, Any]] = []
|
||||
if metrics:
|
||||
tail.append({"id": "card", "type": "metrics", "position": {"x": 0, "y": 0},
|
||||
"data": {"label": "口径卡", "config": {"caliber": (question or "口径")[:40], "caliber_version": "v1",
|
||||
"metrics": metrics}}})
|
||||
"data": {"label": "口径卡", "config": {"caliber": (name.strip() or question or "口径")[:40],
|
||||
"caliber_version": "v1", "metrics": metrics}}})
|
||||
report_cfg["metrics_from"] = ["card"]
|
||||
# 受管出具要写必需指标（不然「不予出具」这一档触发不了）：固化出的每个指标都是这个口径要给的数
|
||||
contract.update({"metrics_from": "card", "required": [m["id"] for m in metrics]})
|
||||
|
||||
@@ -180,7 +180,11 @@ async def test_freezing_makes_a_fixed_sql_draft_that_passes_the_governed_gate(cl
|
||||
"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}"
|
||||
# 有入口参数：不写代表那一轮的问题原文（「九月第一周」参数化之后就不对了），写运行时渲染的参数
|
||||
instructions = nodes["write"]["data"]["config"]["instructions"]
|
||||
assert QUESTION not in instructions and "2026-09" not in instructions, instructions
|
||||
assert "{{ input.p1 }}" in instructions and "{{ input.p2 }}" in instructions
|
||||
assert nodes["card"]["data"]["config"]["caliber"] == 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"], \
|
||||
@@ -189,9 +193,12 @@ async def test_freezing_makes_a_fixed_sql_draft_that_passes_the_governed_gate(cl
|
||||
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}
|
||||
# 队列上标着已经固化过：同一个工作流只列一次（三个来源运行都指着它）
|
||||
body = (await client.get("/api/ledger/certify-queue")).json()
|
||||
[again] = shape_items(body)
|
||||
assert [w["workflow_id"] for w in again["frozen_into"]] == [wf.id]
|
||||
[cluster] = [i for i in body["items"] if i["kind"] == "cluster" and i["question"] == QUESTION]
|
||||
assert [w["workflow_id"] for w in cluster["frozen_into"]] == [wf.id], cluster["frozen_into"]
|
||||
|
||||
# 按受管级别发布：来源运行在后台补比，变成等价已证明、出了队
|
||||
res = await client.post(f"/api/workflows/{wf.id}/publish", json={"level": "governed"})
|
||||
@@ -216,6 +223,47 @@ async def test_freezing_makes_a_fixed_sql_draft_that_passes_the_governed_gate(cl
|
||||
assert stats["frozen"] >= 1 and stats["published"] >= 1
|
||||
|
||||
|
||||
async def test_freezing_uses_the_given_name_and_catalog_labels_and_no_dates_without_parameters(client, monkeypatch):
|
||||
"""传了名字：口径卡用它（下一轮验收：以前用代表问题的原文）。指标名能从数据目录取到中文名的用中文名，取不到照旧用
|
||||
列名。没有入口参数（字面量都没变过）时报告指令照旧写问题原文：那时问题里的日期就是口径的常量。"""
|
||||
from app.data import catalog
|
||||
from app.db.models import DataSource
|
||||
from sqlalchemy import select
|
||||
|
||||
async with SessionLocal() as session:
|
||||
source = (await session.execute(select(DataSource).where(DataSource.name == S.CURRENT["source"]))).scalar_one()
|
||||
await catalog.write_entry(session, source.id, "orders", {"columns": {
|
||||
"amount": {"label": catalog.make_item("金额", "human")}}}, if_version=0, actor="测试员")
|
||||
sql = "SELECT SUM(amount) AS amount, COUNT(*) AS n FROM orders WHERE status = 'paid' AND paid_at >= '2026-09-01'"
|
||||
S.script(monkeypatch, "金额 [[v:Q1.r0.amount]]，[[v:Q1.r0.n]] 单。")
|
||||
conv = await conversation()
|
||||
async with SessionLocal() as session:
|
||||
turn = ConversationTurn(conversation_id=conv, question=QUESTION)
|
||||
session.add(turn)
|
||||
await session.commit()
|
||||
turn_id = turn.id
|
||||
run = await run_manager.start(graph=S.explore_graph(sql), input_payload={"question": QUESTION}, origin_kind="chat",
|
||||
origin={"conversation_id": conv, "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()
|
||||
res = await client.post("/api/copilot/from-run", json={"run_id": row.id, "freeze": True, "name": "九月销售额口径"})
|
||||
assert res.status_code == 200, res.text
|
||||
made = res.json()
|
||||
assert made["params"] == [] and made["name"] == "九月销售额口径"
|
||||
async with SessionLocal() as session:
|
||||
wf = await session.get(Workflow, made["workflow_id"])
|
||||
nodes = {n["id"]: n for n in wf.graph["nodes"]}
|
||||
card = nodes["card"]["data"]["config"]
|
||||
assert card["caliber"] == "九月销售额口径"
|
||||
assert [(m["name"], m["expression"]) for m in card["metrics"]] == [
|
||||
("金额", "cell(nodes.fetch, 0, 'amount')"), ("n", "cell(nodes.fetch, 0, 'n')")]
|
||||
assert nodes["write"]["data"]["config"]["instructions"] == f"回答：{QUESTION}"
|
||||
|
||||
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user