指标集与出具契约：口径卡节点、叙述数字回指、三档出具、周报模板

派生边界收死在固定层：新增 metrics 节点（口径卡），指标由 AST 白名单求值器确定性计算，产出受控指标集（含口径名与版本）。叙述层没有算术权限——不是求模型别编数，而是它编了也过不了校验。 output 节点配 contract 即升级为出具契约：核对必需/期望指标是否齐、从叙述中抽取所有数字逐个回指指标集（支持千分位、百分数↔比率、万/亿后缀、按书写精度给舍入容差；日期、ISO 周、枚举序号、引用标记不参与回指），三档判定 formal / degraded / withheld，缺口清单和口径版本一起印在成果上，另发 issuance 事件入流水。白名单（allow_numbers）是模板配置的一部分，进版本、被审查。新增模板⑧周报：取数固定 → 口径卡算数 → LLM 只许引用清单 → 出具逐数回指，是'模板定骨架、agent 填关节'的完整示范。回指校验器 10 用例 + 三档判定 + 端到端五场景全部通过。
2026-11-15 00:11:59 +00:00 · 2026-09-14 10:17:45 +08:00
parent 4fdd58e3be
commit 46eb674bc2
7 changed files with 400 additions and 3 deletions
@@ -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,
 }

@@ -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"(?<![\w.])(-?\d[\d,]*(?:\.\d+)?)\s*(%|％|万亿|亿|万)?")
+
+# 不参与回指的数字形态——这些不是"结论数字"
+_SKIP_PATTERNS = [
+    re.compile(r"\d{4}\s*[-/年]\s*\d{1,2}(\s*[-/月]\s*\d{1,2}\s*日?)?"),  # 日期
+    re.compile(r"(?<!\d)(19|20)\d{2}\s*年?(?!\d)"),  # 年份
+    re.compile(r"\d{4}-W\d{1,2}", re.IGNORECASE),  # ISO 周（2026-W37）
+    re.compile(r"^\s*\d{1,2}[.、)]\s", re.MULTILINE),  # 行首枚举 "1. " "2、"
+    re.compile(r"\[\d+\]"),  # 引用标记 [1]
+    re.compile(r"\d{1,2}:\d{2}(:\d{2})?"),  # 时间
+]
+
+_SUFFIX_MULT = {"万": 1e4, "亿": 1e8, "万亿": 1e12}
+
+
+@dataclass
+class NumberToken:
+    raw: str
+    value: float
+    decimals: int
+    is_percent: bool
+    context: str
+    start: int
+
+
+@dataclass
+class TraceReport:
+    matched: list[dict[str, Any]] = field(default_factory=list)
+    unmatched: list[dict[str, Any]] = field(default_factory=list)
+
+    @property
+    def ok(self) -> 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"
@@ -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
@@ -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)
@@ -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
@@ -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)
@@ -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 两个出口怎么接。",