Jev Recipe / 可靠性工程

LLM Fallback 策略实战:置信度门控三级降级链

生产级 LLM 可靠性工程不只是 retry。按校准置信度分级降级:高置信度自动执行、中间带换更强模型重问、低置信度转人工或安全默认值。

搭建置信度门控降级链的 6 个步骤

01先把三个术语分清:retry 只处理基础设施故障(5xx、超时),fallback 处理「答案不可信」,escalation 处理「后果不可逆」——大多数降级逻辑上线就翻车,正是因为三者共用同一条代码路径。
02自动化闸门只认校准置信度:0.85 以上自动执行。如果 0.9 的置信度不等于 90% 的实际准确率,阈值就是摆设——Jev 的概率由 RLCD 训练校准,这个等式成立。
03先设计中间带,再补边界:0.60–0.85 用更强模型重问一次,只有低于 0.60 的长尾才进人工队列。
04把决策表写进应用代码:传输错误才 retry(有预算上限)、置信度不足才 fallback、高风险或不可逆操作一律 escalate to human——策略永远不要交给模型。
05加迟滞(hysteresis)与影子模式:易抖动的输入放宽中间带,先对生产流量只记日志不强制,验证分布后再启用。
06每一跳都落日志(answer、confidence、lane),每个季度用真实回流数据重新校准阈值,而不是拍脑袋。
schema / 降级决策契约
{
  "fallback_action": {
    "type": "choice",
    "instructions": "决定这条待审答案的下一步动作",
    "criteria": {
      "execute": "高置信度且可逆 — 自动执行",
      "escalate_model": "置信度处于中间带 — 换更强模型重问一次",
      "human_review": "低置信度或不可逆 — 转人工处理"
    }
  },
  "irreversible": {
    "type": "noul",
    "instructions": "执行这条答案是否难以撤回(支付、发信、删除)?"
  }
}
把一条待审答案作为 state 发送,观察闸门如何把流量分进 execute、escalate_model 与 human_review 三条车道。

降级策略基准对比:置信度门控链 vs 盲目 retry、级联、纯人工

METRIC
Jev
盲目 retry / 级联 / 纯人工
触发信号
校准置信度

每条答案一个数字,直接决定车道

替代信号,不是置信度

retry 盯 5xx/超时,级联盯自报分数,纯人工等用户投诉

附加延迟 (P50)
闸门 ~95ms

仅 0.60–0.85 中间带增加一次 ~2–4s 的强模型重问

层层叠加

retry 加 2^n 退避,级联每跳完整重生成,人工队列从分钟到天

成本结构
仅输入计费

闸门 $0.042/M,贵模型只为中间带买单

上不封顶

retry 为同一个错误答案反复付输出 token,级联每跳按溢价计费,人工时间成本远超两者

失败模式
失败即安全

阈值以下进人工,绝不会自动执行错误答案

或吵或哑

retry 原样重复同一个错误答案,级联静默降质,纯人工把队列淹死

可观测性
逐决策审计

每一跳输出 answer + confidence + lane,阈值在代码里可查

四处散落

故障藏在延迟曲线与质量漂移里,没有逐决策的置信度轨迹

代码对比:盲目 retry vs 置信度门控降级链

python / 脆弱写法:单一模型盲目 retry
import time
import requests

def generate_with_retry(prompt: str, max_retries: int = 5) -> str:
    # 盲目 retry:唯一的信号是传输层健康度。
    # 一个「自信的错误答案」第一次就通过,永远不会进重试路径。
    for attempt in range(max_retries):
        try:
            resp = requests.post(
                "https://api.vendor-llm.example/v1/chat/completions",
                json={
                    "model": "cheap-model",
                    "messages": [{"role": "user", "content": prompt}],
                },
                timeout=30,
            )
            resp.raise_for_status()
            return resp.json()["choices"][0]["message"]["content"]
        except (requests.Timeout, requests.HTTPError):
            time.sleep(2 ** attempt)  # 1s、2s、4s、8s……最多白等 ~31s

    return ""  # 重试风暴结束后静默返回空串

# 典型路径:1500~3000ms 生成延迟,每次尝试都计输出 token,
# 对「格式完美但答案错误」零防护。
python / Jev 置信度门控三级降级链
import requests

JEV_ENDPOINT = "https://api.typesafe.ai/v1/jev/evaluate"
AUTO_EXECUTE_CONFIDENCE = 0.85  # ≥ 0.85 自动执行

QUESTIONS = {
    "fallback_action": {
        "type": "choice",
        "instructions": "决定这条待审答案的下一步动作",
        "criteria": {
            "execute": "高置信度且可逆 - 自动执行",
            "escalate_model": "置信度处于中间带 - 换更强模型重问",
            "human_review": "低置信度或不可逆 - 转人工",
        },
    },
    "irreversible": {
        "type": "noul",
        "instructions": "执行这条答案是否难以撤回(支付、发信、删除)?",
    },
}

def gate(state: dict) -> dict:
    # 值得重试的只有传输错误——闸门本身就是一次廉价无状态调用
    # (~95ms,仅输入计费)。
    resp = requests.post(
        JEV_ENDPOINT,
        json={"state": state, "questions": QUESTIONS},
        timeout=3,
    )
    resp.raise_for_status()
    return resp.json()

def fallback_chain(task: str, answer_model, strong_model) -> dict:
    # 车道 1:便宜模型起草,校准闸门裁决。
    draft = answer_model(task)
    check = gate({"task": task, "proposed_answer": draft})
    decision = check["fallback_action"]

    if (
        decision["answer"] == "execute"
        and decision["confidence"] >= AUTO_EXECUTE_CONFIDENCE
        and not check["irreversible"]["answer"]
    ):
        return {"lane": "auto_execute", "answer": draft, "confidence": decision["confidence"]}

    # 车道 2:0.60~0.85 中间带——用更强模型重问一次。
    stronger = strong_model(task)
    recheck = gate({"task": task, "proposed_answer": stronger})
    final = recheck["fallback_action"]

    if (
        final["answer"] == "execute"
        and final["confidence"] >= AUTO_EXECUTE_CONFIDENCE
    ):
        return {"lane": "escalated_model", "answer": stronger, "confidence": final["confidence"]}

    # 车道 3:升级后仍低于阈值——进人工队列或安全默认值。
    return {
        "lane": "human_review",
        "answer": stronger,
        "confidence": final["confidence"],
        "first_pass_confidence": decision["confidence"],
    }

# 典型路径:每次闸门调用 ~95ms,仅输入计费。车道 2 是例外而非常态——
# 大部分流量不会离开车道 1。
typescript / 脆弱写法:盲目 retry
async function generateWithRetry(prompt: string, maxRetries = 5) {
  // 盲目 retry:只有传输错误会触发循环,
  // 「自信的错误答案」第一次尝试就径直通过。
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const resp = await fetch("https://api.vendor-llm.example/v1/chat/completions", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
          model: "cheap-model",
          messages: [{ role: "user", content: prompt }],
        }),
      });
      if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
      return (await resp.json()).choices[0].message.content as string;
    } catch {
      await new Promise((r) => setTimeout(r, 2 ** attempt * 1000));
    }
  }
  return ""; // 退避 ~31s 之后静默失败
}
typescript / Jev 置信度门控降级链
const JEV_ENDPOINT = "https://api.typesafe.ai/v1/jev/evaluate";
const AUTO_EXECUTE_CONFIDENCE = 0.85; // ≥ 0.85 自动执行

const QUESTIONS = {
  fallback_action: {
    type: "choice",
    instructions: "决定这条待审答案的下一步动作",
    criteria: {
      execute: "高置信度且可逆 - 自动执行",
      escalate_model: "置信度处于中间带 - 换更强模型重问",
      human_review: "低置信度或不可逆 - 转人工",
    },
  },
  irreversible: {
    type: "noul",
    instructions: "执行这条答案是否难以撤回(支付、发信、删除)?",
  },
} as const;

async function gate(state: Record<string, unknown>) {
  const resp = await fetch(JEV_ENDPOINT, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ state, questions: QUESTIONS }),
  });
  if (!resp.ok) throw new Error("Jev evaluate failed");
  return (await resp.json()) as {
    fallback_action: { answer: string; confidence: number };
    irreversible: { answer: boolean };
  };
}

export async function fallbackChain(
  task: string,
  answerModel: (t: string) => Promise<string>,
  strongModel: (t: string) => Promise<string>,
) {
  // 车道 1:便宜模型起草,校准闸门裁决(~95ms,仅输入计费)
  const draft = await answerModel(task);
  const first = await gate({ task, proposed_answer: draft });
  if (
    first.fallback_action.answer === "execute" &&
    first.fallback_action.confidence >= AUTO_EXECUTE_CONFIDENCE &&
    !first.irreversible.answer
  ) {
    return { lane: "auto_execute", answer: draft, confidence: first.fallback_action.confidence };
  }

  // 车道 2:0.60~0.85 中间带——用更强模型重问一次
  const stronger = await strongModel(task);
  const recheck = await gate({ task, proposed_answer: stronger });
  if (
    recheck.fallback_action.answer === "execute" &&
    recheck.fallback_action.confidence >= AUTO_EXECUTE_CONFIDENCE
  ) {
    return { lane: "escalated_model", answer: stronger, confidence: recheck.fallback_action.confidence };
  }

  // 车道 3:升级后仍低于阈值——人工队列或安全默认值
  return {
    lane: "human_review",
    answer: stronger,
    confidence: recheck.fallback_action.confidence,
    first_pass_confidence: first.fallback_action.confidence,
  };
}

继续深入可靠性工程