1. 生产环境里的“暗礁”:ChatGPT 也会突然“降智”
把 ChatGPT 接进业务后,最头疼的不是第一次上线,而是“今天上线好好的,明天就翻车”。
典型症状有三:
- 回答质量忽高忽低,同一 prompt 上午 90 分,下午 60 分;
- 多轮对话里突然“失忆”,把用户刚说过的话当成新话题;
- 指令跟随能力下降,few-shot 示例明明写得很清楚,模型却开始“自由发挥”。
这些问题往往跟模型版本、系统 prompt、温度参数甚至并发压力有关。人工刷 Case 只能看到冰山一角,必须让测试流水线 7×24 盯着,才能第一时间发现“降智”。
2. 人工评估 vs 自动化降智测试
| 维度 | 人工评估 | 自动化降智测试 |
|---|---|---|
| 成本 | 高,需领域专家 | 一次开发,长期复用 |
| 粒度 | 随机抽检,覆盖低 | 全量回归,可压测 |
| 一致性 | 不同人打分偏差大 | 统一指标,历史可对比 |
| 反馈速度 | 天级 | 分钟级 |
降智测试的核心是“三维评估”:
- 事实准确性:答案是否违背常识或业务知识库;
- 逻辑连贯性:多轮上下文是否自洽;
- 指令跟随能力:是否按格式、字数、语言风格输出。
三者任一跌破阈值即触发告警,避免“看起来通顺却在偷换概念”的假阳性通过。
3. 测试框架(Python 3.8+)核心代码
目录结构:
gpt_degrade/ ├── cases/ # 参数化模板 │ └── qa_yaml ├── eval/ │ ├── bleu_scorer.py │ └── human_label.py ├── history/ │ └── version_db.json ├── runner.py └── utils.py3.1 测试用例生成模块
支持 YAML 模板 + 参数注入,方便业务同学随时加 Case,无需改代码。
# cases/template_loader.py import yaml, random, pathlib from string import Template def load_prompt_tpl(name: str) -> Template: tpl = pathlib.Path(__file__).with_name(f"{name}.yaml").read_text() return Template(tpl) def gen_cases(tpl_name, param_grid: dict, n: int = 50) -> list[dict]: """ 按网格随机组合生成 n 条用例 param_grid: {"topic": ["体育", "财经"], "style": ["严谨", "幽默"]} """ tpl = load_prompt_tpl(tpl_name) cases = [] keys = list(param_grid.keys()) for _ in range(n): combo = {k: random.choice(param_grid[k]) for k in keys} cases.append({"prompt": tpl.substitute(combo), "meta": combo}) return casesYAML 示例(qa_yaml):
system: "你是一名{style}风格的客服机器人,请用 50 字内回答" user: "{topic} 领域最新热点是什么?"3.2 结果评估算法
BLEU 快速初筛,人工标注二次校准。
# eval/bleu_scorer.py from nltk.translate.bleu_score import sentence_bleu from eval.human_label import query_labeler # 拉取众包标注 def hybrid_score(pred: str, ref: str, case_id: str) -> float: bleu = sentence_bleu([ref.split()], pred.split()) human = query_labeler(case_id) # 0/1/2 三档人工分 # 人工权重 0.7,BLEU 权重 0.3 return 0.3 * bleu + 0.7 * human3.3 历史版本对比
把每次跑分写进history/version_db.json,支持回滚决策。
# history/version_store.py import json, datetime as dt from pathlib import Path DB = Path(__file__).with_name("version_db.json") def append_score(version: str, dim: str, score: float): ts = dt.datetime.utcnow().isoformat() with DB.open("a") as f: f.write(json.dumps({"v": version, "dim": dim, "s": score, "ts": ts}) + "\n") def degrade_ratio(last_v: str, dim: str, window: int = 7) -> float: """最近 window 天相对 last_v 的下降幅度""" rows = [json.loads(l) for l in DB.read_text().splitlines()] base = [r["s"] for r in rows if r["v"] == last_v and r["dim"] == dim] curr = [r["s"] for r in rows[-window:] if r["dim"] == dim] if not base or not curr: return 0.0 return (sum(base)/len(base) - sum(curr)/len(curr)) / sum(base)*len(base)4. 性能优化三板斧
4.1 异步并发测试
使用asyncio+aiohttp把 1k 条用例压到 2 分钟跑完。
# runner.py import asyncio, aiohttp, os, time from cases.template_loader import gen_cases sem = asyncio.Semaphore(50) # 并发 50 async def req_one(session, payload: dict) -> dict: async with sem: async with session.post( url=os.getenv("GPT_ENDPOINT"), json=payload, headers={"Authorization": f"Bearer {os.getenv('TOKEN')}"} ) as resp: return await resp.json() async def batch_run(cases: list[dict]) -> list[dict]: async with aiohttp.ClientSession() as session: tasks = [req_one(session, c) for c in cases] return await asyncio.gather(*tasks) if __name == "__main__": t0 = time.time() results = asyncio.run(batch_run(gen_cases("qa_yaml", param_grid={...}, n=1000))) print("QPS:", 1000/(time.time()-t0))4.2 缓存机制
对完全相同的 prompt 做 MD5 缓存,减少重复调用,节省预算。
# utils.py import hashlib, redis, json r = redis.Redis(host="localhost", decode_responses=True) def cache_key(prompt: str) -> str: return "gpt:cache:" + hashlib.md5(prompt.encode()).hexdigest() def cached_call(payload: dict, ttl: int = 3600): key = cache_key(payload["prompt"]) if (hit := r.get(key)): return json.loads(hit) resp = real_call(payload) # 真实请求 r.set(key, json.dumps(resp), ex=ttl) return resp4.3 分布式压测
用 Locust 起多进程 slave,把 QPS 再抬一个量级。
# locustfile.py from locust import HttpUser, task class GPTUser(HttpUser): @task def ask(self): self.client.post("/v1/chat/completions", json={"model":"gpt-3.5-turbo","messages":[]}, headers={"Authorization":"Bearer ***"})命令:
locust -f locustfile.py --master-host=10.0.0.1 --worker --users 20005. 安全与合规
5.1 敏感词过滤
采用双重策略:本地前缀树 + 云端审核 API,确保政治、暴力、歧视等内容不出现在测试 prompt 里。
# safety/word_filter.py import ahocorasick A = ahocorasick.Automaton() for w in open("sensitive.txt"): A.add_word(w.strip(), w.strip()) A.make_automaton() def filter(prompt: str) -> str: for end_index, word in A.iter(prompt): prompt = prompt.replace(word, "*" * len(word)) return prompt5.2 测试数据脱敏
把手机号、邮箱、身份证全部用 Faker 生成假数据,并在日志里打敏。
# safety/faker_proxy.py from faker import Faker fk = Faker(locale="zh_CN") def fake_profile() -> dict: return {"phone": fk.phone_number(), "id": fk.ssn()}6. 最佳实践清单
- 测试频率:核心模型每日全量,灰度模型每 4h 抽测 100 条;
- 阈值设定:BLEU<0.4 或人工 0 档占比>5% 即报警;
- 报警通道:飞书群 + Webhook,附带 degrade_ratio 曲线图;
- 版本基线:任何新模型上线前,必须跑完过去 7 天全量回归, degrade_ratio<2% 才允许切流。
7. 一个开放问题
当业务里同时跑 ChatGPT、Claude、自研模型时,如何建立跨模型可比评估体系?
BLEU 偏向 n-gram 匹配,对不同 token 长度不友好;人工标注又受语言风格偏好影响。
是否该引入“任务成功率”作为统一 North Star?或者把奖励模型做成仲裁器?
欢迎留言聊聊你的做法。
把上面的流水线跑通后,我最大的感受是:“降智”不是模型变笨,而是我们没有及时看见它变笨。”
如果你也想亲手搭一套实时对话 AI,顺便把这套降智测试框架直接嵌进去,可以从这个动手实验开始:从0打造个人豆包实时通话AI。
实验把 ASR→LLM→TTS 整条链路都封装好了,你只需专注在“让 AI 不翻车”这件事上。小白也能跟着 README 一步步跑起来,我亲测一下午就搞定。祝调试愉快,少踩坑!