这篇能做出什么
读完并照着做，你会得到一个能跑的双模型对照环境：同一份业务代码，只换一个模型 ID 和一个适配层，就能在旧的 3.8 Flash 和 Gemini 4 Hydrogen 之间来回切换；你会有一份记录延迟、token 用量、错误码的实测数据表；还会拿到一份针对复杂问答的评测清单，用来判断哪些场景可以直接切、哪些必须保留旧模型兜底。文中所有模型 ID、接口地址、配额数字都以官方文档当前版本为准，代码里全部走环境变量，你按文档填进去即可。
前置条件清单
1. 两个模型各自可用的 API Key，以及至少一个能跑 Python 3.10+ 的环境。
2. 官方文档里的三样东西：接口基址、新模型的模型 ID、免费层的限额说明（每分钟请求数、每分钟 token 数、每日请求数通常各有一档）。
3. 一个能发 HTTP 请求的工具（curl 或 Postman），方便先把链路打通再写代码。
4. 一份你线上真实的 20～30 条提问样本，包含简单事实问答、多轮追问、长文摘要、需要推理的复杂问题各若干条。
5. 一个记录用的表格（CSV 就行），字段建议：问题 ID、模型、耗时、输入 token、输出 token、是否报错、人工评分。
第一步：把模型名和地址从代码里拔出来
迁移翻车的头号原因是模型 ID 硬编码散落在十几个文件里。先做集中配置。
```python
config.py
import os
API_BASE = os.environ["API_BASE"] # 以官方文档当前版本为准
API_KEY = os.environ["API_KEY"]
MODEL_NEW = os.environ["MODEL_ID"] # Gemini 4 Hydrogen 的模型 ID
MODEL_OLD = os.environ.get("OLD_MODEL_ID", "") # 3.8 Flash 的模型 ID
TIMEOUT = float(os.environ.get("TIMEOUT", "60"))
```
```bash
export API_BASE="按官方文档填写"
export API_KEY="你的 key"
export MODEL_ID="按官方文档填写"
export OLD_MODEL_ID="按官方文档填写"
```
改完之后，全仓库搜一遍旧的模型字符串，确认只剩配置文件和测试夹具里出现。
第二步：用 curl 打通最小请求
不要一上来就改业务代码，先用最简单的请求确认 key、地址、模型 ID 三者匹配。
```bash
curl -sS "${API_BASE}/models/${MODEL_ID}:generateContent" \
-H "Content-Type: application/json" \
-H "x-goog-api-key: ${API_KEY}" \
-d '{
"contents": [
{"role": "user", "parts": [{"text": "用三句话解释什么是AI 词典：向量数据库">向量数据库"}]}
]
}'
```
如果官方文档要求用 Authorization: Bearer <key> 或者把 key 放在查询参数里，就按文档改这一行，别的地方不用动。返回 200 且能看到文本，说明链路通了；返回 404 大概率是模型 ID 写错，返回 401/403 是鉴权方式不对，这两类错误要在这一步排掉，不要带进后面的调试。
第三步：对齐请求体的结构差异
这是迁移里最费时间的部分。新旧两代接口在“消息怎么组织”这件事上通常不一样，差异集中在下面几处，逐条对照官方文档确认：
| 维度 | 旧式调用常见形态 | 新接口常见形态 |
|---|---|---|
| 消息数组 | messages，角色为 system/user/assistant | contents，角色为 user/model |
| 系统提示 | 作为 messages 的第一条 | 独立的 systemInstruction 字段 |
| 输出长度 | max_tokens | maxOutputTokens（放在 generationConfig 里） |
| 随机性 | temperature 顶层字段 | generationConfig.temperature |
| 停止词 | stop | stopSequences |
| 结构化输出 | response_format | responseMimeType + schema 字段 |
写一个转换函数，把旧格式统一转成新格式，业务层就不用改：
```python
def to_new_payload(messages, max_tokens=1024, temperature=0.3):
system_parts, contents = [], []
for m in messages:
role = m["role"]
if role == "system":
system_parts.append(m["content"])
else:
contents.append({
"role": "model" if role == "assistant" else "user",
"parts": [{"text": m["content"]}],
})
payload = {
"contents": contents,
"generationConfig": {
"maxOutputTokens": max_tokens,
"temperature": temperature,
},
}
if system_parts:
payload["systemInstruction"] = {"parts": [{"text": "\n".join(system_parts)}]}
return payload
```
注意两个细节：一是多轮对话里角色的名字变了，assistant 要映射成 model，否则会报角色非法；二是系统提示不要继续塞进 contents 的第一条，否则模型的遵循度会下降，尤其在长系统提示下差别明显。
第四步：对齐响应解析
旧接口一般取 choices[0].message.content，新接口通常是 candidates[0].content.parts[*].text，而且 parts 可能有多段。解析要写成容错的：
```python
def extract_text(resp: dict) -> str:
cands = resp.get("candidates") or []
if not cands:
fb = resp.get("promptFeedback", {})
raise RuntimeError(f"无候选结果，原因: {fb.get('blockReason', 'unknown')}")
cand = cands[0]
texts = [p.get("text", "") for p in cand.get("content", {}).get("parts", [])]
text = "".join(texts).strip()
if not text:
raise RuntimeError(f"空文本，finishReason={cand.get('finishReason')}")
return text
```
finishReason 一定要打日志。常见值里，正常结束、达到长度上限、被安全策略拦截是三种完全不同的情况，但表现都是“回答不对劲”。上线前把这三个分支的日志都跑一遍。
第五步：流式与重试
流式响应的分块结构也变了，别沿用旧的解析器。用增量拼接的方式写，一次性拿到完整文本再落库：
```python
import json, time, random, requests
from config import API_BASE, API_KEY, MODEL_NEW, TIMEOUT
def call_new(payload, retries=4):
url = f"{API_BASE}/models/{MODEL_NEW}:generateContent"
headers = {"Content-Type": "application/json", "x-goog-api-key": API_KEY}
for i in range(retries):
r = requests.post(url, headers=headers, json=payload, timeout=TIMEOUT)
if r.status_code == 200:
return r.json()
if r.status_code in (429, 500, 502, 503, 504):
wait = float(r.headers.get("Retry-After", 2 ** i)) + random.random()
time.sleep(min(wait, 30))
continue
r.raise_for_status()
raise RuntimeError("重试耗尽")
```
429 要优先读 Retry-After 头，服务端给的等待时间比自己猜的准。别用固定 1 秒重试，免费层很容易被打成连续 429。
第六步：弄清免费配额到底卡在哪
免费层通常同时限制三个维度：每分钟请求数、每分钟 token 数、每日请求数，任一超限都会返回 429，但“什么时候恢复”不一样——按分钟限的几十秒后就恢复，按天限的要等配额重置。具体数值以官方的配额与定价页面当前版本为准，不要照抄博客里的数字。
工程上的三条对策：
1. 本地令牌桶。发请求前先扣额度，宁可排队也不要撞墙。
```python
import time
class RateLimiter:
def __init__(self, per_minute: int):
self.interval = 60.0 / max(per_minute, 1)
self.next_at = 0.0
def acquire(self):
now = time.monotonic()
if now < self.next_at:
time.sleep(self.next_at - now)
self.next_at = max(now, self.next_at) + self.interval
```
2. 加缓存。评测和调试阶段同一条问题会被反复打，用问题文本的哈希做 key 缓存结果，能把请求量降一个量级。
3. 长文本先压缩。输入 token 也计入配额，把不需要的上下文裁掉，比换模型更省钱。
第七步：复杂问答实测怎么做
不要凭感觉说“新的更聪明”。用同一份题集、同一套提示词、同样的温度参数">温度参数跑两遍，只换模型。
```python
import json, time
from config import MODEL_OLD
questions = [json.loads(l) for l in open("eval.jsonl", encoding="utf-8")]
with open("result.jsonl", "w", encoding="utf-8") as f:
for q in questions:
for tag, model in (("old", MODEL_OLD), ("new", MODEL_NEW)):
payload = to_new_payload(
[{"role": "user", "content": q["prompt"]}],
max_tokens=2048, temperature=0.2)
t0 = time.time()
try:
resp = call_new(payload) # 旧模型换成对应调用函数
text, err = extract_text(resp), ""
except Exception as e:
text, err = "", str(e)
f.write(json.dumps({
"qid": q["id"], "model": tag, "cost_s": round(time.time() - t0, 2),
"answer": text, "error": err,
}, ensure_ascii=False) + "\n")
```
题集按四档设计，每档 5～8 条：单点事实（考准确性）、多约束指令（考指令遵循）、多跳推理（考思维链）、长文摘要与抽取（考长上下文）。评分用手工三分制：0 分错，1 分部分正确，2 分完全正确，再单列一栏“是否有编造”。
判断标准要提前定死，建议是：准确性不下降、指令遵循不下降、复杂推理档提升才意味着可以迁移。如果只有简单题变快、难题目持平或变差，那就保留旧模型做兜底，用路由按问题类型分流。
第八步：灰度切换
先影子流量：线上请求照旧走老模型，同时异步复制一份给新模型，只记录不返回。跑够一两天，对比错误率、P95 延迟、人工抽检得分，再按 5%→20%→50% 的比例切换。每一步都保留一键回滚开关，回滚只改配置，不发版。
常见坑与排错
- 429 但配额页面看着没超：多半是 token 维度超了，短请求多也可能被每分钟 token 限卡住，看返回体里的配额字段。
- 返回空字符串：先看
promptFeedback.blockReason和finishReason，安全拦截和长度截断都会给出空文本。 - 多轮对话越聊越傻：检查历史是否被完整回传，以及 assistant 角色有没有映射成 model。
- 中文回答被截断在半句：
maxOutputTokens给小了，中文一个汉字占的 token 比英文多，同长度中文要留更多预算。 - 输出 JSON 解析失败：开了结构化输出仍然要写容错解析，模型偶尔会带解释性前后缀。
- 延迟忽高忽低：长提示词的首 token 延迟本来就高，用 P95 而不是平均值评估。
- 换 key 后全部 401：确认 key 是给新接口用的，有些平台的 key 分作用域。
- 本地测试全过、上线就挂：检查生产环境的环境变量有没有同步，尤其是模型 ID。
下一步建议
把这套对照框架沉淀成团队内部的评测脚本，每换一次模型就跑一次，不要每次重新搭。接着做三件事：给不同问题类型配置路由规则，让简单请求继续走低成本模型；把缓存层从内存换成共享存储，多实例复用；把 finishReason、429 次数、P95 延迟接进监控看板，用数据而不是印象判断迁移是否真的划算。