From 46eb674bc20e08d930ea6077224c754e5999a9e3 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 14 Sep 2026 10:17:45 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8C=87=E6=A0=87=E9=9B=86=E4=B8=8E=E5=87=BA?= =?UTF-8?q?=E5=85=B7=E5=A5=91=E7=BA=A6=EF=BC=9A=E5=8F=A3=E5=BE=84=E5=8D=A1?= =?UTF-8?q?=E8=8A=82=E7=82=B9=E3=80=81=E5=8F=99=E8=BF=B0=E6=95=B0=E5=AD=97?= =?UTF-8?q?=E5=9B=9E=E6=8C=87=E3=80=81=E4=B8=89=E6=A1=A3=E5=87=BA=E5=85=B7?= =?UTF-8?q?=E3=80=81=E5=91=A8=E6=8A=A5=E6=A8=A1=E6=9D=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 派生边界收死在固定层：新增 metrics 节点（口径卡），指标由 AST 白名单 求值器确定性计算，产出受控指标集（含口径名与版本）。叙述层没有算术 权限——不是求模型别编数，而是它编了也过不了校验。 output 节点配 contract 即升级为出具契约：核对必需/期望指标是否齐、 从叙述中抽取所有数字逐个回指指标集（支持千分位、百分数↔比率、 万/亿后缀、按书写精度给舍入容差；日期、ISO 周、枚举序号、引用标记 不参与回指），三档判定 formal / degraded / withheld，缺口清单和 口径版本一起印在成果上，另发 issuance 事件入流水。白名单（allow_numbers） 是模板配置的一部分，进版本、被审查。 新增模板⑧周报：取数固定 → 口径卡算数 → LLM 只许引用清单 → 出具 逐数回指，是'模板定骨架、agent 填关节'的完整示范。 回指校验器 10 用例 + 三档判定 + 端到端五场景全部通过。 --- backend/app/engine/compiler.py | 3 +- backend/app/engine/issuance.py | 175 +++++++++++++++++++++++++++ backend/app/engine/nodes/__init__.py | 2 +- backend/app/engine/nodes/io.py | 79 +++++++++++- backend/app/engine/nodes/metrics.py | 78 ++++++++++++ backend/app/engine/schema.py | 14 +++ backend/app/seed.py | 52 ++++++++ 7 files changed, 400 insertions(+), 3 deletions(-) create mode 100644 backend/app/engine/issuance.py create mode 100644 backend/app/engine/nodes/metrics.py diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index 5f6ea4c9..25ea30be 100644 --- a/backend/app/engine/compiler.py +++ b/backend/app/engine/compiler.py @@ -9,7 +9,7 @@ from langgraph.graph import END, START, StateGraph from app.core.events import EventType from app.engine.context import NodeContext, NodeError, RunContext -from app.engine.nodes import control, human, io, knowledge, llm, multi, tools +from app.engine.nodes import control, human, io, knowledge, llm, metrics, multi, tools from app.engine.schema import GraphNode, GraphSpec, NodeType from app.engine.state import GraphState @@ -31,6 +31,7 @@ RUNNERS: dict[NodeType, NodeRunner] = { NodeType.HUMAN: human.run_human, NodeType.VALIDATE: human.run_validate, NodeType.MEMORY: knowledge.run_memory, + NodeType.METRICS: metrics.run_metrics, NodeType.RETRIEVE: knowledge.run_retrieve, } diff --git a/backend/app/engine/issuance.py b/backend/app/engine/issuance.py new file mode 100644 index 00000000..08ccb13f --- /dev/null +++ b/backend/app/engine/issuance.py @@ -0,0 +1,175 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Any + +# -------------------------------------------------------------------------- +# 出具校验：叙述里的每个数字必须能回指到指标集 +# +# 前提是派生边界已经收死：所有算术都发生在口径卡（metrics 节点）里， +# 叙述层只能引用。所以这里不需要"验证模型算得对不对"——只需要验证 +# "它写的每个数都在清单里"。集合外的数字没有合法来源，一律视为现编。 +# -------------------------------------------------------------------------- + +# 数字 token：可带千分位逗号、小数，后缀 %/％ 或 万/亿 +_NUM_RE = re.compile(r"(? bool: + return not self.unmatched + + +def _skip_spans(text: str) -> list[tuple[int, int]]: + spans: list[tuple[int, int]] = [] + for pattern in _SKIP_PATTERNS: + for m in pattern.finditer(text): + spans.append(m.span()) + return spans + + +def extract_numbers(text: str) -> list[NumberToken]: + """抽出叙述里所有需要回指的数字，剔除日期、枚举序号、引用标记这类形态。""" + skip = _skip_spans(text) + tokens: list[NumberToken] = [] + for m in _NUM_RE.finditer(text): + start, end = m.span() + if any(s <= start < e for s, e in skip): + continue + raw_num = m.group(1).replace(",", "") + suffix = m.group(2) or "" + try: + value = float(raw_num) + except ValueError: + continue + if suffix in _SUFFIX_MULT: + value *= _SUFFIX_MULT[suffix] + decimals = len(raw_num.split(".")[1]) if "." in raw_num else 0 + ctx_lo = max(0, start - 18) + tokens.append( + NumberToken( + raw=m.group(0).strip(), + value=value, + decimals=decimals, + is_percent=suffix in ("%", "％"), + context=text[ctx_lo : min(len(text), end + 18)].replace("\n", " "), + start=start, + ) + ) + return tokens + + +def _close(a: float, b: float, decimals: int) -> bool: + """按 token 自己写出的小数位数给舍入容差：写了 12.3 就允许 |Δ|≤0.05。""" + tolerance = 0.5 * 10 ** (-decimals) + return abs(a - b) <= tolerance + 1e-9 + + +def _token_matches(token: NumberToken, metric_value: float) -> bool: + # 直接比对（含"叙述写百分数、指标本身就是百分数"的情形） + if _close(token.value, metric_value, token.decimals): + return True + if token.is_percent: + # 指标存的是比率（0.123），叙述写成 12.3% + if _close(token.value, metric_value * 100, token.decimals): + return True + return False + + +def trace_numbers( + narrative: str, + metrics: list[dict[str, Any]], + *, + allow: list[Any] | None = None, +) -> TraceReport: + """叙述 vs 指标集的回指校验。 + + allow 是模板作者显式放行的字面量（比如报告标题里的期数）—— + 白名单是配置的一部分，进版本、被审查，而不是校验器私藏的宽容。 + """ + numeric_values = [ + float(m["value"]) + for m in metrics + if isinstance(m.get("value"), (int, float)) and not isinstance(m.get("value"), bool) + ] + allowed = {str(a).strip() for a in (allow or [])} + allowed_values = set() + for a in allowed: + try: + allowed_values.add(float(a.replace(",", "").rstrip("%％"))) + except ValueError: + pass + + report = TraceReport() + for token in extract_numbers(narrative): + if token.raw in allowed or token.raw.rstrip("%％").strip() in allowed: + continue + if any(_close(token.value, av, token.decimals) for av in allowed_values): + continue + hit = next( + ( + m + for m, mv in zip( + [m for m in metrics if isinstance(m.get("value"), (int, float))], + numeric_values, + ) + if _token_matches(token, mv) + ), + None, + ) + if hit is not None: + report.matched.append({"token": token.raw, "metric": hit.get("id")}) + else: + report.unmatched.append({"token": token.raw, "context": token.context}) + return report + + +def decide_tier( + *, + missing_required: list[str], + missing_expected: list[str], + unmatched: list[dict[str, Any]], + strict: bool, +) -> str: + """三档出具的判定。 + + withheld（不予出具）：必需指标缺失，或 strict 下有未回指数字—— + 结论的地基不完整，宁可不出。 + degraded（降档出具）：期望指标缺失或有未回指数字，出但明确声明缺口。 + formal（正式出具）：清单齐、数字全部可回指。 + """ + if missing_required: + return "withheld" + if strict and unmatched: + return "withheld" + if missing_expected or unmatched: + return "degraded" + return "formal" diff --git a/backend/app/engine/nodes/__init__.py b/backend/app/engine/nodes/__init__.py index b0c0d711..3f95eee6 100644 --- a/backend/app/engine/nodes/__init__.py +++ b/backend/app/engine/nodes/__init__.py @@ -1 +1 @@ -from app.engine.nodes import control, human, io, knowledge, llm, multi, tools # noqa: F401 +from app.engine.nodes import control, human, io, knowledge, llm, metrics, multi, tools # noqa: F401 diff --git a/backend/app/engine/nodes/io.py b/backend/app/engine/nodes/io.py index 8eaf4f87..fcbf980e 100644 --- a/backend/app/engine/nodes/io.py +++ b/backend/app/engine/nodes/io.py @@ -32,7 +32,12 @@ async def run_input(state: GraphState, ctx: NodeContext) -> dict[str, Any]: async def run_output(state: GraphState, ctx: NodeContext) -> dict[str, Any]: - """出口节点。按配置把散落在各节点的结果收成一个结构化成果。""" + """出口节点。按配置把散落在各节点的结果收成一个结构化成果。 + + 配了 contract 就升级成出具契约：核对指标齐不齐、叙述里的数字 + 能不能回指到指标集，然后给出三档判定（formal / degraded / withheld）。 + 判定和缺口清单一起印在成果上——降档不是悄悄地降。 + """ mapping = ctx.cfg("fields", None) tctx = template_context(state) @@ -47,9 +52,81 @@ async def run_output(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 没配映射就把最后一条消息当作结果，让新建的图开箱即用 result = {"result": tctx.get("last_message", "")} + contract = ctx.cfg("contract") + if contract: + result["_issuance"] = _apply_contract(contract, state, ctx) + return {"output": result, "nodes": {ctx.node.id: result}} +def _apply_contract( + contract: dict[str, Any], state: GraphState, ctx: NodeContext +) -> dict[str, Any]: + from datetime import datetime, timezone + + from app.engine.issuance import decide_tier, trace_numbers + + nodes = state.get("nodes") or {} + + # 1) 收指标集：contract.metrics_from 指向一个或多个口径卡节点 + metrics: list[dict[str, Any]] = [] + calibers: list[dict[str, str]] = [] + sources = contract.get("metrics_from") or [] + if isinstance(sources, str): + sources = [sources] + for source in sources: + payload = nodes.get(source) + if isinstance(payload, dict) and payload.get("kind") == "metric_set": + metrics.extend(payload.get("metrics") or []) + calibers.append( + { + "node": source, + "caliber": payload.get("caliber", ""), + "version": payload.get("caliber_version", ""), + } + ) + + present = {m["id"] for m in metrics} + missing_required = [m for m in (contract.get("required") or []) if m not in present] + missing_expected = [m for m in (contract.get("expected") or []) if m not in present] + + # 2) 叙述数字回指 + narrative = ctx.render_str(str(contract.get("narrative", "")), state) + trace = ( + trace_numbers(narrative, metrics, allow=contract.get("allow_numbers")) + if narrative + else None + ) + unmatched = trace.unmatched if trace else [] + + tier = decide_tier( + missing_required=missing_required, + missing_expected=missing_expected, + unmatched=unmatched, + strict=bool(contract.get("strict")), + ) + + issuance = { + "tier": tier, + "calibers": calibers, # 口径版本照常印 + "metrics_checked": len(metrics), + "missing_required": missing_required, + "missing_expected": missing_expected, # 缺数据声明照常印 + "unmatched_numbers": unmatched[:20], + "matched_numbers": len(trace.matched) if trace else 0, + "declared_at": datetime.now(timezone.utc).isoformat(), + } + ctx.emit( + EventType.ISSUANCE, + tier=tier, + missing_required=missing_required, + missing_expected=missing_expected, + unmatched=len(unmatched), + calibers=calibers, + ) + return issuance + + async def run_transform(state: GraphState, ctx: NodeContext) -> dict[str, Any]: """数据整形。用受限表达式把上游输出揉成下游要的形状，不用为此专门起一个 LLM。""" tctx = template_context(state) diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py new file mode 100644 index 00000000..b58d9e28 --- /dev/null +++ b/backend/app/engine/nodes/metrics.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +from typing import Any + +from app.core.events import EventType +from app.engine.context import NodeContext, NodeError +from app.engine.expressions import ExpressionError, eval_expression +from app.engine.state import GraphState, template_context + + +async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: + """口径卡：把数值计算收进固定层，产出受控的指标集（MetricSet）。 + + 表达式走 AST 白名单求值器——确定性、无模型参与、可复算。 + 下游叙述节点只许引用这里登记过的数，出具校验器按这份清单抽数字回指。 + 这就是"叙述层无算术权限"的机制载体：不是求模型别算，而是它算了也过不了校验。 + """ + definitions = ctx.cfg("metrics", []) or [] + if not definitions: + raise NodeError(ctx.node.id, "口径卡没有定义任何指标") + + tctx = template_context(state) + caliber = ctx.render_str(ctx.cfg("caliber", "") or ctx.node.title, state) + caliber_version = str(ctx.cfg("caliber_version", "") or "v1") + + metrics: list[dict[str, Any]] = [] + errors: list[str] = [] + for definition in definitions: + metric_id = str(definition.get("id") or "").strip() + expr = str(definition.get("expression") or "").strip() + if not metric_id or not expr: + errors.append(f"指标定义不完整：{definition}") + continue + try: + value = eval_expression(expr, tctx) + except ExpressionError as e: + errors.append(f"{metric_id}: 表达式错误（{e}）") + continue + except ZeroDivisionError: + errors.append(f"{metric_id}: 除以零") + continue + decimals = definition.get("decimals") + if isinstance(value, float) and decimals is not None: + value = round(value, int(decimals)) + metrics.append( + { + "id": metric_id, + "name": definition.get("name") or metric_id, + "unit": definition.get("unit") or "", + "value": value, + "expression": expr, + } + ) + + if errors: + # 口径卡是固定层：任何一个指标算不出来都是模板缺陷，不能带病出具 + raise NodeError(ctx.node.id, "口径卡计算失败：" + "；".join(errors)) + + # text 是给叙述节点 prompt 用的清单——它是叙述层唯一的数字来源 + lines = [f"- {m['id']}（{m['name']}）= {m['value']}{m['unit']}" for m in metrics] + result = { + "kind": "metric_set", + "caliber": caliber, + "caliber_version": caliber_version, + "metrics": metrics, + "text": f"指标清单（口径 {caliber} @ {caliber_version}）：\n" + "\n".join(lines), + } + ctx.emit( + EventType.LOG, + level="info", + message=f"口径卡 {caliber}@{caliber_version}：产出 {len(metrics)} 个受控指标", + ) + + updates: dict[str, Any] = {"nodes": {ctx.node.id: result}} + var_name = ctx.cfg("assign_to", "") + if var_name: + updates["vars"] = {var_name: result} + return updates diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index d26547c2..aad91d7e 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -29,6 +29,7 @@ class NodeType(StrEnum): HUMAN = "human" # 人工介入 VALIDATE = "validate" # 结构化校验 + 自动重试 + METRICS = "metrics" # 口径卡：受控指标集，叙述层唯一合法的数字来源 class Position(BaseModel): @@ -195,6 +196,19 @@ def validate_graph(spec: GraphSpec) -> ValidationResult: elif node.type == NodeType.VALIDATE: if not cfg.get("schema"): result.add("校验节点需要一个 JSON Schema", node_id=node.id) + elif node.type == NodeType.METRICS: + defs = cfg.get("metrics") or [] + if not defs: + result.add("口径卡还没有定义指标", node_id=node.id) + for d in defs: + if not d.get("id") or not d.get("expression"): + result.add(f"指标定义缺 id 或 expression：{d.get('id') or '(空)'}", + node_id=node.id) + elif node.type == NodeType.OUTPUT: + contract = cfg.get("contract") + if contract and not contract.get("metrics_from"): + result.add("出具契约缺 metrics_from（指标来自哪个口径卡节点）", + node_id=node.id) elif node.type == NodeType.LOOP: if not spec.outgoing(node.id): result.add("循环节点没有循环体", node_id=node.id) diff --git a/backend/app/seed.py b/backend/app/seed.py index 6c2193bb..0d567837 100644 --- a/backend/app/seed.py +++ b/backend/app/seed.py @@ -173,6 +173,58 @@ TEMPLATES: list[dict[str, Any]] = [ ], "edges": [_e("start", "team"), _e("team", "done")], }, + { + "name": "⑧ 周报（口径卡 + 三档出具）", + "description": "模板定骨架、agent 填关节的完整示范：取数固定、口径卡算数、" + "叙述只许引用清单里的数、出具时逐数回指——现编的数字会被点名并降档。", + "tags": ["出具", "口径卡", "周报"], + "nodes": [ + _n("start", "input", "周期", fields=[ + {"name": "week", "default": "2026-W37", "description": "报告周期"}]), + _n("fetch", "code", "取数（数据底座）", language="python", network=False, + assign_to="agg", + code=( + "# 固定层：取数与汇总。真实场景换成对数据库/API 的查询，\n" + "# 结果会作为快照落工件库，出具溯源指回这里。\n" + "import json\n" + "print(json.dumps({'orders': 1234, 'amount': 45678.5,\n" + " 'amount_prev': 42010.0, 'returns': 29}))" + )), + _n("caliber", "metrics", "周报口径卡", + caliber="周报口径", caliber_version="v1", assign_to="m", + metrics=[ + {"id": "orders", "name": "订单数", "unit": "单", + "expression": "vars.agg.orders"}, + {"id": "amount", "name": "销售额", "unit": "元", + "expression": "vars.agg.amount"}, + {"id": "wow", "name": "环比增幅", "unit": "%", + "expression": "round((vars.agg.amount - vars.agg.amount_prev) / vars.agg.amount_prev * 100, 1)"}, + {"id": "return_rate", "name": "退货率", "unit": "%", + "expression": "round(vars.agg.returns / vars.agg.orders * 100, 2)"}, + ]), + _n("narrate", "llm", "叙述（关节层）", + system="你是周报撰写人。只允许引用下面指标清单里出现的数值，" + "一个清单外的数字都不要写；重点是挑哪些变化值得说、怎么归因。", + prompt="{{ nodes.caliber.text }}\n\n请为 {{ input.week }} 写一段 120 字以内的周报正文。", + assign_to="report"), + _n("done", "output", "出具", + fields=[ + {"name": "周报", "value": "{{ vars.report }}"}, + {"name": "指标清单", "value": "{{ nodes.caliber.text }}"}], + contract={ + "metrics_from": ["caliber"], + "narrative": "{{ vars.report }}", + "required": ["amount", "orders"], + "expected": ["wow", "return_rate"], + "allow_numbers": ["37", "120"], + "strict": False, + }), + ], + "edges": [ + _e("start", "fetch"), _e("fetch", "caliber"), + _e("caliber", "narrate"), _e("narrate", "done"), + ], + }, { "name": "⑦ 批量处理（循环）", "description": "对列表逐项处理再汇总。演示 loop 节点的 body / done 两个出口怎么接。",