支持评估型指标：rubric 加权评分、等级与 assessed 标签

backend/app/data/caliber_identity.py

@@ -26,7 +26,7 @@ from typing import Any
 
 #: 指标的语义字段：只有这几样算定义，运行时值（value、rendered、inputs……）不参与哈希
 SEMANTIC_FIELDS = ("id", "name", "expression", "unit", "decimals", "format", "kind", "definition", "params", "external",
-                   "rule")
+                   "rule", "assessment")
 
 
 def _blank(value: Any) -> bool:

backend/app/engine/nodes/metrics.py

@@ -105,6 +105,52 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
         if problem:
             errors.append(f"指标「{metric_id}」：{problem}")
             continue
+        if str(definition.get("kind") or "formula") == "assessment":
+            assessment, problem = _assessment_of(definition)
+            if problem:
+                errors.append(f"指标「{metric_id}」：{problem}")
+                continue
+            dimensions = assessment["dimensions"]
+            values_by_name = {str(d["name"]): _value_of(str(d["input"]), tctx) for d in dimensions}
+            missing = [name for name, value in values_by_name.items() if value is None]
+            if missing and on_missing != "null":
+                errors.append(f"指标「{metric_id}」：" + _missing_reason([str(d["input"]) for d in dimensions
+                                                                         if d["name"] in missing]))
+                continue
+            value: Any = None
+            details: list[dict[str, Any]] = []
+            level: str | None = None
+            if not missing:
+                total = sum(float(d.get("weight", 1)) for d in dimensions)
+                points = 0.0
+                for d in dimensions:
+                    low, high = float(d.get("min", 0)), float(d.get("max", 100))
+                    raw = float(values_by_name[str(d["name"])])
+                    normalized = max(0.0, min(100.0, (raw - low) / (high - low) * 100))
+                    weight = float(d.get("weight", 1))
+                    points += normalized * weight
+                    details.append({"name": d["name"], "value": raw, "normalized": normalized, "weight": weight})
+                value = round_metric(points / total, display["decimals"])
+                level = next((str(l["name"]) for l in assessment["levels"] if value >= float(l["min"])), None)
+            inputs = [_input_of(str(d["input"]), values_by_name[str(d["name"])], state, ctx, tctx)
+                      for d in dimensions]
+            metrics.append({
+                "id": metric_id,
+                "name": definition.get("name") or metric_id,
+                "unit": definition.get("unit") or "",
+                "value": value,
+                "expression": "",
+                "decimals": display["decimals"],
+                "format": display["format"],
+                "rendered": _rendered(value, definition, display),
+                "inputs": inputs,
+                "substituted": "加权评分",
+                "recompute_ok": True,
+                "status": "ok" if value is not None else "missing_input",
+                "assessment": {"score": value, "level": level, "dimensions": details,
+                               "levels": assessment["levels"], **({"missing": missing} if missing else {})},
+            })
+            continue
         if str(definition.get("kind") or "formula") == "rule":
             rule, problem = _rule_of(definition)
             if problem:
@@ -501,6 +547,27 @@ def _external_of(definition: dict[str, Any]) -> tuple[Any, dict[str, Any], str |
     return value, {"source": source, "as_of": as_of.isoformat(), "stale_after_days": days, "stale": stale}, None
 
 
+def _assessment_of(definition: dict[str, Any]) -> tuple[dict[str, Any], str | None]:
+    """评估型指标：{dimensions: [{name, input, weight, min, max}], levels: [{name, min}]}。"""
+    assessment = definition.get("assessment")
+    if not isinstance(assessment, dict):
+        return {}, "评估型指标需要 assessment 对象（dimensions / levels）"
+    dimensions = assessment.get("dimensions")
+    if not isinstance(dimensions, list) or not dimensions:
+        return {}, "评估需要至少一个维度"
+    for dimension in dimensions:
+        if not isinstance(dimension, dict) or not isinstance(dimension.get("name"), str) \
+                or not isinstance(dimension.get("input"), str) \
+                or isinstance(dimension.get("weight", 1), bool) \
+                or not isinstance(dimension.get("weight", 1), (int, float)) \
+                or float(dimension.get("weight", 1)) <= 0:
+            return {}, "评估维度需要 name、input 和正的 weight"
+    levels = assessment.get("levels")
+    if not isinstance(levels, list) or not levels:
+        return {}, "评估需要至少一个等级"
+    return assessment, None
+
+
 def _rule_of(definition: dict[str, Any]) -> tuple[dict[str, Any], str | None]:
     """规则型指标：{inputs: {名字: 取值路径}, all: [{input, op, value|values}], then, else}。"""
     rule = definition.get("rule")

backend/app/engine/schema.py

@@ -1088,7 +1088,7 @@ def _check_challenge(node: GraphNode, spec: GraphSpec, result: ValidationResult)
 
 #: 指标定义的类型。现在只有公式（表达式）这一种在运行时能算；规则、评估、外部常量还没实现，
 #: 先在校验里挡住，免得写进图里以后被当成能算的指标
-METRIC_KINDS = ("formula", "external", "rule")
+METRIC_KINDS = ("formula", "external", "rule", "assessment")
 #: 定义卡参数的标量类型（和入口表单的字段类型同一套子集）
 METRIC_PARAM_TYPES = ("string", "number", "integer", "boolean")
 #: 规则条件能用的比较
@@ -1170,6 +1170,56 @@ def metric_definition_issues(node: GraphNode, metric: Any, index: int) -> list[V
             for key in ("then", "else"):
                 if key in rule and (isinstance(rule[key], bool) or not isinstance(rule[key], (int, float))):
                     add(f"规则的 {key} 需要是数", f"{field}.rule.{key}", code="metrics.rule_invalid")
+    elif kind == "assessment":
+        assessment = metric.get("assessment")
+        if not isinstance(assessment, dict):
+            add("评估型指标需要 assessment 对象（dimensions / levels）", f"{field}.assessment",
+                code="metrics.assessment_invalid")
+        else:
+            dimensions = assessment.get("dimensions")
+            if not isinstance(dimensions, list) or not dimensions:
+                add("评估需要 dimensions：至少一个维度", f"{field}.assessment.dimensions",
+                    code="metrics.assessment_invalid")
+                dimensions = []
+            names: set[str] = set()
+            for j, dimension in enumerate(dimensions):
+                key = f"{field}.assessment.dimensions[{j}]"
+                if not isinstance(dimension, dict):
+                    add("评估维度需要是一个对象", key, code="metrics.assessment_invalid")
+                    continue
+                name = dimension.get("name")
+                if not isinstance(name, str) or not name.strip() or name in names:
+                    add("评估维度需要唯一的名字", f"{key}.name", code="metrics.assessment_invalid")
+                else:
+                    names.add(name)
+                if not isinstance(dimension.get("input"), str) or not dimension["input"].strip():
+                    add("评估维度需要 input（取值路径）", f"{key}.input", code="metrics.assessment_invalid")
+                weight = dimension.get("weight", 1)
+                if isinstance(weight, bool) or not isinstance(weight, (int, float)) or weight <= 0:
+                    add("评估维度的 weight 需要是正数", f"{key}.weight", code="metrics.assessment_invalid")
+                low, high = dimension.get("min", 0), dimension.get("max", 100)
+                if any(isinstance(v, bool) or not isinstance(v, (int, float)) for v in (low, high)) or high <= low:
+                    add("评估维度的 min / max 需要是数，且 max 大于 min", key, code="metrics.assessment_invalid")
+            levels = assessment.get("levels")
+            if not isinstance(levels, list) or not levels:
+                add("评估需要 levels：至少一个等级", f"{field}.assessment.levels",
+                    code="metrics.assessment_invalid")
+                levels = []
+            previous: float | None = None
+            for j, level in enumerate(levels):
+                key = f"{field}.assessment.levels[{j}]"
+                if not isinstance(level, dict):
+                    add("评估等级需要是一个对象", key, code="metrics.assessment_invalid")
+                    continue
+                if not isinstance(level.get("name"), str) or not level["name"].strip():
+                    add("评估等级需要名字", f"{key}.name", code="metrics.assessment_invalid")
+                minimum = level.get("min")
+                if isinstance(minimum, bool) or not isinstance(minimum, (int, float)):
+                    add("评估等级的 min 需要是数", f"{key}.min", code="metrics.assessment_invalid")
+                elif previous is not None and minimum >= previous:
+                    add("评估等级要按 min 从高到低排列", f"{key}.min", code="metrics.assessment_invalid")
+                else:
+                    previous = float(minimum)
     elif not metric.get("expression"):
         add(f"指标定义缺少表达式：{metric.get('id') or '（空）'}", code="metrics.expression_missing")
 

backend/app/ledger/descriptor.py

@@ -28,7 +28,8 @@ from app.look.calibers import card_inputs
 from app.tools.datasource import QUERY_PREFIX
 
 #: 指标只写口径卡配置里真有的这几个键
-METRIC_KEYS = ("id", "name", "unit", "decimals", "format", "kind", "definition", "params", "external", "rule")
+METRIC_KEYS = ("id", "name", "unit", "decimals", "format", "kind", "definition", "params", "external", "rule",
+               "assessment")
 
 
 async def list_calibers(session: AsyncSession, *, workflow_id: str | None = None) -> list[dict[str, Any]]:

backend/app/ledger/labels.py

@@ -34,7 +34,7 @@ from typing import Any
 from app.analysis.contract import LEDGER_KIND
 
 #: 标签的四种、认法的两种
-KINDS = ("caliber", "external", "derived", "related", "reasoning", "unknown")
+KINDS = ("caliber", "external", "derived", "assessed", "related", "reasoning", "unknown")
 HOWS = ("independent", "composed")
 #: 界面上标签要写明的两句：比对在封存之后做，不在封存范围内；事后补比的是按现在的口径回看
 POST_SEAL_NOTE = "封存后比对，不在封存范围内"
@@ -313,6 +313,11 @@ def metric_label(card: Mapping[str, Any] | None, metric_id: Any, queries: Mappin
         return {"kind": "derived", "rule": metric.get("rule"),
                 "queries": [str(i["artifact"]) for i in from_queries],
                 **({"definition_hash": card.get("definition_hash")} if card.get("definition_hash") else {})}
+    if metric.get("assessment") is not None:
+        # 评估值：按评分协议从输入算出，不是测量值，也不参与自动出具
+        return {"kind": "assessed", "assessment": metric.get("assessment"),
+                "queries": [str(i["artifact"]) for i in from_queries],
+                **({"definition_hash": card.get("definition_hash")} if card.get("definition_hash") else {})}
     if not from_queries:
         return {"kind": "reasoning"}
     rows = [queries.get(str(i["artifact"])) for i in from_queries]

backend/tests/test_analysis_judge_ledger.py

@@ -88,7 +88,8 @@ def test_analysis_numbers_get_no_caliber_label_and_are_not_counted():
     labelled = {s["id"]: s.get("ref") for s in iter_segments(doc) if s["id"] in labels}
     assert list(labelled.values()) == ["v:Q1.r0.gmv"]
     whole = ledger_labels.whole_of(labels, {})
-    assert whole["counts"] == {"caliber": 0, "external": 0, "derived": 0, "related": 0, "reasoning": 1, "unknown": 0}
+    assert whole["counts"] == {"caliber": 0, "external": 0, "derived": 0, "assessed": 0, "related": 0,
+                               "reasoning": 1, "unknown": 0}
     assert ledger_labels.referenced_columns([doc], lambda _: None) == {q: {"gmv"}}
 
 

backend/tests/test_ledger_cells.py

@@ -274,7 +274,8 @@ async def test_last_weeks_sales_labels_only_the_sales_number_as_from_the_caliber
     assert {c: cols[c]["kind"] for c in COLUMNS[1:]} == {c: "related" for c in COLUMNS[1:]}
     whole = content["whole"]
     assert whole["equivalent"] is False and whole["counts"] == {"caliber": 1, "external": 0, "derived": 0,
-                                                                 "related": 3, "reasoning": 0, "unknown": 0}
+                                                                 "assessed": 0, "related": 3, "reasoning": 0,
+                                                                 "unknown": 0}
     # 工件 meta：逐格证明了的口径也记上（召回按旧版本反查只读它）
     from sqlalchemy import select
 

backend/tests/test_metric_definition.py

@@ -29,7 +29,7 @@ def test_a_formula_metric_can_carry_a_definition_and_params():
 
 
 def test_an_unimplemented_kind_is_refused():
-    errors = _errors(_graph(kind="assessment"))
+    errors = _errors(_graph(kind="normative"))
     assert any(i.code == "metrics.kind_unsupported" and "还没有实现" in i.message for i in errors)
 
 
@@ -148,3 +148,59 @@ def test_a_rule_metric_is_labelled_derived_with_its_input_queries():
     label = metric_label(card, "big", {}, {})
     assert label["kind"] == "derived" and label["queries"] == ["q1"]
     assert label["definition_hash"] == "abc" and label["rule"]["all"][0]["op"] == "eq"
+
+
+def test_an_assessment_metric_needs_unique_dimensions_and_descending_levels():
+    bad = _errors(_graph(kind="assessment", assessment={
+        "dimensions": [{"name": "a", "input": "x"}, {"name": "a", "input": "y"}],
+        "levels": [{"name": "B", "min": 70}, {"name": "A", "min": 85}]}))
+    assert [i.code for i in bad].count("metrics.assessment_invalid") >= 2
+    good = _graph(kind="assessment", assessment={
+        "dimensions": [{"name": "响应", "input": "cell(nodes.fetch, 0, 'response')", "weight": 0.5},
+                       {"name": "解决", "input": "cell(nodes.fetch, 0, 'resolution')", "weight": 0.5}],
+        "levels": [{"name": "A", "min": 85}, {"name": "B", "min": 70}, {"name": "C", "min": 0}]})
+    assert _errors(good) == []
+
+
+async def test_an_assessment_metric_scores_and_picks_a_level(tmp_path):
+    from app.api.evidence import _sealed, _sealed_cards
+    from app.engine.runner import run_manager
+
+    await run_manager.setup()
+    try:
+        await S.fresh_source(tmp_path / "assessment.db")
+        graph = S.chain(
+            S.node("start", "input"),
+            S.node("fetch", "tool", tool=S.tool(), args={"sql": "SELECT 90 AS response, 70 AS resolution"}),
+            S.node("card", "metrics", caliber="服务质量", caliber_version="v1", metrics=[{
+                "id": "quality", "name": "服务质量", "unit": "分", "kind": "assessment",
+                "assessment": {
+                    "dimensions": [{"name": "响应", "input": "cell(nodes.fetch, 0, 'response')", "weight": 0.5},
+                                   {"name": "解决", "input": "cell(nodes.fetch, 0, 'resolution')", "weight": 0.5}],
+                    "levels": [{"name": "A", "min": 85}, {"name": "B", "min": 70}, {"name": "C", "min": 0}],
+                },
+            }]),
+            S.node("out", "output", fields=[{"name": "v", "value": "{{ nodes.card.metrics[0].value }}"}]))
+        started = await run_manager.start(graph=graph, input_payload={})
+        row = await S.wait(started.id)
+        assert row.status == "succeeded", row.error
+        assert float(row.output["v"]) == 80
+        sealed = await _sealed(row.id)
+        [card] = [c for _, _, c in _sealed_cards(sealed)]
+        assessment = card["metrics"][0]["assessment"]
+        assert assessment["level"] == "B" and assessment["score"] == 80
+        assert [d["normalized"] for d in assessment["dimensions"]] == [90.0, 70.0]
+    finally:
+        await run_manager.shutdown()
+
+
+def test_an_assessment_metric_is_labelled_assessed():
+    from app.ledger.labels import metric_label
+
+    card = {"definition_hash": "abc", "metrics": [{
+        "id": "quality", "assessment": {"score": 80, "level": "B"},
+        "inputs": [{"artifact": "q1", "via": "tool_cell"}],
+    }]}
+    label = metric_label(card, "quality", {}, {})
+    assert label["kind"] == "assessed" and label["queries"] == ["q1"]
+    assert label["assessment"]["level"] == "B" and label["definition_hash"] == "abc"

frontend/src/run/LedgerStep.tsx

@@ -367,6 +367,20 @@ export function LedgerPart({ runId, runClass, report, segId, segText, segLedger,
           {label.rule?.all?.length ? <p className="text-2xs text-faint">条件：{label.rule.all.length} 条</p> : null}
         </div>
       )}
+      {label.kind === 'assessed' && (
+        <div className="space-y-1" data-ledger-assessed="">
+          <p className="font-medium">评估值</p>
+          <p className="text-2xs text-dim">
+            按评分协议从输入算出，不是测量值；不参与自动正式出具。
+          </p>
+          {label.assessment?.level && (
+            <p className="text-2xs text-faint">
+              等级：{label.assessment.level}
+              {label.assessment.score != null ? ` · 得分：${label.assessment.score}` : ''}
+            </p>
+          )}
+        </div>
+      )}
       {label.kind === 'related' && (
         <div className="space-y-2">
           <p className="text-2xs text-dim">{LEDGER_UI.step.relatedSub}{label.reason ? `：${label.reason}` : ''}</p>

frontend/src/typesLedger.ts

@@ -53,12 +53,13 @@ export interface LedgerRelated {
  * - unknown：没比对（reason 写为什么）
  */
 export interface LedgerLabel {
-  kind: 'caliber' | 'external' | 'derived' | 'related' | 'reasoning' | 'unknown'
+  kind: 'caliber' | 'external' | 'derived' | 'assessed' | 'related' | 'reasoning' | 'unknown'
   how?: 'independent' | 'composed'
   source?: string
   as_of?: string
   stale?: boolean
   rule?: { inputs?: Record<string, string>; all?: unknown[]; then?: number; else?: number }
+  assessment?: { score?: number | null; level?: string | null; dimensions?: unknown[]; levels?: unknown[] }
   caliber?: string
   workflow_id?: string
   version?: number