这篇能做出什么
跟着做一遍,你会拿到四样东西:
1. 一张档位映射表——哪些任务走 Sol、哪些走 Luna、哪些暂时留在原地,以及每种选择的成本差;
2. 一套缓存友好的提示词结构,系统提示、工具定义、知识材料稳定在前,变化的内容全部后移,命中率可以量化;
3. 一个能离线跑的回归脚本,用同一批样例对比迁移前后的通过率与 token 消耗;
4. 一个路由层 + 灰度开关 + 一键回滚方案,改配置不发版。
主线选一个真实场景:客服工单系统。它有几类调用——工单分类、发票字段抽取、多轮排障问答、需要调工具的工单升级判断。迁移的目标是让分类和抽取换到轻量档位、排障问答换到高能力档位,同时把账单压下来。
前置条件清单
- 已经有一批 GPT-5 的线上调用,能拿到请求日志(至少含任务名、输入输出 token、延迟);
- 代码里的模型名集中在少数几处,或者你愿意先做这一步收敛;
- 有一个能热更新的配置源:配置中心、特性开关、环境变量注入都行;
- 有一个存放评测样例的文件(三十条也能开始,越多越好);
- 能读到官方定价页、模型列表、提示缓存文档——模型 ID、参数名、计价口径全部以官方文档当前版本为准,本文不写任何具体数字。
步骤一:先把调用点盘点清楚
没有清单就没法选档位。从日志里按任务聚合,看清每个任务的规模。
```python
inventory.py —— 按任务聚合 token 与延迟
import json, statistics
from collections import defaultdict
buckets = defaultdict(list)
with open("requests.jsonl") as f:
for line in f:
r = json.loads(line)
buckets[(r["task"], r["model"])].append(r)
print(f"{'task':<24}{'model':<16}{'n':>7}{'in_p50':>9}{'out_p50':>9}{'p95_s':>8}")
for (task, model), rows in sorted(buckets.items()):
ins = statistics.median(x["usage"]["input_tokens"] for x in rows)
outs = statistics.median(x["usage"]["output_tokens"] for x in rows)
lats = sorted(x["latency_s"] for x in rows)
p95 = lats[min(int(len(lats) * 0.95), len(lats) - 1)]
print(f"{task:<24}{model:<16}{len(rows):>7}{ins:>9.0f}{outs:>9.0f}{p95:>8.2f}")
```
跑完你会看到典型的分布:分类和抽取类请求数量大、输入短、输出更短;排障问答请求少但输入长、输出长、还带多轮历史。这两类的成本结构完全不同,也就决定了它们不该用同一个档位。
步骤二:选档与定价对比
选档的判断规则可以简化为四条:
- 失败代价高、需要多步推理或工具编排 → 高能力档位(Sol);
- 请求量大、结构固定、输出可以用规则或 schema 校验 → 轻量档位(Luna);
- 错误可以自动重试且有校验手段 → 轻量档位 + 重试;
- 长上下文检索问答 → 先看缓存能不能命中,再看档位,因为长输入的成本大头在输入侧。
定价不要抄进代码,放到配置文件里,改价不用发版。下面的数值留空,请照抄官方定价页。
```yaml
prices.yaml —— 数值以官方定价页当前版本为准
gpt-5:
input_per_mtok: 0.0
cached_input_per_mtok: 0.0
output_per_mtok: 0.0
gpt-6-sol:
input_per_mtok: 0.0
cached_input_per_mtok: 0.0
output_per_mtok: 0.0
gpt-6-luna:
input_per_mtok: 0.0
cached_input_per_mtok: 0.0
output_per_mtok: 0.0
```
```python
cost.py
import yaml
P = yaml.safe_load(open("prices.yaml"))
def cost(model, input_tokens, cached_tokens, output_tokens):
p = P[model]
fresh = max(input_tokens - cached_tokens, 0)
return (
fresh * p["input_per_mtok"]
+ cached_tokens * p["cached_input_per_mtok"]
+ output_tokens * p["output_per_mtok"]
) / 1_000_000
def cost_per_1k(model, sample_rows):
total = 0.0
for r in sample_rows:
u = r["usage"]
total += cost(model, u["input_tokens"], u.get("cached_tokens", 0),
u["output_tokens"])
return total / len(sample_rows) * 1000
```
对比时看三个数,而不是只看单价:
- 每千次请求成本;
- 每成功完成任务成本(把重试和失败算进去,失败也是要付钱的);
- p95 延迟,以及它是否卡在业务的等待预算里。
一个真实会遇到的结论是:某个任务在轻量档位单价低,但失败率上升带来重试,折算到"每完成一次"反而更贵。这就是必须把重试率一起算的原因。
步骤三:提示缓存改写
缓存命中的本质很朴素:请求的前缀逐字节相同,服务端就能复用之前算好的中间结果。所以改写的核心动作只有一个——把稳定的放前面,把会变的放最后。
推荐的顺序:
```
系统指令 → 工具/schema 定义 → 长期知识或示例 → 对话历史 → 本轮用户输入
```
前缀里绝对不能出现的东西:时间戳、请求 ID、用户 ID、随机数、每次键序不同的 JSON、末位抖动的小数。
改写前后对比:
```python
改写前:前缀每天都变,缓存基本不会命中
prompt = f"""你是客服助手。当前时间 {now}。用户ID {uid}。
知识库:{kb_text}
用户问:{question}"""
```
```python
改写后:只有最后一段是变化的
import json
SYSTEM = "你是客服助手。严格按照给定知识库回答,信息不足时建议转人工。"
TOOLS = json.dumps(TOOL_SPECS, sort_keys=True, ensure_ascii=False)
def build_messages(kb_text, history, question):
return [
{"role": "system", "content": SYSTEM},
{"role": "system", "content": "<tools>" + TOOLS + "</tools>"},
{"role": "system", "content": "<kb>" + kb_text + "</kb>"},
*history, # 追加,不重排、不裁剪
{"role": "user", "content": question},
]
```
几个细节:
- 时间、用户 ID 这类信息挪到最后一条 user 消息或单独的 metadata 字段;
- 序列化统一用
sort_keys=True,缩进固定,浮点数固定小数位; - 多轮对话采用追加式构造,不要每轮重排历史,也不要为了省 token 从中间裁剪——裁剪会改变前缀;
- 工具定义的顺序固定;只挂载本次需要的工具时要注意,工具列表一变,前缀就断了;
- 知识材料如果每天更新一次,把它放在系统段靠后的位置,比放在最前面更划算。
命中率要看得到,才能优化:
```python
def cache_stats(usage):
cached = getattr(usage, "cached_tokens", 0) or 0
total = getattr(usage, "input_tokens", 0) or 0
return {"input": total, "cached": cached,
"hit_rate": cached / max(total, 1)}
```
字段名在不同 SDK 版本里可能不一样,以官方文档为准。提升命中率的运维动作:
- 网关按"系统提示前缀的哈希"做粘性路由,让同一套前缀的请求尽量落到同一处理路径;
- 同一批任务集中时间跑,避免请求太稀疏、缓存过期;
- 上线前先发一批携带完整前缀的预热请求;
- 缓存写入本身通常也计费,命中率低的时候缓存可能是负收益,用步骤二的脚本算总账。
步骤四:回归测试
评测分三层,从便宜到贵:
1. 结构断言:输出能否解析成 JSON、必需字段是否齐全、枚举值是否合法;
2. 规则断言:数字算得对不对、引用的内容是否来自给定材料;
3. 模型裁判:主观项用一个模型按评分标准打分,只用在无法规则化的地方。
```python
regression.py
import json, statistics, sys
CASES = [json.loads(line) for line in open("eval/cases.jsonl")]
THRESHOLD = 0.90
def check(case, resp):
if case["expect"] == "json":
try:
obj = json.loads(resp.text)
except Exception:
return False
return all(k in obj for k in case["required_keys"])
return case["must_contain"] in resp.text
def run(model, case):
resp = call_model(model, case["messages"], temperature=0)
return {"ok": check(case, resp), "text": resp.text, "usage": resp.usage}
report = {}
for model in ("gpt-5", "gpt-6-sol", "gpt-6-luna"):
rows = [run(model, c) for c in CASES]
report[model] = {
"pass_rate": sum(r["ok"] for r in rows) / len(rows),
"avg_out_tokens": statistics.mean(r["usage"].output_tokens for r in rows),
}
print(json.dumps(report, indent=2, ensure_ascii=False))
if report["gpt-6-sol"]["pass_rate"] < THRESHOLD:
sys.exit(1)
```
几个经验:
- 样例从线上真实请求采样,必须包含长尾和历史上失败的输入,不能只放顺手写的好例子;
temperature=0只是降低随机性,不等于可复现;关键任务同一条跑几次取多数;- 保存迁移前的输出作为基线,跑完做 diff,人工看二十条变化最大的样例;
- 评测集固定版本号,改了要留记录,否则指标没法纵向比。
步骤五:路由层与分批灰度
把模型选择从业务代码里抽出来,集中到路由层。
```yaml
routing.yaml
defaults:
model: gpt-6-sol
fallback: gpt-5
tasks:
classify_ticket: {model: gpt-6-luna}
extract_invoice: {model: gpt-6-luna}
agent_planning: {model: gpt-6-sol}
long_context_qa:
model: gpt-6-sol
canary: {experiment: sol-rollout, steps: [1, 5, 25, 50, 100]}
```
```python
router.py
import hashlib, yaml
CFG = yaml.safe_load(open("routing.yaml"))
def bucket(key: str) -> int:
稳定哈希:同一个用户始终落在同一桶,避免体验来回抖
h = hashlib.sha256(key.encode()).hexdigest()
return int(h[:8], 16) % 100
def pick_model(task, key, canary_percent=None):
cfg = dict(CFG["defaults"])
cfg.update(CFG["tasks"].get(task, {}))
if canary_percent is not None and bucket(key) >= canary_percent:
return cfg["fallback"]
return cfg["model"]
```
灰度节奏:
- 影子阶段:新模型只跑不返回,落库比对输出,用户无感;
- 1% 到 5%:盯错误率、超时率、schema 校验失败率;
- 25% 到 50%:盯业务指标——任务完成率、转人工率、重试率、点踩率;
- 100%:每一档至少观察一个完整业务周期再往上加。
回滚就一件事:把灰度比例归零,或把默认模型改回旧值,热更新下发。
```bash
用现有配置中心的等价命令,路径与字段按你的系统改
curl -X PATCH "$CONFIG_ENDPOINT/v1/configs/llm-routing" \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"defaults":{"model":"gpt-5"},"canary_percent":0}'
```
回滚能力的要求:分钟级生效、入口只有一个、值班的人知道怎么点。旧模型的代码路径至少保留一个迭代周期再清理。
常见坑与排错
缓存命中率上不去。 用 grep 在提示模板里搜时间、ID、随机数这类字段,多半能找到。其次是 JSON 键序不稳定,统一 sort_keys=True。再就是请求被分散到不同处理路径上,检查网关的粘性策略。缓存的作用范围、TTL 与计价口径以官方文档为准。
迁移后输出格式变了。 结构化输出、工具调用 schema、停止序列在新旧模型上的严格程度不完全一致,先用步骤四的结构断言跑一遍,再把"模型会自己输出 markdown"这类隐含假设改成显式要求。
token 数对不上。 不同模型的切词方式不同,同样的输入 token 数会变。预算按实测 token 算,不要按字符数估。账单要用缓存命中部分和新增部分拆开看。
技术指标好但用户不满。 只看了延迟和错误率,漏了业务指标。把转人工率、重试率、点踩率加进看板。另外分桶不要用 random(),否则同一用户一会儿新一会儿旧,体验会抖。
回滚回不干净。 业务代码里残留硬编码的模型名,或者提示词只针对新模型调过、回滚后效果反而变差。做法是提示词按目标模型分版本,或者保持一段时间的双兼容。
参数不通用。 推理强度、思考预算这类参数名在新旧模型上不一定一致,逐项对照官方文档,缺项就先用默认值,别硬塞。
下一步建议
- 把回归脚本接进 CI,改提示词或换模型自动跑,不通过就挡住发布;
- 建一个"每完成任务成本"看板,把缓存命中率、重试率、各档位流量占比放在同一屏;
- 每条请求记录模型名和提示词版本号,出问题能按版本回放;
- 想进一步省成本,可以试级联:先走轻量档位,置信度低再升级到高能力档位;
- 定期重看官方定价页与模型列表,档位映射不是一次定死的事,跟着业务量和价格调整即可。
