这套方案最后能做出什么
做完这套改造后,你的系统会有下面四个看得见摸得着的东西。
第一,一个路由层。用户请求先进入端侧 Jev,Jev 正常作答就直接返回;当它拿不准时,这次请求会被自动升级给云端的 Opus,用户端只是多等一小会儿,不需要重问一遍。
第二,一张门限配置表。不同业务意图有不同的升级阈值,这张表是跑数据跑出来的,不是凭感觉定的,改模型版本时能跟着重算。
第三,一个上下文裁剪器。升级到云端的请求不重放整段对话,只带「任务指令 + 关键槽位 + 最近一轮 + 端侧已经答了什么」,输入 token 通常能压到原来的三到五成。
第四,一本成本账。每条请求都记录路由、token 数、延迟和最终判定,随时能算出「混合模式 vs 纯云端」的准确率和成本对照。
目标效果是:准确率贴着纯云端走,成本按 token 计费口径落到纯云端的两三成到四分之一这个量级。这个数字取决于你的升级率和裁剪率,不是固定值,下文会给你公式自己代入。
前置条件清单
动手之前先确认这几件事,缺一件都会让后面的步骤卡住。
- 端侧推理栈能给出置信度素材。至少要能给 token 级对数概率,或者能让你连续采样 N 次比较结果一致性。如果什么都拿不到,只能走「结构化自评」这一条路,精度会差一些。
- 云端模型有可用的 API,并且走你自己的服务端代理。绝对不要把密钥打进端侧包体,端上是能被扒出来的。
- 一个可回放的评测集。100~300 条真实请求就够起步,每条带上输入、标准答案、业务意图标签。没有这个,门限就是拍脑袋,后面所有数字都不可信。
- 埋点管道。能把每条请求的路由决策、token 用量、延迟、是否正确落库。
- 脱敏规则。手机号、身份证、地址、账号这些,升级前要单独过一遍,裁剪不等于脱敏,两件事别混。
- 一次纯云端全量跑的基线。用同一批评测集,跑一遍纯端侧、跑一遍纯云端,拿到两个准确率和云端的总成本。这是你后面所有对照的分母。
分步骤实操
第 1 步:先把基线钉住,别急着写路由
很多人一上来就写升级逻辑,结果没有基线,跑完不知道是好了还是坏了。先用评测集跑三组数字:
```python
eval_baseline.py(伪代码骨架,按你的推理栈替换调用)
import json
def run_eval(cases, answer_fn, tag):
rows = []
for c in cases:
out = answer_fn(c["input"])
rows.append({
"id": c["id"],
"intent": c["intent"],
"pred": out["answer"],
"gold": c["gold"],
"correct": judge(out["answer"], c["gold"]), # 规则或人工
"score": out.get("confidence"), # 端侧才有
"in_tokens": out.get("in_tokens"),
"out_tokens": out.get("out_tokens"),
})
json.dump(rows, open(f"eval_{tag}.jsonl", "w"), ensure_ascii=False, indent=0)
return rows
```
跑完你会得到三行结论:纯端侧准确率、纯云端准确率、纯云端总 token 成本。端侧和云端之间的准确率差距,就是这条求救通道要补的窟窿。
第 2 步:把「拿不准」变成一个分数
单一信号都不够好,建议合成。四个可用信号:
- 序列概率:整段输出的平均对数概率,或最差那个 token 的对数概率。小模型在事实类问题上胡说时,这个值往往明显偏低。
- 自一致性:温度调高一点采样 3~5 次,看答案是否收敛。收敛不了说明模型本身没主见。
- 结构化自评:让 Jev 输出 JSON,带
answer、confidence、reason_code。注意小模型自报的 confidence 往往严重高估,只能当弱信号。 - 硬规则:命中这些直接升级,不看分数。比如涉及金额计算、需要实时外部信息、输入长度超过端侧窗口、输出里出现「不确定」「可能」这类词、检索召回分数低于阈值。
合成方式用加权求和就够起步:
```python
confidence.py
W = {"logprob": 0.5, "consistency": 0.3, "self_report": 0.2}
def score(req):
s = 0.0
if req.logprob_norm is not None:
s += W["logprob"] * normalize_logprob(req.logprob_norm)
if req.samples:
s += W["consistency"] * agreement_ratio(req.samples)
if req.self_report is not None:
s += W["self_report"] * req.self_report
return s # 0~1,越大越有把握
```
关键一步是校准。 拿评测集把样本按分数分箱(0.5 以下、0.5~0.6、0.6~0.7……),统计每箱的实际准确率。你会发现「分数 0.8 的样本实际只对了 0.65」,那就做一次单调校准(分箱映射或 isotonic 回归都行),让分数近似等于准确率。校准之后,门限才有物理意义:门限 0.8 大致就等于「接受 80% 左右的正确率」。
第 3 步:扫门限,别拍门限
门限不是选一个好看的数,是在「准确率」和「升级率」之间做取舍。写个扫描脚本,把门限从 0 到 1 按 0.02 步长走一遍:
```python
sweep.py
def sweep(rows, target_acc):
best = None
for i in range(0, 51):
t = i / 50
kept, escalated = [], []
for r in rows:
(escalated if (r["score"] < t or r["hard_rule"]) else kept).append(r)
升级的那部分按云端准确率计(用纯云端评测的准确率代入)
acc = (sum(r["correct"] for r in kept) + len(escalated) * CLOUD_ACC) / len(rows)
rate = len(escalated) / len(rows)
if acc >= target_acc:
if best is None or rate < best["rate"]:
best = {"t": t, "acc": acc, "rate": rate}
return best
```
选出来的门限满足两个条件:准确率达到你定的目标线,同时升级率尽量低。三个经验补充:
- 分意图设门限。闲聊类可以放到 0.5 以下,账务、合同、医疗相关拉到 0.9 以上。用一张配置表管住,不要写死在代码里。
- 双门限防抖动。低于
low一定升级,高于high一定本地处理,中间灰区再看第二信号(自一致性)。这样同一个人连着问三遍相似问题,不会一会儿上云一会儿不上。 - 灰区比例要盯着。灰区如果占了四成,说明你的置信度信号分辨力不够,回去补特征,别硬调门限。
```yaml
thresholds.yaml
default:
low: 0.55
high: 0.75
intents:
billing:
low: 0.85
high: 0.92
smalltalk:
low: 0.35
high: 0.60
hard_rules:
escalate_on: ["amount_calc", "realtime_info", "long_input"]
```
第 4 步:裁剪上下文,这是省钱的主战场
云端那次请求,真正需要的不是「用户说过的一切」,而是「做这个判断需要的最小信息」。四刀砍下去:
1. 砍历史:只保留最近 2~3 轮,更早的压成一句摘要(谁、要什么、已经确认了哪些条件)。
2. 砍草稿:Jev 的内部推理过程不要传给云端,但它的最终答案和不升级的理由要传,让云端做「审阅 + 纠错」而不是从零重答。
3. 砍文档:长文档走检索,只带 top-k 片段(3~5 段),并且带原文定位 id 方便回查。
4. 砍格式:把冗长的多轮对话压成槽位 JSON,比自然语言省很多 token。
```python
build_payload.py
def build_escalation_payload(session, question, jev_answer, score, retrieved):
slots = extract_slots(session) # 结构化槽位
recent = session.turns[-2:] # 最近两轮原文
summary = session.summary or summarize(session.turns[:-2])
system = (
"你是复核者。下面是一次边缘模型的作答,它自评把握不足。"
"请独立判断后给出最终答案;如果边缘答案正确,可以简短确认;"
"如果错误,直接给出正确答案并说明关键差异。不要复述推理过程。"
)
user = {
"task": question,
"known_slots": slots,
"context_summary": summary,
"recent_turns": recent,
"retrieved_snippets": retrieved[:5],
"edge_draft": {"answer": jev_answer, "self_score": round(score, 2)},
}
return system, user
```
两个必须注意的地方:端侧草稿放在最后,别放在最前面,否则云端容易被锚定,跟着错;prompt 里明确写「允许推翻上面的草稿」。另外裁剪日志一定要留,出问题时能一眼看出是不是把关键约束裁掉了。
裁剪后给自己定个 token 预算,比如输入不超过 1500 token,超了就打日志告警。这条硬上限能挡住 90% 的成本失控。
第 5 步:云端调用要能失败
升级通道是「额外的一条路」,它挂了不能把主流程带崩。
```python
escalate.py
import time
def escalate(system, user, fallback_answer, timeout_s=8.0, retries=1):
for attempt in range(retries + 1):
try:
resp = cloud_client.chat(
system=system, user=user,
timeout=timeout_s, max_tokens=800,
)
return {"answer": resp.text, "source": "cloud",
"in_tokens": resp.usage.input, "out_tokens": resp.usage.output}
except TimeoutError:
continue
except Exception as e:
log.warning("escalate_failed", err=str(e), attempt=attempt)
break
return {"answer": fallback_answer, "source": "edge_fallback", "degraded": True}
```
配套三件事:
- 超时。端侧已经等了一会儿,云端再等太久用户就跑了。设置 5~10 秒,超时直接回退到端侧答案,并加一句「建议人工确认」。
- 缓存。用问题指纹(归一化后的文本哈希 + 意图)做短期缓存,同一个问题反复问不重复上云。
- 回填。云端的答案和判定结果写回样本库,这是后面做蒸馏的原料,也是升级率能随时间下降的原因。
第 6 步:算成本账,用公式不用感觉
单请求成本就是输入 token × 输入单价 + 输出 token × 输出单价,单价以官方页面当前版本为准。混合模式下的相对成本:
```
相对成本 = 升级率 × (裁剪率 × 输入成本占比 + 输出成本占比)
```
其中「输入成本占比」指纯云端成本里输入占的比例。多数场景输出比输入贵,取 0.3 / 0.7 做示例:
| 升级率 | 上下文裁剪到 | 相对成本 |
|---|---|---|
| 40% | 60% | 0.41 |
| 30% | 40% | 0.25 |
| 25% | 50% | 0.19 |
| 20% | 30% | 0.16 |
表里 0.25 那一行就是「准确率贴着纯云端、成本四分之一」的典型参数:三成请求升级,上下文压到原来的四成。你的实际值要靠埋点跑出来,别照抄。
埋点字段至少这些:
```json
{
"request_id": "r_123",
"intent": "billing",
"route": "cloud",
"score": 0.41,
"escalate_reason": "below_low_threshold",
"in_tokens_raw": 3120,
"in_tokens_trimmed": 1180,
"out_tokens": 260,
"latency_ms": 1420,
"cloud_ok": true,
"verdict": "correct"
}
```
有了这张表,你可以按天、按意图拆开看:哪个意图贡献了最多的升级量、裁剪率有没有退化、云端失败率多少。
第 7 步:先影子跑,再放量
上线前跑一周影子模式:只计算「如果升级会怎样」,不真的调用。拿影子结果和端侧实际答案对比,看升级是不是真的救回了错误。确认之后再放 5% 流量做 A/B,比对混合模式和纯云端的准确率、P95 延迟、单位请求成本三组指标。三组都达标再全量。
常见坑与排错
置信度不校准,门限全白设。 现象是门限设 0.8,实际准确率只有 0.6。原因是小模型自评普遍高估。解法:用评测集做分箱校准,校准后再画曲线。每次换模型或改 prompt 都要重跑校准。
门限跟着模型漂移。 换了端侧模型版本,分数分布整体平移,老门限直接失效。解法:门限配置里绑定模型标识,升级模型必须重跑扫描脚本,不做灰度就等于埋雷。
裁掉了关键约束。 最典型的是把「这个不要报销」「仅限周三之前」这类否定和条件裁掉了,云端给出完全相反的建议。解法:裁剪器保留「系统指令 + 全部槽位 + 最近一轮原文」三件套,只砍更早的闲聊;同时给裁剪前后做 diff 抽样人工看 20 条。
云端被端侧答案锚定。 把错误草稿放在最前面,云端倾向于附和。解法:草稿放末尾,prompt 明确允许推翻,必要时做一次「不给草稿」的对照实验,看准确率差多少再决定要不要保留草稿。
同一问题反复上云。 用户追问「真的吗」,触发第二次升级,成本翻倍。解法:会话级升级记忆,一个问题升级过之后,短时间内该会话的相似问句直接走云端结果。
延迟叠加。 端侧 800ms + 云端 3000ms,用户体感是从「快」变成「卡」。解法:端侧先把草稿流式吐出来,标注「正在复核」,云端结果到了再替换;或者对小请求设更短的云端超时。
密钥和 PII 出端。 密钥放端上等于公开;裁剪不等于脱敏。解法:端侧只调你自己的网关,网关做鉴权、脱敏、限流三件事,再转发云端。
只看准确率不看错误代价。 账单场景答错和闲聊答错代价差几十倍。解法:门限按意图分档,高风险意图宁可升级率翻倍。
下一步建议
跑顺之后,按这个顺序继续优化:
1. 门限自动化。每天抽样线上请求做人工判定,每周重估一次分数-准确率曲线,门限用脚本更新而不是手工改。
2. 分级升级。把云端拆成两档,灰区先给中间档模型,只有中间档也低置信才升到大模型,成本还能再压一截。
3. 蒸馏回流。把云端救回来的样本做成端侧的少样本示例或微调数据,升级率会随时间自然下降,这是这套架构最有复利的部分。
4. 成本看板。把升级率、裁剪率、单位请求成本、准确率四条曲线放一起,任何一条拐头都能立刻定位到是哪次改动。
5. 评测集扩容。每次线上一句「不对」,都是新的评测样本,把它收进来,门限和裁剪规则才有持续改进的燃料。
最后一句提醒:这套东西的价值不在「把请求发给更大的模型」,而在那个门限和那段裁剪逻辑。前者花钱就能买到,后者才是你自己的资产。
