用 Jev 给机器人装一层"可算账的决策皮层"：从 SoC 选型到上线验证

{
  "state": "...",
  "model": "jev-latest",
  "questions": {
    "department": { "type": "choice", "instructions": "...", "criteria": {...} },
    "frustration": { "type": "score", "instructions": "...", "criteria": [...] },
    "is_urgent":  { "type": "noul",  "instructions": "..." }
  }
}

状态裁剪后 ≈ 600 input tokens
单次成本 = 600 × 0.042 / 1_000_000 = 2.52e-5 ≈ $0.000025
每天 5000 次决策 → $0.126/天 ≈ $3.8/月

# state_contract.py
from dataclasses import dataclass, asdict, field
from typing import Any

@dataclass
class RobotFailureState:
    """喂给 Jev 的状态。字段越少越好，每个字段都要能在归因时被引用。"""
    task_id: str
    task_goal: str                    # 自然语言目标，≤ 30 字
    failed_step: str                  # 失败发生在哪一步
    elapsed_ms: int

    # 感知证据
    detected_objects: list[dict]      # [{label, conf, bbox_cx, bbox_cy}]，最多 8 个
    target_object: str | None
    depth_valid_ratio: float          # 深度图有效率，0~1

    # 力觉证据（失败归因的杀手锏）
    gripper_force_peak_n: float       # 夹持力峰值
    joint_torque_spike: float | None  # 关节扭矩尖峰
    slip_detected: bool               # 是否检出滑移

    # 运动证据
    ik_success: bool
    collision_flag: bool
    approach_axis: str                # "top" | "side" | "front"

    # 上下文
    retry_count: int
    last_failure_mode: str | None     # 上一次尝试的归因结果
    # ❌ 不要放：完整点云、图像 base64、完整 joint trajectory

    def to_state(self) -> dict:
        d = asdict(self)
        d["detected_objects"] = d["detected_objects"][:8]
        return {"robot_failure_context": d}

# questions.py
from typesafe_sdk import Choice, Noul, Score

FAILURE_MODES = {
    "perception_miss":  "目标未被检出，或检出框与真实位姿偏差过大",
    "pose_error":       "目标检出正常，但末端到位姿态与实际不符（如倾斜、深度失效）",
    "grasp_slip":       "已接触目标，夹持过程中发生滑移或力觉异常",
    "collision":        "运动过程中触发碰撞检测或扭矩尖峰",
    "ik_unreachable":   "目标位姿逆解失败或超出工作空间",
    "occlusion":        "目标被其他物体遮挡，无法建立稳定抓取",
    "hardware_fault":   "末端/关节/传感器硬件异常（夹爪未响应、相机断流）",
    "other":            "以上均不符合，需要人工判断",
    # ↑ other 是强制项，见第 5 节坑点 2
}

RECOVERY_ACTIONS = {
    "retry_same":       "原样重试（适用于偶发抖动）",
    "re_perceive":      "重新感知：换视角/补光/重新标定深度",
    "change_approach":  "更换接近方向或抓取姿态",
    "clear_occlusion":  "先移开遮挡物，再重试目标",
    "change_tool":      "更换末端工具（夹爪→吸盘）",
    "replan_task":      "当前子任务不可行，请求重新规划",
    "escalate_human":   "升级人工接管",
}

def build_questions(state: dict) -> dict:
    return {
        # ① 归因：8 选 1
        "failure_mode": Choice(
            instructions=(
                "根据以下机器人任务执行状态，判断本次失败的最可能根本原因。"
                "优先依据力觉与运动证据；若证据冲突，选择证据最强的一类。"
            ),
            criteria=FAILURE_MODES,
        ),
        # ② 恢复：7 选 1（与①隔离评估，不会互相干扰）
        "recovery": Choice(
            instructions="针对本次失败，选择最合适的一步恢复动作。只考虑单步动作，不要规划多步。",
            criteria=RECOVERY_ACTIONS,
        ),
        # ③ 严重度：4 级有序
        "severity": Score(
            instructions="评估本次失败的严重程度和继续自动重试的风险",
            criteria=[
                "轻微，安全重试多次无风险",
                "中等，可重试但需限制次数",
                "较严重，重试可能造成损坏或安全问题",
                "严重，必须立即停止并人工介入",
            ],
        ),
        # ④ 安全闸门：是/否
        "hardware_risk": Noul(
            instructions="本次失败表现出硬件故障或安全风险迹象（异常扭矩、夹爪无响应、传感器断流）",
        ),
        # ⑤ 值得再试吗：是/否
        "worth_retry": Noul(
            instructions="在更换策略的前提下，再次尝试有较大概率成功",
        ),
    }

# jev_client.py
import os, time, logging
from dataclasses import dataclass
from typing import Any
import httpx
from typesafe_sdk import TypeSafeClient

log = logging.getLogger("jev")

@dataclass
class DecisionBudget:
    """所有超时/重试都从这里走，禁止在业务代码里硬编码。"""
    timeout_s: float = 1.2          # 单次超时：> 此值走降级
    max_retries: int = 2            # 只重试可重试错误
    retry_backoff: float = 0.15
    rtt_alert_s: float = 1.5        # 超过此 RTT 触发降级标记

RETRYABLE = {429, 529, 500, 502, 503, 504}

class JevDecisionLayer:
    def __init__(self, budget: DecisionBudget | None = None):
        self.budget = budget or DecisionBudget()
        self.client = TypeSafeClient()          # 读 TYPESAFE_API_KEY
        self._degraded = False

    @property
    def degraded(self) -> bool:
        return self._degraded

    def decide(self, state: dict, questions: dict) -> dict | None:
        """返回原始 answers dict；失败返回 None（由调用方降级）。"""
        last_err = None
        for attempt in range(self.budget.max_retries + 1):
            t0 = time.perf_counter()
            try:
                resp = self.client.system_one(
                    state=state,
                    questions=questions,
                    # ⚠️ SDK 是否原生支持 timeout 参数请以官方文档为准；
                    # 若不支持，请改用 httpx 直连端点并设置 timeout，见下方 fallback_call()
                    timeout=self.budget.timeout_s,
                )
                rtt = time.perf_counter() - t0
                if rtt > self.budget.rtt_alert_s:
                    log.warning("Jev RTT %.3fs exceeds budget", rtt)
                    self._degraded = True
                else:
                    self._degraded = False
                return resp.answers

            except Exception as e:                       # noqa: BLE001
                last_err = e
                status = getattr(e, "status_code", None) or getattr(e, "status", None)
                if status not in RETRYABLE:
                    log.error("Jev non-retryable error: %s", e)
                    break
                if attempt < self.budget.max_retries:
                    time.sleep(self.budget.retry_backoff * (2 ** attempt))
                    continue

        log.error("Jev decision failed after retries: %s", last_err)
        self._degraded = True
        return None

    # —— 若 SDK 不支持 timeout，用这个直连版本 ——
    def fallback_call(self, state: dict, questions: dict) -> dict:
        r = httpx.post(
            "https://api.typesafe.ai/v1/systemone",
            headers={"Authorization": f"Bearer {os.environ['TYPESAFE_API_KEY']}"},
            json={"state": state, "model": "jev-latest", "questions": questions},
            timeout=self.budget.timeout_s,
        )
        r.raise_for_status()
        return r.json()["answers"]

# router.py
from dataclasses import dataclass

@dataclass(frozen=True)
class CostMatrix:
    """每种错误的代价（元）。阈值由这张表反推，不是拍脑袋。"""
    auto_wrong: float      = 120.0   # 自动执行但判断错（可能撞机/损坏）
    ask_human: float       = 8.0     # 转人工的成本（运维时间）
    retry_wasted: float    = 3.0     # 白重试一次的电耗+时长

    def threshold(self) -> float:
        """
        当 p 为模型给出的置信度：
          自动执行期望代价 = (1-p) * auto_wrong
          转人工期望代价   = ask_human
        令二者相等 → p* = 1 - ask_human / auto_wrong
        """
        return 1.0 - self.ask_human / self.auto_wrong

@dataclass(frozen=True)
class RouteBands:
    auto: float      # ≥ 自动执行
    review: float    # ≥ 保守重试一次；低于此值转人工

def derive_bands(cost: CostMatrix) -> RouteBands:
    p_auto = cost.threshold()                       # 1 - 8/120 = 0.933
    p_review = 1.0 - cost.retry_wasted / cost.ask_human   # 1 - 3/8 = 0.625
    return RouteBands(auto=p_auto, review=p_review)

def route(answers: dict, bands: RouteBands) -> str:
    mode = answers["failure_mode"]
    sev  = answers["severity"]
    conf = mode.confidence
    score = sev.score                    # 可落在等级之间，如 1.035

    # ① 安全闸门优先于一切：Noul 无 confidence，概率本身就是信念
    if answers["hardware_risk"].noul > 0.5:
        return "escalate_human"
    # ② 严重度落在 2.0 以上（"较严重"及更差）直接人工
    if score >= 2.0:
        return "escalate_human"
    # ③ 归因与恢复应当一致，不一致说明模型自己也没把握 → 交叉校验
    if mode.choice == "other" or answers["recovery"].choice == "escalate_human":
        return "escalate_human"
    # ④ 按代价矩阵推导的阈值分档
    if conf >= bands.auto and answers["worth_retry"].noul > 0.5:
        return f"auto:{answers['recovery'].choice}"
    if conf >= bands.review:
        return f"retry_once:{answers['recovery'].choice}"
    return "escalate_human"

# crosscheck.py
INCOMPATIBLE = [
    # (failure_mode, recovery) 组合不合理 → 说明本次判断不可信
    ("perception_miss",  "change_tool"),
    ("ik_unreachable",   "clear_occlusion"),
    ("hardware_fault",   "retry_same"),
    ("collision",        "re_perceive"),
]

def sanity_check(answers: dict) -> tuple[bool, str]:
    mode = answers["failure_mode"].choice
    act  = answers["recovery"].choice
    if (mode, act) in INCOMPATIBLE:
        return False, f"归因({mode})与恢复动作({act})互斥"
    if answers["failure_mode"].confidence < 0.3:
        return False, "归因置信度过低"
    # Noul 与 Choice 的隐含一致性：判硬件故障却建议原样重试 → 矛盾
    if answers["hardware_risk"].noul > 0.6 and act == "retry_same":
        return False, "安全风险与恢复动作矛盾"
    return True, ""

# jev_decider_node.py
import rclpy
from rclpy.node import Node
from std_msgs.msg import String
from atexit import register

class JevDeciderNode(Node):
    def __init__(self):
        super().__init__("jev_decider")
        self.layer  = JevDecisionLayer()
        self.bands  = derive_bands(CostMatrix())
        self.sub = self.create_subscription(String, "/task/failed", self.on_failed, 10)
        self.pub = self.create_publisher(String, "/task/decision", 10)
        self.get_logger().info("Jev decider up")

    def on_failed(self, msg: String):
        state = RobotFailureState.from_ros(msg).to_state()   # 见 3.1
        answers = self.layer.decide(state, build_questions(state))

        if answers is None or self.layer.degraded:
            # 网络不可用 → 端侧 4B 兜底，或直接保守升级
            self.pub.publish(String(data="escalate_human:degraded"))
            return

        ok, why = sanity_check(answers)
        if not ok:
            self.get_logger().warn("crosscheck failed: %s", why)
            self.pub.publish(String(data="escalate_human:crosscheck"))
            return

        decision = route(answers, self.bands)
        self.get_logger().info(
            "mode=%s(p=%.3f conf=%.3f) sev=%.2f -> %s",
            answers["failure_mode"].choice,
            answers["failure_mode"].probabilities[answers["failure_mode"].choice],
            answers["failure_mode"].confidence,
            answers["severity"].score,
            decision,
        )
        self.pub.publish(String(data=decision))

def decide_with_fallback(state, questions):
    ans = jev.decide(state, questions)
    if ans is None:
        # 端侧 4B 量化模型 + grammar 约束解码，只输出 failure_mode 枚举
        local = local_model.classify(state, allowed=list(FAILURE_MODES))
        # 端侧模型置信度未校准 → 一律按"需人工复核"处理，不允许自动执行
        return f"review:{local.label}"
    return route(ans, bands)

# main.py —— 端到端最小闭环
import logging, os
from jev_client import JevDecisionLayer, DecisionBudget
from questions import build_questions
from router import CostMatrix, derive_bands, route
from crosscheck import sanity_check
from state_contract import RobotFailureState

logging.basicConfig(level=logging.INFO)

def decide_once(state: RobotFailureState) -> str:
    layer = JevDecisionLayer(DecisionBudget(timeout_s=1.2, max_retries=2))
    bands = derive_bands(CostMatrix(auto_wrong=120.0, ask_human=8.0, retry_wasted=3.0))

    s = state.to_state()
    answers = layer.decide(s, build_questions(s))
    if answers is None:
        return "escalate_human:api_unavailable"

    ok, why = sanity_check(answers)
    if not ok:
        return f"escalate_human:{why}"

    return route(answers, bands)

if __name__ == "__main__":
    demo = RobotFailureState(
        task_id="T-2041", task_goal="把桌上的马克杯放进洗碗机",
        failed_step="grasp", elapsed_ms=8400,
        detected_objects=[{"label": "mug", "conf": 0.91, "bbox_cx": 322, "bbox_cy": 241}],
        target_object="mug", depth_valid_ratio=0.62,
        gripper_force_peak_n=18.4, joint_torque_spike=None, slip_detected=True,
        ik_success=True, collision_flag=False, approach_axis="top",
        retry_count=1, last_failure_mode=None,
    )
    print(decide_once(demo))
    # 期望输出形如：auto:change_approach   （因为有 slip_detected=True）

def audit_tokens(state: dict) -> int:
    """用官方 usage 回读，别自己估"""
    resp = client.system_one(state=state, questions={"probe": Noul(instructions="state 非空")})
    return resp.usage.input_tokens

# bench_latency.py
import time, statistics
from collections import deque

def bench(n=500):
    lat = deque(maxlen=n)
    for _ in range(n):
        t0 = time.perf_counter()
        try:
            jev.decide(STATE_SAMPLE, QUESTIONS)
        except Exception:
            continue
        lat.append((time.perf_counter() - t0) * 1000)
    lat = sorted(lat)
    return {
        "n": len(lat),
        "p50": lat[len(lat)//2],
        "p95": lat[int(len(lat)*0.95)],
        "p99": lat[int(len(lat)*0.99)],
        "mean": statistics.mean(lat),
    }

# calibration.py
import numpy as np

def expected_calibration_error(samples, bins=10):
    """samples: [(confidence: float, is_correct: bool), ...]"""
    conf = np.array([s[0] for s in samples])
    corr = np.array([float(s[1]) for s in samples], dtype=float)
    ece, edges = 0.0, np.linspace(0, 1, bins + 1)
    for i in range(bins):
        m = (conf > edges[i]) & (conf <= edges[i+1])
        if m.sum() == 0:
            continue
        ece += m.sum() / len(conf) * abs(corr[m].mean() - conf[m].mean())
    return ece

# 从 usage 回读，别估算
total_in = sum(r.usage.input_tokens for r in all_responses)
cost_usd = total_in * 0.042 / 1_000_000        # 输出免费，不计
print(f"{len(all_responses)} calls, {total_in} tok, ${cost_usd:.6f}")