AI Agent 模型路由：质量门禁、升级与成本控制

step_contract:
  step_type: "create_refund"
  risk_tier: "high"
  allowed_tools: ["payments.get_order", "payments.create_refund"]
  output_schema: "refund_decision_v3"
  max_end_to_end_ms: 3500
  max_attempts: 2
  requires_evidence: true
  requires_human_approval: true
  abstain_when:
    - "订单归属不确定"
    - "退款金额超出策略"

步骤验收率 =
  已验收步骤数 / 尝试步骤数

轨迹验收率 =
  已验收轨迹数 / 启动轨迹数

误放行率 =
  门禁放行的无效输出数 / 门禁放行总数

升级率 =
  升级步骤数 / 路由步骤数

浪费首轮成本 =
  被拒绝的首轮尝试成本

单次验收轨迹成本 =
  路由总成本 / 已验收轨迹数

路由期望成本 =
    首选路径成本
  + 升级概率 * 回退成本
  + 验证器成本
  + 误放行概率 * 修复与风险成本

from __future__ import annotations

import argparse
import json
import math
from collections import defaultdict
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable


@dataclass(frozen=True)
class Outcome:
    policy: str
    trajectory_id: str
    accepted: bool
    gate_accepted: bool
    escalated: bool
    latency_ms: float
    route_cost: float

    @classmethod
    def from_json(cls, value: dict) -> "Outcome":
        required = {
            "policy",
            "trajectory_id",
            "accepted",
            "gate_accepted",
            "escalated",
            "latency_ms",
            "route_cost",
        }
        missing = required - value.keys()
        if missing:
            raise ValueError(f"missing fields: {sorted(missing)}")
        for key in ("policy", "trajectory_id"):
            if not isinstance(value[key], str) or not value[key]:
                raise ValueError(f"{key} must be a non-empty string")
        for key in ("accepted", "gate_accepted", "escalated"):
            if type(value[key]) is not bool:
                raise ValueError(f"{key} must be a boolean")
        for key in ("latency_ms", "route_cost"):
            if (
                isinstance(value[key], bool)
                or not isinstance(value[key], (int, float))
            ):
                raise ValueError(f"{key} must be numeric")
        outcome = cls(**{key: value[key] for key in required})
        if outcome.latency_ms < 0 or outcome.route_cost < 0:
            raise ValueError("latency_ms and route_cost must be nonnegative")
        return outcome


def read_jsonl(path: Path) -> Iterable[Outcome]:
    with path.open(encoding="utf-8") as handle:
        for line_number, line in enumerate(handle, 1):
            if not line.strip():
                continue
            try:
                yield Outcome.from_json(json.loads(line))
            except (TypeError, ValueError, json.JSONDecodeError) as error:
                raise ValueError(f"{path}:{line_number}: {error}") from error


def percentile(values: list[float], quantile: float) -> float:
    if not values:
        raise ValueError("cannot calculate a percentile for an empty list")
    ordered = sorted(values)
    index = max(0, math.ceil(len(ordered) * quantile) - 1)
    return ordered[index]


def evaluate(outcomes: Iterable[Outcome], latency_slo_ms: float) -> dict:
    grouped: dict[str, list[Outcome]] = defaultdict(list)
    for outcome in outcomes:
        grouped[outcome.policy].append(outcome)
    if not grouped:
        raise ValueError("input contains no outcomes")

    report = {}
    for policy, rows in sorted(grouped.items()):
        accepted = sum(row.accepted for row in rows)
        false_accepts = sum(
            row.gate_accepted and not row.accepted for row in rows
        )
        total_cost = sum(row.route_cost for row in rows)
        report[policy] = {
            "trajectories": len(rows),
            "acceptance_rate": accepted / len(rows),
            "false_acceptance_rate": false_accepts
            / max(1, sum(row.gate_accepted for row in rows)),
            "escalation_rate": sum(row.escalated for row in rows) / len(rows),
            "p95_latency_ms": percentile(
                [row.latency_ms for row in rows], 0.95
            ),
            "latency_slo_pass_rate": sum(
                row.latency_ms <= latency_slo_ms for row in rows
            )
            / len(rows),
            "total_cost": total_cost,
            "cost_per_accepted_trajectory": (
                total_cost / accepted if accepted else None
            ),
        }
    return report


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("outcomes", type=Path, help="JSONL route outcomes")
    parser.add_argument("--latency-slo-ms", type=float, required=True)
    args = parser.parse_args()
    if args.latency_slo_ms <= 0:
        parser.error("--latency-slo-ms must be positive")
    try:
        report = evaluate(read_jsonl(args.outcomes), args.latency_slo_ms)
    except (OSError, ValueError) as error:
        parser.error(str(error))
    print(json.dumps(report, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()

{"policy":"baseline","trajectory_id":"t-001","accepted":true,"gate_accepted":true,"escalated":false,"latency_ms":1800,"route_cost":0.042}
{"policy":"candidate","trajectory_id":"t-001","accepted":true,"gate_accepted":true,"escalated":true,"latency_ms":2400,"route_cost":0.031}
{"policy":"candidate","trajectory_id":"t-002","accepted":false,"gate_accepted":true,"escalated":false,"latency_ms":900,"route_cost":0.008}

python3 evaluate_routes.py outcomes.jsonl --latency-slo-ms 3000