diff --git a/backend/app/apps/describe.py b/backend/app/apps/describe.py index 284fe38d..61f359ca 100644 --- a/backend/app/apps/describe.py +++ b/backend/app/apps/describe.py @@ -47,6 +47,13 @@ def describe_version(version: Any, *, workflow_name: str) -> str: else f"口径卡{title}，{caliber}，指标转抄自别处的口径卡") lines += [f"- {node_id}.{m['id']} {m.get('name') or ''}{('（' + m['unit'] + '）') if m.get('unit') else ''}" for m in metrics] + elif node["type"] == "analysis": + from app.engine.labels import ANALYSIS_OP_LABEL + + items = [i for i in cfg.get("items") or [] if isinstance(i, dict) and i.get("id")] + lines.append(f"分析卡{title}，分析项 {len(items)} 个：" if items else f"分析卡{title}，还没有分析项") + lines += [f"- {node_id}.results.{i['id']} {ANALYSIS_OP_LABEL.get(str(i.get('op')), str(i.get('op') or ''))}" + for i in items] elif node["type"] == "report": lines.append(f"报告撰写{title}") elif node["type"] in ("tool", "merge"): diff --git a/backend/app/core/events.py b/backend/app/core/events.py index fa295c27..e03d0b8a 100644 --- a/backend/app/core/events.py +++ b/backend/app/core/events.py @@ -51,6 +51,9 @@ class EventType(StrEnum): # 合并查询做完了：哪几个输入（别名、节点、行数、数据源）、哪条合并 SQL、得到几行、有哪些警告。 # 运行详情据此展示这一步；完整结果在 query_artifact 指向的查询快照里 MERGE_END = "merge.end" + # 分析卡里的一项做完了（engine/nodes/analysis.py）：哪一项、什么分析、状态、判定、判定语、结论工件（analysis_artifact）。 + # 运行详情据此展示这一步，运行对比按项对齐两次的判定；整份结论在工件里。agent 分析工具的结论记在 tool.end 上 + ANALYSIS_END = "analysis.end" SANDBOX_START = "sandbox.start" SANDBOX_END = "sandbox.end" diff --git a/backend/app/data/catalog_impact.py b/backend/app/data/catalog_impact.py index c0d330e4..31f1c49b 100644 --- a/backend/app/data/catalog_impact.py +++ b/backend/app/data/catalog_impact.py @@ -150,6 +150,8 @@ def template_refs(spec: Any, source: str, table: str, tables: Iterable[str] | No - Agent 节点、协作成员：绑定了 db_query__<源> → possible（每个成员各一项，带 member）。 - 合并查询：输入（别名 → 节点）里有用到这张表的 → 取输入里最强的那一级，via 写经由哪些输入。合并的合并 一路追到底；成环（校验会报错的图）时停下。 + - 分析卡：各项的输入（主输入、两份查询、总数两格落在的节点）里有用到这张表的，同合并查询（目录里度量类型、业务日期 + 列改了，分析的判定可能跟着变），via 写经由哪些输入（alias 是那一项的标识）。 """ wanted = name_key(table) known = list(tables) if tables is not None else None @@ -175,16 +177,15 @@ def template_refs(spec: Any, source: str, table: str, tables: Iterable[str] | No resolved: dict[str, dict[str, Any] | None] = {} def through(node_id: str, trail: frozenset[str]) -> dict[str, Any] | None: - """合并查询节点经由输入用到这张表的样子；用不到返回 None。""" + """合并查询、分析卡经由输入用到这张表的样子；用不到返回 None。""" if node_id in resolved: return resolved[node_id] node = by_id.get(node_id) - if node is None or _kind(node) != "merge" or node_id in trail: + if node is None or _kind(node) not in ("merge", "analysis") or node_id in trail: return own.get(node_id) via: list[dict[str, Any]] = [] best: str | None = None - inputs = (node.config or {}).get("inputs") - for alias, target in (inputs.items() if isinstance(inputs, Mapping) else []): + for alias, target in _inputs_of(node): if not isinstance(target, str): continue hit = own.get(target) or through(target, trail | {node_id}) @@ -198,13 +199,28 @@ def template_refs(spec: Any, source: str, table: str, tables: Iterable[str] | No for node in nodes: if node.id in own: out.append(own[node.id]) - elif _kind(node) == "merge" and (hit := through(node.id, frozenset())): + elif _kind(node) in ("merge", "analysis") and (hit := through(node.id, frozenset())): out.append(hit) order = {n.id: i for i, n in enumerate(nodes)} out.sort(key=lambda r: order.get(r["node_id"], 0)) return out +def _inputs_of(node: Any) -> list[tuple[str, Any]]: + """合并查询、分析卡的输入：[(别名, 节点 id)]。合并查询是 inputs 的别名；分析卡是各项的标识（一项引用几个节点就几条， + schema.analysis_refs）。""" + config = node.config or {} + if _kind(node) == "merge": + inputs = config.get("inputs") + return list(inputs.items()) if isinstance(inputs, Mapping) else [] + from app.engine.schema import analysis_refs + + items = config.get("items") + found = [(str(item.get("id") or ""), target) for item in (items if isinstance(items, list) else []) + if isinstance(item, Mapping) for _, target in analysis_refs(dict(item))] + return list(dict.fromkeys(found)) + + async def _published(session: AsyncSession) -> list[tuple[Any, Any]]: """每个发布过的模板和它当前的已发布版本：[(Workflow, WorkflowVersion)]。""" from app.db.models import Workflow, WorkflowVersion diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index 25dd3c17..8e965bb2 100644 --- a/backend/app/engine/compiler.py +++ b/backend/app/engine/compiler.py @@ -13,7 +13,7 @@ from app.core.events import EventType from app.engine.context import NodeContext, NodeError, NodeRejected, RunContext from app.engine.expressions import open_template_scope, template_scope from app.engine.nodes import ( - challenge, control, dual_check, human, io, knowledge, llm, merge, metrics, multi, report, tools, + analysis, challenge, control, dual_check, human, io, knowledge, llm, merge, metrics, multi, report, tools, ) from app.engine.schema import ( GATE_TYPES, REJECT_HANDLE, GraphNode, GraphSpec, NodeType, approve_mode, back_edges, innermost_loops, @@ -42,6 +42,7 @@ RUNNERS: dict[NodeType, NodeRunner] = { NodeType.VALIDATE: human.run_validate, NodeType.MEMORY: knowledge.run_memory, NodeType.METRICS: metrics.run_metrics, + NodeType.ANALYSIS: analysis.run_analysis, NodeType.RETRIEVE: knowledge.run_retrieve, NodeType.REPORT: report.run_report, NodeType.DUAL_CHECK: dual_check.run_dual_check, diff --git a/backend/app/engine/governance.py b/backend/app/engine/governance.py index 3f7c0c54..e2f389b7 100644 --- a/backend/app/engine/governance.py +++ b/backend/app/engine/governance.py @@ -214,8 +214,9 @@ def _lint_sql(spec: GraphSpec, flag: Callable[..., None], checkers: Mapping[str, sql_level=check.level) -#: validate 里和门禁 G4、G5 说同一件事的两条提示（schema.evidence_issues） -_REPEATS = frozenset({"evidence.caliber_compute_input", "evidence.cite_fields_off"}) +#: validate 里和门禁 G4、G5 说同一件事的两条提示（schema.evidence_issues），和受管级分析卡输入那条（governed.analysis_input） +#: 说同一件事的画布提醒（schema.analysis_issues 对调用别的工具的「调用工具」节点的那条） +_REPEATS = frozenset({"evidence.caliber_compute_input", "evidence.cite_fields_off", "analysis.input_invalid"}) def _not_repeated(issues: list[Any], gate: list[Any], spec: GraphSpec) -> list[Any]: @@ -231,6 +232,7 @@ def _not_repeated(issues: list[Any], gate: list[Any], spec: GraphSpec) -> list[A cited = {i.node_id for i in gate if i.code == "governed.caliber_agent_cite_fields"} compute = {i.node_id for i in gate if i.code == "governed.caliber_compute_input"} covered = {("evidence.cite_fields_off", nid, "output_schema") for nid in cited} + covered |= {("analysis.input_invalid", i.node_id, i.field) for i in gate if i.code == "governed.analysis_input"} for card in spec.nodes: if card.type == NodeType.METRICS and compute: covered |= {("evidence.caliber_compute_input", card.id, f"metrics[{i}].expression") @@ -344,7 +346,8 @@ REPORT_POLICY_ALSO: dict[str, tuple[str, ...]] = {"claims": (CLAIMS_JUDGE,)} JUDGE_UNLIMITED = "不设上限，费用只受句数、时长上限和每日上限约束" #: 产出本身就是证据、或者由系统确定地算出来的节点。成果字段取它们的值，不算「别的节点写的字」 -_TRACEABLE = (NodeType.REPORT, NodeType.METRICS, NodeType.INPUT, NodeType.TOOL, NodeType.RETRIEVE, NodeType.MERGE) +_TRACEABLE = (NodeType.REPORT, NodeType.METRICS, NodeType.INPUT, NodeType.TOOL, NodeType.RETRIEVE, NodeType.MERGE, + NodeType.ANALYSIS) _WRITERS = (NodeType.LLM, NodeType.AGENT, NodeType.SUPERVISOR) #: 往对话里写消息的节点（emit_message 没关时），「最后一条消息」就是它们里最后跑的那个写的 _EMITTERS = (*_WRITERS, NodeType.REPORT) @@ -710,6 +713,39 @@ def _lint_evidence(spec: GraphSpec, flag: Callable[..., None], *, strict: bool) flag(f"{_who(feeder)}为{to}提供数据：「模型调用」节点不调用工具，写出的数字没有出处。请改由 Agent" "（配置「结构化输出 Schema」并开启「按出处核对字段」）或「调用工具」节点取数", code="governed.caliber_model_input", node_id=feeder.id, hard=True) + elif feeder.type == NodeType.ANALYSIS: + # D16：分析是叙述的依据，不是口径的原料。口径卡读它的结果按「追不到原始数据」处理，这里只提示 + flag(f"{_who(feeder)}的结果提供给了{to}：分析结论不作口径卡的输入，口径卡读到的数追不到原始数据。" + "指标请在口径卡里按格从查询结果取数，分析结论在报告里直接引用", + code="governed.caliber_analysis_input", node_id=feeder.id) + + # 受管级的分析卡：输入要在发布时就确定是平台执行的查询（调用数据库查询工具的「调用工具」节点、合并查询） + if strict: + _lint_analysis_inputs(spec, flag) + + +def _lint_analysis_inputs(spec: GraphSpec, flag: Callable[..., None]) -> None: + """受管级别的分析卡，每项的输入（主输入、两份查询、总数两格落在的节点）都要是调用数据库查询工具的「调用工具」节点 + 或「合并查询」。画布校验（schema.analysis_issues）对调用别的工具的「调用工具」节点只提醒（声明了表格的端点读取运行时 + 也交回查询快照，静态看不出），已发布级别照那条提醒；受管出具要发布时就能确定分析的是平台执行的查询，按错误挡， + 画布那条同义的提醒不再列（_not_repeated）。""" + from app.engine.schema import _produces_query, analysis_refs + + nodes = spec.node_map() + for card in spec.nodes: + if card.type != NodeType.ANALYSIS or not isinstance(card.config.get("items"), list): + continue + for i, item in enumerate(card.config["items"]): + if not isinstance(item, dict): + continue + for field, target_id in analysis_refs(item): + target = nodes.get(target_id) + if target is None or target.type != NodeType.TOOL or _produces_query(target): + continue # 不存在、不是查询节点的，画布校验已经按错误报了 + flag(f"{_who(card)}的分析项「{item.get('id')}」分析的是{_who(target)}的结果，它调用的不是数据库查询工具：" + "受管级别要求分析卡的输入是调用数据库查询工具的「调用工具」节点或「合并查询」，发布时就能确定分析的" + "是平台执行的查询。请改选这样的节点", + code="governed.analysis_input", node_id=card.id, hard=True, field=f"items[{i}].{field}") async def unresolved_caliber_upgrades( diff --git a/backend/app/engine/labels.py b/backend/app/engine/labels.py index 1ba796a7..95e25941 100644 --- a/backend/app/engine/labels.py +++ b/backend/app/engine/labels.py @@ -21,7 +21,7 @@ TYPE_LABEL: dict[str, str] = { "supervisor": "多 Agent 协作", "tool": "调用工具", "code": "沙箱代码", "merge": "合并查询", "branch": "条件分支", "loop": "循环", "subgraph": "子工作流", "memory": "长期记忆", "retrieve": "知识检索", "transform": "数据整形", - "human": "人工审批", "validate": "结构校验", "metrics": "口径卡", "report": "报告撰写", + "human": "人工审批", "validate": "结构校验", "metrics": "口径卡", "analysis": "分析卡", "report": "报告撰写", "dual_check": "双盲复算", "challenge": "反方审稿", } @@ -99,10 +99,19 @@ _FIELD_BY_TYPE: dict[str, dict[str, str]] = { "stop_on_reject": "驳回即终止运行"}, "validate": {"source": "待校验内容", "fail_fast": "校验失败即中断"}, "merge": {"inputs": "输入", "sql": "合并 SQL"}, + "analysis": {"items": "分析项", "overrides": "阈值覆盖"}, "dual_check": DUAL_CHECK_FIELD_LABEL, "challenge": CHALLENGE_LABEL, } +#: 分析卡一项里的分析方法、模式的叫法（app/analysis/contract.py 的 OPS、MODES；前端分析卡的表单照这一份） +ANALYSIS_OP_LABEL: dict[str, str] = {"decompose": "变化拆解", "baseline": "波动基线", "trend": "趋势", + "differ": "差异检验", "comparable": "可比性检查"} +ANALYSIS_MODE_LABEL: dict[str, dict[str, str]] = { + "decompose": {"additive": "可加（按维度算贡献）", "ratio": "比率（拆结构与自身）", "factor": "乘积（拆因子）"}, + "differ": {"rate": "比率", "mean": "均值", "rows": "逐行的值"}, +} + #: 出具契约里的键（canvas/issues.ts 的 CONTRACT_KEYS、SUB_KEYS） CONTRACT_KEY_LABEL: dict[str, str] = { "metrics_from": "指标来自", "report_from": "报告来自", "required": "必需指标", "expected": "期望指标", diff --git a/backend/app/engine/nodes/analysis.py b/backend/app/engine/nodes/analysis.py new file mode 100644 index 00000000..f6a2a584 --- /dev/null +++ b/backend/app/engine/nodes/analysis.py @@ -0,0 +1,166 @@ +"""分析卡（节点类型 analysis，docs/design/analysis-round/00-总方案.md §3.2）：对上游查询节点的结果固定做几项确定性分析。 + +**模板定骨架。** 做哪几项、每项分析哪个上游查询节点、用哪种方法，写在节点上（items），随版本固定；阈值来自发起时的设置 +快照，卡上可以写覆盖（overrides，同样随版本固定）。问数据里要看了数才知道拆哪个维度的，交给 Agent 的分析工具——两个入口 +共用一套算子（app.analysis.run）、一种结论形状、一个证据种类。 + +**读工件库里那份完整结果。** 照合并查询的做法（engine/nodes/merge.py）从 state.nodes 取上游节点交回的查询快照 id，按 id +取回快照（取回时复验哈希），而且只认这次运行台账里 kind query 的那几份：别的工具拼一个同形的 JSON，里面的快照不是平台 +落的。遮罩列（数据源设的、按 SQL 认出改了名字的）不当维度、不当度量，和证据面板同一套认法（app/tools/analysis.py）。 + +**产出。** nodes.<卡>.results.<项> = {status, verdict, direction, values, phrase, artifact}：verdict 是判定码（outside_band、 +driver……），下游条件分支能按它走（周报「超出常规波动就转人工」）。每项的整份结论落成工件，台账记一条 kind analysis +（contract.ledger_entry，带 item），报告的写作目录编成 A{n}；每项另发一条 analysis.end（运行详情、运行对比按项对齐）。 + +**一项算不了不让卡失败。** 算不了（历史不够、本期不完整、截断……）就是这一项的结论（status refused），照实记、发一条 +analysis_refused 提醒，接着做下一项。调用写错（列不存在、比率当可加……）是模板的毛病：整张卡失败，说清是哪一项、怎么改。 + +分析的结果不作口径卡的输入（D16）：口径卡读 nodes.<卡>.results 的，按追不到原始数据处理（受管级发布时另有提示）。 +""" +from __future__ import annotations + +import asyncio +from typing import Any + +from pydantic import ValidationError + +from app.analysis import contract +from app.analysis.contract import AnalysisResult, CallError, InputSnapshot +from app.core import artifact_store +from app.core.errors import describe_exception +from app.core.events import EventType +from app.engine.context import NodeContext, NodeError +from app.engine.evidence import ledger_enabled, next_exec +from app.engine.labels import ANALYSIS_OP_LABEL +from app.engine.nodes.merge import _FAILED_PREFIXES, _input_payload +from app.engine.state import GraphState +from app.tools.analysis import ( + REFUSED_CODE, context_of, hypothesis_of, input_snapshot, masked_of, run_snapshot, settings_of, store, tool_context, +) + + +def card_view(result: AnalysisResult, artifact: str) -> dict[str, Any]: + """一项结论在节点产出里的样子（nodes.<卡>.results.<项>）：状态、判定码、方向、具名标量、判定语、结论工件。""" + verdict = result.get("verdict") or {} + return {"status": result.get("status"), "verdict": verdict.get("code"), "direction": verdict.get("direction"), + "values": dict(result.get("values") or {}), "phrase": result.get("phrase") or "", "artifact": artifact} + + +async def run_analysis(state: GraphState, ctx: NodeContext) -> dict[str, Any]: + from app.engine.schema import analysis_issues, analysis_refs + + # 配置的规矩和画图时同一套（schema.analysis_issues）：发起前的校验已经挡过，这里兜底，写错了照样说清是哪一项 + if errors := [i for i in analysis_issues(ctx.run.spec, ctx.node) if i.level == "error"]: + raise NodeError(ctx.node.id, errors[0].message) + items: list[dict[str, Any]] = ctx.config["items"] + settings = await settings_of(run_snapshot(ctx.run), ctx.config.get("overrides") or None) + + queries = [e for e in state.get("evidence") or [] if isinstance(e, dict) and e.get("kind") == "query"] + inputs: dict[str, InputSnapshot] = {} + for item in items: + for _, node_id in analysis_refs(item): + if node_id not in inputs: + inputs[node_id] = await _load_input(state, ctx, node_id, queries) + context = await context_of(list(inputs.values()), tool_ctx=tool_context(ctx.run, ctx.node.id, ctx.emit)) + + import app.analysis as analysis_ops # 按模块取 run：测试、合并前的替身都只换模块上的这一个函数 + + exec_no = next_exec(state, ctx.node.id) + done: dict[str, AnalysisResult] = {} + results: dict[str, Any] = {} + entries: list[dict[str, Any]] = [] + for item in items: + item_id, op = item["id"], item["op"] + who = f"分析项「{item_id}」（{ANALYSIS_OP_LABEL[op]}）" + request = _request(item, state, ctx, who) + try: + # 下钻（parent）只认这张卡里排在前面、已经做完的项：done 里此刻只有它们 + result = await analysis_ops.run(request, inputs.get, settings=settings, context=context, parents=done.get) + except CallError as e: + raise NodeError(ctx.node.id, f"{who}：{e.message}。请在分析卡中改正这一项") from None + except Exception as e: # noqa: BLE001 - 算子本身出错：说清是哪一项，原因照实 + raise NodeError(ctx.node.id, f"{who}执行出错：{describe_exception(e)}") from e + try: + artifact = await store(result, run_id=ctx.run.run_id, node_id=ctx.node.id) + except Exception as e: # noqa: BLE001 - 存不下就没有出处可言，不能带着一份引用不了的结论往下走 + raise NodeError(ctx.node.id, f"{who}的结论没能存下来（工件库写入失败），下游无法引用它。" + "请稍后点「继续运行」重试", retryable=True) from e + done[item_id] = result + results[item_id] = card_view(result, artifact) + entries.append(contract.ledger_entry(result, node_id=ctx.node.id, exec_no=exec_no, artifact=artifact, + item=item_id)) + verdict = result.get("verdict") or {} + ctx.emit(EventType.ANALYSIS_END, item=item_id, op=op, mode=result.get("mode"), status=result.get("status"), + verdict=verdict.get("code"), direction=verdict.get("direction"), phrase=result.get("phrase") or "", + reason_code=result.get("reason_code"), analysis_artifact=artifact) + if result.get("status") == "refused": + # 算不了是一条结论，不让卡失败；照实提醒一句（结构化字段给界面读，不让它从 message 里解析中文） + ctx.emit(EventType.LOG, level="warn", code=REFUSED_CODE, item=item_id, op=op, + reason_code=result.get("reason_code"), reason=result.get("reason"), + message=f"「{ctx.node.title}」的{who}没有做成：{result.get('reason') or '原因未知'}") + + output: dict[str, Any] = {"results": results} + updates: dict[str, Any] = {"nodes": {ctx.node.id: output}} + if ledger_enabled(ctx.run) and entries: + updates["evidence"] = entries + var_name = str(ctx.config.get("assign_to") or "").strip() + if var_name: + updates["vars"] = {var_name: output} + return updates + + +def _request(item: dict[str, Any], state: GraphState, ctx: NodeContext, who: str) -> contract.AnalysisRequest: + """一项 → 一次分析的请求。参数里的 {{ }} 先按当前状态渲染（按运行输入选期次）；主输入（from）就是 agent 工具的 + query。参数按 agent 工具同一份参数模型校验（tools/builtin/analysis.ARGS），写错了说清是哪一项、它接受什么。""" + from app.tools.builtin.analysis import ARGS, request_of + from app.tools.registry import describe_args + + op = item["op"] + params = ctx.render(dict(item.get("params") or {}), state) + if isinstance(item.get("from"), str) and item["from"].strip(): + params["query"] = item["from"].strip() + if op in contract.MODES: + params["mode"] = item.get("mode") + if hypothesis := hypothesis_of(item.get("hypothesis")): + params["hypothesis"] = hypothesis + try: + args = ARGS[op].model_validate(params) + except ValidationError as e: + raise NodeError(ctx.node.id, f"{who}的{describe_args(ARGS[op], params, e)}") from None + return request_of(op, args) + + +async def _load_input(state: GraphState, ctx: NodeContext, node_id: str, + queries: list[dict[str, Any]]) -> InputSnapshot: + """上游一个查询节点这次交回的查询快照 → 算子的一份输入（含遮罩列）。只认这次运行台账里 kind query 的快照。""" + target = ctx.run.spec.node_map()[node_id] # 节点在不在、在不在上游、是不是查询节点，执行前已经校验过 + who = f"输入「{target.title}」" + value = (state.get("nodes") or {}).get(node_id) + if value is None: + raise NodeError(ctx.node.id, f"{who}还没有运行结果：请确认它连在分析卡之前") + if isinstance(value, dict) and value.get("skipped"): + raise NodeError(ctx.node.id, f"{who}满足跳过条件被跳过了，没有查询结果可以分析") + if isinstance(value, dict) and value.get("failed"): + raise NodeError(ctx.node.id, f"{who}执行失败，没有查询结果可以分析：{value.get('error') or '原因未知'}") + if isinstance(value, str) and value.startswith(_FAILED_PREFIXES): + raise NodeError(ctx.node.id, f"{who}的查询没有成功，无法分析：{value[:300]}") + payload = _input_payload(value) + artifact = payload.get("artifact") if payload else None + if not isinstance(artifact, str) or not artifact: + raise NodeError(ctx.node.id, f"{who}的结果不是存成快照的查询结果。分析卡只能分析数据库查询（调用数据库查询工具的" + "「调用工具」节点）或「合并查询」的结果") + if not any(e.get("artifact") == artifact for e in queries): + raise NodeError(ctx.node.id, f"{who}交回的不是这次运行里平台执行的查询（台账里没有这份查询快照），不能分析。" + "请改选调用数据库查询工具的「调用工具」节点或「合并查询」") + try: + content = await asyncio.to_thread(artifact_store.load, artifact) + except ValueError: + raise NodeError(ctx.node.id, f"{who}的查询快照与哈希不一致，疑似被修改，不能分析") from None + if not isinstance(content, dict): + raise NodeError(ctx.node.id, f"{who}的查询快照已不存在，无法分析。请重新运行它") + snapshot = input_snapshot(content, ref=node_id, artifact=artifact, node_id=node_id) + snapshot["masked"] = await masked_of(content, artifact) + return snapshot + + +__all__ = ["REFUSED_CODE", "card_view", "run_analysis"] diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index ac1c88de..d040fa29 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -694,7 +694,7 @@ _REF_HEAD = re.compile(r"^(vars|nodes|input)\.([A-Za-z_一-鿿][\w\-一-鿿]*)(. _VIA = {"input": "input", "code": "code", "agent": "agent", "llm": "llm", "tool": "tool", "transform": "transform", "metrics": "metric", "retrieve": "retrieval", "loop": "loop", "human": "human", "validate": "validate", "supervisor": "team", "subgraph": "subgraph", - "memory": "memory", "merge": "merge"} + "memory": "memory", "merge": "merge", "analysis": "analysis"} def _input_of(path: str, value: Any, state: GraphState, ctx: NodeContext, diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index e6e439da..447591cb 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -43,6 +43,8 @@ class NodeType(StrEnum): HUMAN = "human" # 人工介入 VALIDATE = "validate" # 结构化校验 + 自动重试 METRICS = "metrics" # 口径卡：受控指标集，叙述层唯一合法的数字来源 + # 分析卡：对上游查询快照做固定的几项确定性分析（拆解、基线、趋势、差异、可比性），和口径卡平级（engine/nodes/analysis.py） + ANALYSIS = "analysis" REPORT = "report" # 报告撰写：模型只写引用标记，数字由系统从口径卡取出来渲染 # 下面两种目前只是占位（执行器原样放行），各自的工作包补上真正的逻辑 DUAL_CHECK = "dual_check" # 双盲复算：另起一路独立算一遍，和原结果比对 @@ -632,6 +634,167 @@ def _check_merge(node: GraphNode, spec: GraphSpec, result: ValidationResult) -> result.add(problem, node_id=node.id, field="sql", code="merge.sql_invalid") +#: 分析卡一项的标识：要能写进取值链（nodes.<卡>.results.<标识>），条件分支按它取判定 +ANALYSIS_ITEM_ID = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +#: 分析卡一项里写上游节点 id 的参数：基期、本期是两份查询时各选一个（主输入写在 from，对应 agent 工具的 query） +ANALYSIS_NODE_PARAMS = ("base_query", "target_query") +#: 一项里写成「节点 id.r行号.列名」的格（总数 headline 的两格） +ANALYSIS_CELL = re.compile(r"^([A-Za-z_][\w-]*)\.r(\d+)\.(.+)$") + + +def analysis_refs(item: dict[str, Any]) -> list[tuple[str, str]]: + """分析卡一项引用的上游节点：[(这一项里的字段路径, 节点 id)]，按 from、base_query、target_query、总数两格的先后。 + 写错了的（不是字符串、格的写法不对）不在里面，由 analysis_issues 报。""" + params = item.get("params") if isinstance(item.get("params"), dict) else {} + out: list[tuple[str, str]] = [] + if isinstance(item.get("from"), str) and item["from"].strip(): + out.append(("from", item["from"].strip())) + out += [(f"params.{k}", params[k].strip()) for k in ANALYSIS_NODE_PARAMS + if isinstance(params.get(k), str) and params[k].strip()] + headline = params.get("headline") + for side in ("base", "target"): + cell = ANALYSIS_CELL.match(headline[side].strip()) \ + if isinstance(headline, dict) and isinstance(headline.get(side), str) else None + if cell is not None: + out.append((f"params.headline.{side}", cell.group(1))) + return out + + +def _analysis_input(spec: GraphSpec, target_id: str, upstream: set[str]) -> tuple[str, str] | None: + """分析卡一项的一个输入有什么问题：(级别, 半句说明)，没问题为 None。规则同合并查询：只认能稳定取到一份查询快照的 + 上游节点。调用的不是数据库查询工具的「调用工具」节点只提醒：登记里声明了表格的端点读取运行时也交回查询快照，静态看 + 不出来；交回的不是平台执行的查询，运行到这一项就失败（受管级别发布时按错误挡，engine/governance.py）。""" + target = spec.node_map().get(target_id) + if target is None: + return "error", f"选择的输入「{target_id}」在工作流中不存在" + if target.type == NodeType.AGENT: + return "error", (f"选择的输入「{target.title}」是 Agent：Agent 可能查询多次，无法确定分析哪一次的结果。" + "请把要分析的那条查询单独放进「调用工具」节点（选择数据库查询工具），再把它选作输入；" + "要 Agent 看了数再决定拆哪个维度的，给它绑分析工具") + if target.type not in (NodeType.TOOL, NodeType.MERGE): + return "error", (f"选择的输入「{target.title}」是「{type_label(target.type)}」节点，不产出查询结果。" + "输入只能是调用数据库查询工具的「调用工具」节点，或「合并查询」节点") + if target_id not in upstream: + return "error", f"选择的输入「{target.title}」不在分析卡的上游：分析时它还没有运行。请把它连到分析卡之前" + if not _produces_query(target): + return "warning", (f"选择的输入「{target.title}」调用的不是数据库查询工具：运行时它交回的要是平台执行的查询结果" + "才能分析，否则这一项会失败") + return None + + +def analysis_issues(spec: GraphSpec, node: GraphNode) -> list[ValidationIssue]: + """分析卡的配置有什么问题（画图时的校验和执行时同一套，engine/nodes/analysis.py 执行前先过一遍，模板写错了早暴露）。 + + - items：至少一项；每项的标识合法、不重复；分析方法（op）、模式（mode）是契约里的；输入（from，或两份查询时的 + base_query、target_query，总数两格落在的节点）指向上游的「调用工具」/「合并查询」节点 + - params：只能写这种分析的参数（和 agent 工具的参数同一份定义，tools/builtin/analysis.py），不能写阈值——阈值统一 + 写在 overrides（阈值覆盖）里，按 contract.settings_problem 校验 + - parent（下钻）：指向这张卡里排在它前面的一项变化拆解 + """ + from app.analysis import contract + from app.engine.labels import ANALYSIS_MODE_LABEL, ANALYSIS_OP_LABEL + from app.tools.builtin.analysis import param_names + + out: list[ValidationIssue] = [] + + def add(message: str, field: str, code: str, level: str = "error") -> None: + out.append(ValidationIssue(level=level, message=message, node_id=node.id, field=field, # type: ignore[arg-type] + code=code)) + + cfg = node.config + items = cfg.get("items") + if not isinstance(items, list) or not items: + add("「分析卡」还没有配置分析项：至少写一项，说清做什么分析、分析哪个上游查询节点的结果", "items", + "analysis.items_missing") + items = [] + upstream = _ancestors(spec, node.id) + seen: dict[str, int] = {} + for i, item in enumerate(items): + at = f"items[{i}]" + if not isinstance(item, dict): + add(f"第 {i + 1} 个分析项的格式有误，需要是一个对象", at, "analysis.item_invalid") + continue + item_id = item.get("id") + who = f"分析项「{item_id}」" if isinstance(item_id, str) and item_id else f"第 {i + 1} 个分析项" + if not isinstance(item_id, str) or not ANALYSIS_ITEM_ID.match(item_id): + add(f"{who}的标识只能用字母、数字和下划线，且不以数字开头：下游按 nodes.{node.id}.results.<标识> 取它的判定", + f"{at}.id", "analysis.item_invalid") + elif item_id in seen: + add(f"有两个分析项的标识都是「{item_id}」，标识要唯一", f"{at}.id", "analysis.item_invalid") + else: + seen[item_id] = i + op, mode = item.get("op"), item.get("mode") + if op not in contract.OPS: + add((f"{who}的分析方法「{op}」不认识" if op not in (None, "") else f"{who}还没有选择分析方法") + "，可选：" + + "、".join(ANALYSIS_OP_LABEL.values()), f"{at}.op", "analysis.op_invalid") + continue + modes = contract.MODES.get(op) + if modes and mode not in modes: + add(f"{who}是{ANALYSIS_OP_LABEL[op]}，要选一种模式：" + "、".join(ANALYSIS_MODE_LABEL[op].values()) + + (f"（当前为「{mode}」）" if mode not in (None, "") else ""), f"{at}.mode", "analysis.mode_invalid") + elif not modes and mode not in (None, ""): + add(f"{who}是{ANALYSIS_OP_LABEL[op]}，没有模式可选，请去掉模式", f"{at}.mode", "analysis.mode_invalid") + params = item.get("params") if item.get("params") is not None else {} + if not isinstance(params, dict): + add(f"{who}的参数需要是一个对象", f"{at}.params", "analysis.params_invalid") + continue + allowed = [k for k in param_names(op) if k != "query"] + for key in params: + if key in contract.DEFAULTS: + add(f"{who}的参数里写了阈值「{contract.LABELS[key]}」：阈值不写在分析项里，统一写在这张卡的" + f"「{field_label('overrides', NodeType.ANALYSIS)}」里", f"{at}.params.{key}", "analysis.params_invalid") + elif key == "query": + add(f"{who}要分析的查询节点写在输入里，参数里不写", f"{at}.params.{key}", "analysis.params_invalid") + elif key not in allowed: + add(f"{who}的参数里没有「{key}」，{ANALYSIS_OP_LABEL[op]}可以写：" + "、".join(allowed), + f"{at}.params.{key}", "analysis.params_invalid") + if item.get("hypothesis") is not None and not isinstance(item.get("hypothesis"), str): + add(f"{who}的「在验证什么」需要是一段文字", f"{at}.hypothesis", "analysis.item_invalid") + both = all(isinstance(params.get(k), str) and params[k].strip() for k in ANALYSIS_NODE_PARAMS) + if not (isinstance(item.get("from"), str) and item["from"].strip()) and not both: + two = "；基期、本期是两份查询时，在参数里选上这两个查询节点" if op in ("decompose", "comparable") else "" + add(f"{who}还没有选择输入：选一个上游的查询节点（调用数据库查询工具的「调用工具」节点，或「合并查询」）{two}", + f"{at}.from", "analysis.input_invalid") + headline = params.get("headline") + if headline is not None and not (isinstance(headline, dict) + and all(isinstance(headline.get(k), str) for k in ("base", "target"))): + add(f"{who}的总数要写成基期、本期两格，每格写成「节点 id.r行号.列名」，行号从 0 数", + f"{at}.params.headline", "analysis.params_invalid") + elif isinstance(headline, dict): + for side, said in (("base", "基期"), ("target", "本期")): + if not ANALYSIS_CELL.match(headline[side].strip()): + add(f"{who}的总数里{said}那一格要写成「节点 id.r行号.列名」（行号从 0 数），当前为「{headline[side]}」", + f"{at}.params.headline.{side}", "analysis.params_invalid") + for field, target_id in analysis_refs(item): + if (found := _analysis_input(spec, target_id, upstream)) is not None: + add(f"{who}{found[1]}", f"{at}.{field}", "analysis.input_invalid", found[0]) + parent = params.get("parent") if "parent" in allowed else None # 别的分析写了下钻，上面已按多写的参数报 + if parent is not None: + of = parent.get("of") if isinstance(parent, dict) else None + if not (isinstance(parent, dict) and isinstance(of, str) and of and isinstance(parent.get("key"), list) + and parent["key"]): + add(f"{who}的下钻要写清是哪一项里的哪一段：上一层分析项的标识和那一段的维度值", + f"{at}.params.parent", "analysis.parent_invalid") + elif seen.get(of, i) >= i: + add(f"{who}下钻的上一层「{of}」要是这张分析卡里排在它前面的一项", f"{at}.params.parent", + "analysis.parent_invalid") + elif (items[seen[of]] or {}).get("op") != "decompose": + add(f"{who}下钻的上一层「{of}」不是变化拆解", f"{at}.params.parent", "analysis.parent_invalid") + overrides = cfg.get("overrides") + if overrides not in (None, {}): + label = field_label("overrides", NodeType.ANALYSIS) + if not isinstance(overrides, dict): + add(f"「{label}」需要是一个对象", "overrides", "analysis.overrides_invalid") + elif problem := contract.settings_problem(overrides): + add(f"「{label}」有误：{problem}", "overrides", "analysis.overrides_invalid") + return out + + +def _check_analysis(node: GraphNode, spec: GraphSpec, result: ValidationResult) -> None: + for issue in analysis_issues(spec, node): + result.add(issue.message, level=issue.level, node_id=issue.node_id, field=issue.field, code=issue.code) + + #: 报告撰写节点的结论句策略（claims）。off：结论句不参与判档（默认，保持以前的行为）； #: require_citation：没挂引用的结论句计入缺口、按出具档位降档。这两档是确定性的（发布前自动修复 #: 给人选的也只有这两档） @@ -989,6 +1152,8 @@ def validate_graph(spec: GraphSpec) -> ValidationResult: _check_report_claims(node, spec, result) elif node.type == NodeType.MERGE: _check_merge(node, spec, result) + elif node.type == NodeType.ANALYSIS: + _check_analysis(node, spec, result) elif node.type == NodeType.CHALLENGE: _check_challenge(node, spec, result) elif node.type == NodeType.OUTPUT: diff --git a/backend/app/engine/upgrade.py b/backend/app/engine/upgrade.py index 2bb86478..f7eb6b45 100644 --- a/backend/app/engine/upgrade.py +++ b/backend/app/engine/upgrade.py @@ -51,9 +51,9 @@ LEVELS = ("published", "governed") _LLM_ONLY = ("prompt", "system", "skills", "use_history", "output_schema") #: 新的报告撰写节点照搬过去的模型配置（R3 从 agent 带过去的只有 provider、model：温度、上限是查数用的） _MODEL_KEYS = ("provider", "model") -#: 报告的指令里照搬了这几类节点的文字，就要提醒：报告撰写自己会收集它们的证据 +#: 报告的指令里照搬了这几类节点的文字，就要提醒：报告撰写自己会收集它们的证据（分析卡的结论进台账，写作目录编成 A{n}） _EVIDENCE_TYPES = (NodeType.METRICS, NodeType.TOOL, NodeType.AGENT, NodeType.RETRIEVE, NodeType.CODE, - NodeType.SUBGRAPH, NodeType.MERGE) + NodeType.SUBGRAPH, NodeType.MERGE, NodeType.ANALYSIS) #: 写整段文字的节点：出具契约的 narrative 拼了其中几个的文字，就说不准该换哪一个 _WRITER_TYPES = (NodeType.LLM, NodeType.AGENT) _QUESTION = re.compile(r"(? VariableReport: f"nodes.{node.id}", kind="node", label=f"「{label}」这一步的输出", produced_by=node.id, produced_by_label=label, order=order, ) + # 分析卡每一项的结论（engine/nodes/analysis.py 的 card_view）：条件分支按判定走、成果取判定语都从这里取 + if node.type == "analysis" and isinstance(cfg.get("items"), list): + for item in cfg["items"]: + item_id = item.get("id") if isinstance(item, dict) else None + if isinstance(item_id, str) and item_id: + ensure( + f"nodes.{node.id}.results.{item_id}", kind="node", + label=f"「{label}」分析项「{item_id}」的结论", + produced_by=node.id, produced_by_label=label, order=order, + description="status 状态、verdict 判定、direction 方向、values 关键数、phrase 判定语", + ) # 每个根命名空间本身也是合法引用：{{ input }} 取整个输入 dict， # {{ vars }} 取整个变量池——template_context 里它们就是顶层的键。 @@ -345,6 +356,17 @@ def analyze(spec: GraphSpec) -> VariableReport: node_id=node.id, node_label=label, field=field, expr=m.group(0), )) + # 分析卡的输入同样按节点 id 选（主输入、两份查询、总数两格落在的节点）：记成对上游节点的引用 + if node.type == "analysis" and isinstance(cfg.get("items"), list): + from app.engine.schema import analysis_refs + + for i, item in enumerate(cfg["items"]): + for field, source in analysis_refs(item) if isinstance(item, dict) else []: + target = index.get(f"nodes.{source}") + if target is not None: + target.refs.append(VarRef(node_id=node.id, node_label=label, field=f"items[{i}].{field}", + expr=f"nodes.{source}")) + # 合并查询的输入按节点 id 选，不写模板：不认这一条的话，变量表里看不出上游那条查询被谁用了。 # 选错的节点（不存在、不在上游、不是查询）由 validate_graph 报，这里只记引用 if node.type == "merge" and isinstance(cfg.get("inputs"), dict): diff --git a/backend/app/plain.py b/backend/app/plain.py index 6c38d3a4..0058ff60 100644 --- a/backend/app/plain.py +++ b/backend/app/plain.py @@ -93,6 +93,13 @@ PLAIN: dict[str, Plain] = { "给口径卡供数的 Agent 没核对字段出处", "口径卡读到的仍是模型给的数。", _A, "打开按出处核对字段。"), "governed.caliber_model_input": Plain( "口径卡的数来自「模型调用」节点", "这种节点不调工具，写出的数没有出处。", _A, "改由「调用工具」节点或配好结构化输出的 Agent 取数。"), + # 分析卡（engine/nodes/analysis.py，docs/design/analysis-round/） + "governed.caliber_analysis_input": Plain( + "口径卡的数来自分析卡", "分析结论是叙述的依据，不是口径的原料，口径卡读到的数追不到原始数据。", _A, + "口径卡里按格从查询结果取数，分析结论在报告里直接引用。"), + "governed.analysis_input": Plain( + "分析卡分析的不是数据库查询的结果", "受管级别要求发布时就能确定分析的是平台执行的查询，否则结论说不清出处。", _A, + "把分析卡这一项的输入改选调用数据库查询工具的「调用工具」节点或「合并查询」。"), # ---------------------------------------------------------------- 写外部系统（engine/lint_capabilities.py） "action.write_in_agent": Plain( "Agent 的工具里有会写外部系统的能力", "写入只能放在「调用工具」节点里，先确认再发送、留生效记录。", _Y, diff --git a/backend/tests/test_analysis_card.py b/backend/tests/test_analysis_card.py new file mode 100644 index 00000000..0aae7593 --- /dev/null +++ b/backend/tests/test_analysis_card.py @@ -0,0 +1,419 @@ +"""分析卡（节点类型 analysis，docs/design/analysis-round/00-总方案.md §3.2）：画图时的校验、执行、产出的变量、受管规则、 +登记处（叫法、变量表、助手查看、目录影响面、升级提醒）。运行对比按项对齐见 tests/test_analysis_compare.py。 + +算子（app.analysis.run）换成按请求交回假结论的替身（tests/fixtures/analysis_ops.FakeOps）。库是虚构的零售库 shop。 +""" +from __future__ import annotations + +import asyncio +import sqlite3 + +import pytest +from sqlalchemy import select + +from app.analysis import contract +from app.analysis.contract import CallError +from app.core import artifact_store +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 +from app.engine.schema import GraphSpec, NodeType, validate_graph +from tests.fixtures.analysis_ops import FakeOps + +TOOL = "db_query__shop" +WEEKLY = "SELECT week, SUM(amount) AS gmv FROM orders GROUP BY week ORDER BY week" +REGION = "SELECT region, week, SUM(amount) AS gmv FROM orders GROUP BY region, week ORDER BY region, week" +ROWS = [(1, "华东", "2026-W41", 820.0), (2, "华东", "2026-W42", 700.0), (3, "华南", "2026-W41", 700.0), + (4, "华南", "2026-W42", 666.0), (5, "华北", "2026-W41", 600.0), (6, "华北", "2026-W42", 560.0)] +WOW = {"id": "wow", "op": "baseline", "from": "weekly", "params": {"period": "week", "measure": "gmv", + "label": "销售额"}} +WHERE = {"id": "where", "op": "decompose", "mode": "additive", "from": "by_region", + "params": {"by": ["region"], "period": "week", "base": "2026-W41", "target": "2026-W42", "measure": "gmv", + "headline": {"base": "weekly.r0.gmv", "target": "weekly.r1.gmv"}, "label": "销售额"}, + "hypothesis": "下降是不是集中在某个地区"} +DRILL = {"id": "east", "op": "decompose", "mode": "additive", "from": "by_region", + "params": {"by": ["week"], "period": "week", "base": "2026-W41", "target": "2026-W42", "measure": "gmv", + "parent": {"of": "where", "key": ["华东"]}}} + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +async def _shop(tmp_path, options=None): + path = tmp_path / "shop.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, week TEXT, amount REAL);") + db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", ROWS) + 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, options or {} + await session.commit() + source_id = row.id + await engines.invalidate(source_id) + return source_id + + +@pytest.fixture +async def shop(tmp_path): + source_id = await _shop(tmp_path) + 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 edge(a, b, handle=None): + return {"source": a, "target": b, **({"sourceHandle": handle} if handle else {})} + + +def card_graph(items=(WOW, WHERE), *, extra_nodes=(), extra_edges=(), **card): + nodes = [node("start", "input"), + node("weekly", "tool", tool=TOOL, args={"sql": WEEKLY}), + node("by_region", "tool", tool=TOOL, args={"sql": REGION}), + node("card", "analysis", items=[dict(i) if isinstance(i, dict) else i for i in items], **card), + node("out", "output", fields=[{"name": "判定", "value": "{{ nodes.card.results.wow.phrase }}"}]), + *extra_nodes] + edges = [edge("start", "weekly"), edge("weekly", "by_region"), edge("by_region", "card"), edge("card", "out"), + *extra_edges] + return {"nodes": nodes, "edges": edges} + + +def card_issues(graph, level="error"): + return [i for i in validate_graph(GraphSpec.model_validate(graph)).issues + if i.node_id == "card" and (level is None or i.level == level)] + + +async def wait(run_id, statuses=("succeeded", "failed")): + 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 statuses and 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()] + + +async def run_graph(graph, **kw): + run = await run_manager.start(graph=graph, input_payload={}, **kw) + return await wait(run.id) + + +async def output_of(run_id, node_id): + [finished] = await events(run_id, "node.finished", node_id) + return finished, artifact_store.load(finished["artifact"]) + + +# -------------------------------------------------------------------------- +# 画图时的校验 +# -------------------------------------------------------------------------- + + +def test_node_type_and_labels(): + from app.engine.labels import field_label, type_label + + assert NodeType.ANALYSIS == contract.NODE_TYPE == "analysis" + assert type_label("analysis") == "分析卡" + assert field_label("items", "analysis") == "分析项" and field_label("overrides", "analysis") == "阈值覆盖" + + +def test_a_well_formed_card_validates(): + assert card_issues(card_graph(items=(WOW, WHERE, DRILL), overrides={"main_share": 0.6}), level=None) == [] + + +@pytest.mark.parametrize("items, field, code, why", [ + ([], "items", "analysis.items_missing", "至少写一项"), + (["wow"], "items[0]", "analysis.item_invalid", "需要是一个对象"), + ([{**WOW, "id": "1st"}], "items[0].id", "analysis.item_invalid", "字母、数字和下划线"), + ([WOW, {**WHERE, "id": "wow"}], "items[1].id", "analysis.item_invalid", "唯一"), + ([{**WOW, "op": "forecast"}], "items[0].op", "analysis.op_invalid", "不认识"), + ([{**WHERE, "mode": None}], "items[0].mode", "analysis.mode_invalid", "要选一种模式"), + ([{**WOW, "mode": "additive"}], "items[0].mode", "analysis.mode_invalid", "没有模式可选"), + ([{**WOW, "from": None}], "items[0].from", "analysis.input_invalid", "还没有选择输入"), + ([{**WOW, "from": "nope"}], "items[0].from", "analysis.input_invalid", "不存在"), + ([{**WOW, "params": {**WOW["params"], "k_warn": 3}}], "items[0].params.k_warn", "analysis.params_invalid", + "阈值覆盖"), + ([{**WOW, "params": {**WOW["params"], "query": "weekly"}}], "items[0].params.query", "analysis.params_invalid", + "输入里"), + ([{**WOW, "params": {**WOW["params"], "bogus": 1}}], "items[0].params.bogus", "analysis.params_invalid", + "没有「bogus」"), + ([{**WHERE, "params": {**WHERE["params"], "headline": {"base": "weekly.gmv", "target": "weekly.r1.gmv"}}}], + "items[0].params.headline.base", "analysis.params_invalid", "节点 id.r行号.列名"), + ([{**WHERE, "params": {**WHERE["params"], "headline": "weekly.r0.gmv"}}], "items[0].params.headline", + "analysis.params_invalid", "两格"), + ([DRILL, WHERE], "items[0].params.parent", "analysis.parent_invalid", "排在它前面"), + ([WOW, {**DRILL, "params": {**DRILL["params"], "parent": {"of": "wow", "key": ["华东"]}}}], + "items[1].params.parent", "analysis.parent_invalid", "不是变化拆解"), + ([WHERE, {**DRILL, "params": {**DRILL["params"], "parent": {"of": "where"}}}], "items[1].params.parent", + "analysis.parent_invalid", "写清是哪一项里的哪一段"), + ([WHERE, {**WOW, "params": {**WOW["params"], "parent": {"of": "where", "key": ["华东"]}}}], + "items[1].params.parent", "analysis.params_invalid", "没有「parent」"), +]) +def test_item_problems(items, field, code, why): + found = [i for i in card_issues(card_graph(items=items)) if i.field == field] + assert found and found[0].code == code and why in found[0].message, [i.message for i in card_issues( + card_graph(items=items))] + + +@pytest.mark.parametrize("extra, why", [ + (node("think", "llm", prompt="写点什么"), "不产出查询结果"), + (node("asker", "agent", prompt="查一下", tools=[TOOL]), "Agent 可能查询多次"), +]) +def test_inputs_must_be_query_nodes(extra, why): + graph = card_graph(items=[{**WOW, "from": extra["id"]}], extra_nodes=[extra], + extra_edges=[edge("by_region", extra["id"]), edge(extra["id"], "card")]) + [issue] = card_issues(graph) + assert issue.field == "items[0].from" and why in issue.message and issue.code == "analysis.input_invalid" + + +def test_inputs_must_be_upstream_and_other_tools_only_warn(): + lonely = node("lonely", "tool", tool=TOOL, args={"sql": WEEKLY}) + [issue] = card_issues(card_graph(items=[{**WOW, "from": "lonely"}], extra_nodes=[lonely], + extra_edges=[edge("start", "lonely")])) + assert "不在分析卡的上游" in issue.message + # headline 那两格落在的节点同样要是上游的查询节点 + headline = {**WHERE["params"], "headline": {"base": "lonely.r0.gmv", "target": "weekly.r1.gmv"}} + [issue] = card_issues(card_graph(items=[{**WHERE, "params": headline}], extra_nodes=[lonely], + extra_edges=[edge("start", "lonely")])) + assert issue.field == "items[0].params.headline.base" and "不在分析卡的上游" in issue.message + # 调用别的工具的调用工具节点：登记里声明了表格的端点读取也交回快照，静态看不出，只提醒 + other = node("ext", "tool", tool="crm_report", args={}) + graph = card_graph(items=[{**WOW, "from": "ext"}], extra_nodes=[other], + extra_edges=[edge("by_region", "ext"), edge("ext", "card")]) + assert card_issues(graph) == [] + [warning] = card_issues(graph, level="warning") + assert warning.code == "analysis.input_invalid" and "不是数据库查询工具" in warning.message + + +@pytest.mark.parametrize("overrides, why", [ + ({"k_warn": 3.0, "k_alert": 2.5}, "要大于"), + ({"main_share": 2}, "之间"), + ({"bogus": 1}, "不认识的分析设置"), + ("tight", "需要是一个对象"), +]) +def test_overrides_are_checked_by_the_contract(overrides, why): + [issue] = card_issues(card_graph(overrides=overrides)) + assert issue.field == "overrides" and issue.code == "analysis.overrides_invalid" and why in issue.message + + +def test_two_snapshot_items_need_both_nodes(): + two = {"id": "same", "op": "comparable", "params": {"base_query": "weekly", "target_query": "by_region"}} + assert card_issues(card_graph(items=[two])) == [] + half = {"id": "same", "op": "comparable", "params": {"base_query": "weekly"}} + [issue] = card_issues(card_graph(items=[half])) + assert issue.field == "items[0].from" and "两个查询节点" in issue.message + + +# -------------------------------------------------------------------------- +# 执行 +# -------------------------------------------------------------------------- + + +async def test_the_card_runs_each_item_and_records_it(monkeypatch, shop): + ops = FakeOps(monkeypatch, verdicts=["outside_band", "driver", "driver"], peek=["wow", "where", "east"]) + row = await run_graph(card_graph(items=(WOW, WHERE, DRILL), overrides={"main_share": 0.6}), + options={contract.SETTINGS_KEY: {"k_warn": 2.5}}) + assert row.status == "succeeded", row.error + + # 算子拿到的：请求、按上游节点 id 解析的快照（只认台账里的那几份）、阈值（缺省 ← 发起时的快照 ← 卡上的覆盖） + wow, where, east = ops.calls + assert wow["request"] == {"op": "baseline", "mode": None, + "params": {"period": "week", "measure": "gmv", "label": "销售额", "query": "weekly"}} + assert where["request"]["hypothesis"] == "下降是不是集中在某个地区" and where["request"]["mode"] == "additive" + assert where["request"]["params"]["headline"] == {"base": "weekly.r0.gmv", "target": "weekly.r1.gmv"} + weekly = wow["snapshots"]["weekly"] + [tool_end] = await events(row.id, "tool.end", "weekly") + assert weekly["ref"] == "weekly" and weekly["node_id"] == "weekly" and weekly["artifact"] == tool_end["query_artifact"] + assert weekly["columns"] == ["week", "gmv"] and weekly["rows"] == [["2026-W41", 2120.0], ["2026-W42", 1926.0]] + assert weekly["sql"] == WEEKLY and weekly["source"] == "shop" and weekly["masked"] == [] + assert wow["settings"] == {**contract.DEFAULTS, "k_warn": 2.5, "main_share": 0.6} + # 下钻取到同卡里前一项的结论；只看得到排在前面、已经做完的项 + assert east["parent"]["op"] == "decompose" and east["parent"]["params"]["query"] == "by_region" + assert east["visible"] == {"wow": True, "where": True, "east": False} + assert wow["visible"] == {"wow": False, "where": False, "east": False} + + finished, output = await output_of(row.id, "card") + results = output["results"] + assert set(results) == {"wow", "where", "east"} # 产出工件按规范 JSON 存，键的先后看配置里项的先后 + assert results["wow"]["verdict"] == "outside_band" and results["wow"]["status"] == "ok" + assert results["wow"]["phrase"].startswith("销售额本期变化") and results["wow"]["direction"] == "down" + assert results["where"]["values"]["top_share"] == "0.6186" + # 台账：每项一条 kind analysis（带 item），工件就是整份结论 + entries = finished["evidence"] + assert [(e["kind"], e["item"]) for e in entries] == [("analysis", "wow"), ("analysis", "where"), + ("analysis", "east")] + for entry, item in zip(entries, ("wow", "where", "east")): + stored = artifact_store.load(entry["artifact"]) + assert entry == contract.ledger_entry(stored, node_id="card", exec_no=1, artifact=entry["artifact"], + item=item) + assert results[item]["artifact"] == entry["artifact"] + # 每项一条 analysis.end + ends = await events(row.id, "analysis.end", "card") + assert [(e["item"], e["op"], e["verdict"], e["analysis_artifact"]) for e in ends] == [ + (e["item"], {"wow": "baseline"}.get(e["item"], "decompose"), results[e["item"]]["verdict"], + results[e["item"]]["artifact"]) for e in ends] + # 成果字段取判定语 + assert row.output["判定"] == results["wow"]["phrase"] + + +async def test_a_refused_item_does_not_fail_the_card(monkeypatch, shop): + FakeOps(monkeypatch, statuses=["refused", "ok"]) + row = await run_graph(card_graph()) + assert row.status == "succeeded", row.error + _, output = await output_of(row.id, "card") + assert output["results"]["wow"]["status"] == "refused" and output["results"]["wow"]["verdict"] is None + assert output["results"]["wow"]["phrase"].endswith("历史只有 5 期，至少要 8 期") + assert output["results"]["where"]["status"] == "ok" + [warn] = [e for e in await events(row.id, "log", "card") if e.get("code") == "analysis_refused"] + assert warn["item"] == "wow" and warn["reason_code"] == "history_short" and "wow" in warn["message"] + + +async def test_a_call_error_fails_the_card_and_names_the_item(monkeypatch, shop): + FakeOps(monkeypatch, errors=[None, CallError("ratio_as_additive", "比率不能按段相加，改用 ratio 模式，给分子、分母两列")]) + row = await run_graph(card_graph()) + assert row.status == "failed" + assert "分析项「where」（变化拆解）" in row.error and "比率不能按段相加" in row.error and "改正这一项" in row.error + + +async def test_a_branch_can_follow_the_verdict(monkeypatch, shop): + FakeOps(monkeypatch, verdicts=["outside_band", "driver"]) + graph = card_graph(extra_nodes=[ + node("route", "branch", cases=[{"key": "alert", "condition": "nodes.card.results.wow.verdict == 'outside_band'"}]), + node("alert", "output", fields=[{"name": "转人工", "value": "{{ nodes.card.results.wow.phrase }}"}])]) + graph["edges"] = [e for e in graph["edges"] if e["target"] != "out"] + [ + edge("card", "route"), edge("route", "alert", "alert"), edge("route", "out", "default")] + row = await run_graph(graph) + assert row.status == "succeeded", row.error + assert "转人工" in row.output + + +async def test_inputs_that_are_not_query_snapshots_fail_clearly(monkeypatch, shop): + FakeOps(monkeypatch) + other = node("calc", "tool", tool="json_query", args={"data": '{"a": 1}', "path": "a"}) + graph = card_graph(items=[{**WOW, "from": "calc"}], extra_nodes=[other], + extra_edges=[edge("by_region", "calc"), edge("calc", "card")]) + row = await run_graph(graph) + assert row.status == "failed" and "不是存成快照的查询结果" in row.error + + +async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path): + source_id = await _shop(tmp_path, {"mask_columns": ["region"]}) + try: + ops = FakeOps(monkeypatch) + row = await run_graph(card_graph(items=[WHERE])) + assert row.status == "succeeded", row.error + assert "region" in ops.calls[0]["snapshots"]["by_region"]["masked"] + finally: + await engines.invalidate(source_id) + + +# -------------------------------------------------------------------------- +# 受管规则、登记处 +# -------------------------------------------------------------------------- + + +def _governed(graph): + from app.engine.governance import lint_for_publish + + return lint_for_publish(GraphSpec.model_validate(graph), level="governed").issues + + +def test_governed_cards_must_analyze_platform_queries(): + other = node("ext", "tool", tool="crm_report", args={}) + graph = card_graph(items=[{**WOW, "from": "ext"}], extra_nodes=[other], + extra_edges=[edge("by_region", "ext"), edge("ext", "card")]) + [issue] = [i for i in _governed(graph) if i.code == "governed.analysis_input"] + assert issue.level == "error" and issue.node_id == "card" and issue.field == "items[0].from" + assert [i for i in _governed(card_graph()) if i.code == "governed.analysis_input"] == [] + # 发布时：受管级按错误挡、画布那条同义的提醒不再列；已发布级只有画布那条提醒 + from app.engine.governance import publish_issues + + spec = GraphSpec.model_validate(graph) + governed = [(i.code, i.level) for i in publish_issues(spec, level="governed") if i.node_id == "card"] + assert ("governed.analysis_input", "error") in governed and ("analysis.input_invalid", "warning") not in governed + published = [(i.code, i.level) for i in publish_issues(spec, level="published") if i.node_id == "card"] + assert published == [("analysis.input_invalid", "warning")] + + +def test_a_caliber_card_reading_analysis_results_gets_a_hint(): + from app.engine.governance import _traceable + + graph = card_graph(extra_nodes=[node("kpi", "metrics", metrics=[ + {"id": "share", "expression": "nodes.card.results.where.values.top_share"}])], + extra_edges=[edge("card", "kpi")]) + [issue] = [i for i in _governed(graph) if i.code == "governed.caliber_analysis_input"] + assert issue.level == "warning" and issue.node_id == "card" and "追不到原始数据" in issue.message + assert _traceable(GraphSpec.model_validate(graph).node_map()["card"]) + + +def test_plain_wording_for_the_new_codes(): + from app.plain import plain_of + + for code in ("governed.analysis_input", "governed.caliber_analysis_input"): + assert plain_of(code) is not None, code + + +async def test_a_caliber_card_reading_an_analysis_card_is_untraced(monkeypatch, shop): + from app.engine.lineage import untraced + + FakeOps(monkeypatch) + graph = card_graph(extra_nodes=[node("kpi", "metrics", metrics=[ + {"id": "share", "expression": "float(nodes.card.results.where.values.top_share)"}])], + extra_edges=[edge("card", "kpi")]) + row = await run_graph(graph) + assert row.status == "succeeded", row.error + _, card = await output_of(row.id, "kpi") + [item] = card["metrics"][0]["inputs"] + assert item["via"] == "analysis" and item["node_id"] == "card" and untraced([item]) == [item] + + +def test_variables_list_each_item_and_count_the_inputs_as_references(): + from app.engine.variables import analyze + + report = analyze(GraphSpec.model_validate(card_graph(items=(WOW, WHERE)))) + found = {v.path: v for v in report.variables} + assert {"nodes.card.results.wow", "nodes.card.results.where"} <= set(found) + assert found["nodes.card.results.wow"].produced_by == "card" + assert any(r.node_id == "card" and r.field == "items[0].from" for r in found["nodes.weekly"].refs) + assert any(r.node_id == "card" and r.field == "items[1].params.headline.base" for r in found["nodes.weekly"].refs) + assert not [i for i in report.issues if i.level != "info"] + + +def test_describe_impact_and_upgrade_know_the_card(): + from types import SimpleNamespace + + from app.apps.describe import describe_version + from app.data.catalog_impact import template_refs + from app.engine.upgrade import _EVIDENCE_TYPES + + graph = card_graph(items=(WOW, WHERE)) + text = describe_version(SimpleNamespace(graph=graph, version=2, graph_hash="ab" * 32, level="published"), + workflow_name="区域周报") + assert "分析卡「card」（card），分析项 2 个" in text and "card.results.where 变化拆解" in text + refs = template_refs(GraphSpec.model_validate(graph), "shop", "orders", None) + [card] = [r for r in refs if r["node_id"] == "card"] + assert card["type"] == "analysis" and card["impact"] == "direct" + assert {(v["node_id"], v["alias"]) for v in card["via"]} == {("weekly", "wow"), ("by_region", "where"), + ("weekly", "where")} + assert NodeType.ANALYSIS in _EVIDENCE_TYPES