diff --git a/backend/app/data/caliber_identity.py b/backend/app/data/caliber_identity.py
index 693bb106..2f12915d 100644
--- a/backend/app/data/caliber_identity.py
+++ b/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:
diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py
index 341b887d..3dd97c88 100644
--- a/backend/app/engine/nodes/metrics.py
+++ b/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")
diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py
index 9467d584..26fc5de5 100644
--- a/backend/app/engine/schema.py
+++ b/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")
diff --git a/backend/app/ledger/descriptor.py b/backend/app/ledger/descriptor.py
index 590dff92..c32bce18 100644
--- a/backend/app/ledger/descriptor.py
+++ b/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]]:
diff --git a/backend/app/ledger/labels.py b/backend/app/ledger/labels.py
index 103dbc6f..a124d4c8 100644
--- a/backend/app/ledger/labels.py
+++ b/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]
diff --git a/backend/tests/test_analysis_judge_ledger.py b/backend/tests/test_analysis_judge_ledger.py
index 49d73ea6..a95248e1 100644
--- a/backend/tests/test_analysis_judge_ledger.py
+++ b/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"}}
diff --git a/backend/tests/test_ledger_cells.py b/backend/tests/test_ledger_cells.py
index 468de7a9..3f397d81 100644
--- a/backend/tests/test_ledger_cells.py
+++ b/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
diff --git a/backend/tests/test_metric_definition.py b/backend/tests/test_metric_definition.py
index 5c6953c6..e3beedd4 100644
--- a/backend/tests/test_metric_definition.py
+++ b/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"
diff --git a/frontend/src/run/LedgerStep.tsx b/frontend/src/run/LedgerStep.tsx
index 00cf03e2..f7b2bbf4 100644
--- a/frontend/src/run/LedgerStep.tsx
+++ b/frontend/src/run/LedgerStep.tsx
@@ -367,6 +367,20 @@ export function LedgerPart({ runId, runClass, report, segId, segText, segLedger,
{label.rule?.all?.length ? 条件：{label.rule.all.length} 条
: null}
)}
+ {label.kind === 'assessed' && (
+
+
评估值
+
+ 按评分协议从输入算出，不是测量值；不参与自动正式出具。
+
+ {label.assessment?.level && (
+
+ 等级：{label.assessment.level}
+ {label.assessment.score != null ? ` · 得分：${label.assessment.score}` : ''}
+
+ )}
+
{LEDGER_UI.step.relatedSub}{label.reason ? `：${label.reason}` : ''}
diff --git a/frontend/src/typesLedger.ts b/frontend/src/typesLedger.ts
index 7a64d96e..166dd9c8 100644
--- a/frontend/src/typesLedger.ts
+++ b/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
; all?: unknown[]; then?: number; else?: number }
+ assessment?: { score?: number | null; level?: string | null; dimensions?: unknown[]; levels?: unknown[] }
caliber?: string
workflow_id?: string
version?: number