1. 指令微调翻车现场:数据没洗,训完模型只会复读
大模型指令遵从微调这件事,很多人卡在第一步不是显存不够,而是数据太脏。我见过最典型的场景:拿一份爬来的客服对话直接丢进 SFTTrainer,训了 3 个 epoch,loss 降到 0.3,结果推理时模型开始复读用户问题,或者把 JSON 格式吐成半截。排查半天发现训练集里混着大量 HTML 标签、重复样本、以及 assistant 回复为空的行。
指令微调的数据工程其实分两段:清洗和合成。清洗解决"原始语料里的噪声与格式错误",合成解决"高质量指令对不够用"。这两段做完,才轮到配置训练脚本。而中间所有需要调用大模型 API 的环节——比如用 LLM 做数据质量打分、批量合成指令对、跑合成数据的抽检——如果每个脚本都单独配一套 Key,管理成本会爆炸。这篇就用 TaoToken 的统一 Key 把清洗、合成、抽检这条链路串起来,配置示例落在 Cline 的 settings.json 里,方便你边写脚本边调 API。
适合谁看:已经跑通过一次 SFT、但被数据质量拖累效果的开发者;或者正准备做指令微调、想先把数据管线搭对的人。下面所有脚本都可直接复制改路径运行。
2. TaoToken 前置:一个 Key 覆盖清洗与合成两段调用
数据工程里调用大模型 API 的频次其实比训练还高。清洗阶段要用模型判断"这条指令是否语义完整",合成阶段要用模型批量生成指令对,抽检阶段还要用模型给合成数据打分。如果清洗脚本用一家、合成脚本用另一家,Key 散落在不同 .env 里,换机器就崩。
TaoToken 在这里的角色是统一入口:一个 Key、一个 base_url,兼容 OpenAI 风格的 chat/completions 接口,清洗脚本和合成脚本共用同一份配置。它的 API 地址是https://taotoken.net/api,注意这个地址不带任何查询参数,直接作为 base_url 用。
你需要先拿到 Key。登录后在控制台的 API Keys 页面创建,建议按用途分 Key:一个给清洗脚本、一个给合成脚本,方便单独看用量。创建入口在这里:
API Keys 管理:https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite
拿到 Key 之后,先别急着写清洗脚本,用模型对话页面手动发一条测试请求,确认 Key 和网络都通:
模型对话测试:https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite
这一步能省掉后面 80% 的"脚本报 401 但不知道是 Key 错还是网络错"的排查时间。确认能正常返回后,把 Key 存进环境变量,别硬编码进脚本:
export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"如果你后续要长期跑数据合成任务,或者把清洗+合成做成 Agent 自动流转,可以看下 Coding Plan 的额度方案,比按次调用更适合批量任务:
Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite
3. 可复制配置:清洗脚本骨架 + 合成模板 + Cline settings.json
3.1 数据清洗脚本骨架
清洗的核心逻辑是:读原始 JSONL → 逐条做规则过滤 → 对规则拿不准的样本调模型判断 → 输出干净数据集。下面这个骨架可以直接跑,规则部分按你的语料改。
import json import re import os from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], ) # 规则层:先干掉明显脏数据,减少 API 调用量 def rule_filter(item): instruction = item.get("instruction", "").strip() output = item.get("output", "").strip() # 空指令或空回复直接丢 if not instruction or not output: return False # 含 HTML 标签的丢 if re.search(r"<[^>]+>", instruction + output): return False # 回复过短(少于 10 字)大概率是噪声 if len(output) < 10: return False # 指令和回复完全相同的重复样本 if instruction == output: return False return True # 模型层:规则拿不准的,让模型判断语义完整性 def llm_judge(item): prompt = f"""判断下面这条指令微调样本是否可用。 指令:{item['instruction']} 回复:{item['output']} 只回答 YES 或 NO,YES 表示指令清晰且回复完整回答了指令。""" resp = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0, max_tokens=5, ) return "YES" in resp.choices[0].message.content.upper() def clean(input_path, output_path): kept, dropped = 0, 0 with open(input_path, "r", encoding="utf-8") as fin, \ open(output_path, "w", encoding="utf-8") as fout: for line in fin: item = json.loads(line) if not rule_filter(item): dropped += 1 continue if not llm_judge(item): dropped += 1 continue fout.write(json.dumps(item, ensure_ascii=False) + "\n") kept += 1 print(f"保留 {kept} 条,丢弃 {dropped} 条") if __name__ == "__main__": clean("raw_data.jsonl", "clean_data.jsonl")这里有个省钱的细节:规则层先过滤掉 60% 以上的明显脏数据,模型层只处理剩下的。我试过在 5 万条语料上跑,规则层干掉 3.2 万条,模型层只调了 1.8 万次,成本直接砍掉一大半。
3.2 合成模板配置
清洗完的数据量往往不够,尤其是垂直领域。合成阶段用模板批量生成指令对,模板配置单独放一个 JSON,方便改:
{ "task_type": "instruction_synthesis", "system_prompt": "你是一个指令数据合成助手。根据给定的领域和种子内容,生成一条高质量的指令-回复对。", "user_template": "领域:{domain}\n种子内容:{seed}\n请生成一条指令和对应的详细回复,输出 JSON 格式:{{\"instruction\": \"...\", \"output\": \"...\"}}", "domains": ["客服问答", "代码解释", "数据分析"], "seeds_per_domain": 50, "temperature": 0.8, "max_tokens": 1024 }合成脚本读这个配置,循环调用 API:
import json from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], ) def synthesize(config_path, output_path): cfg = json.load(open(config_path, encoding="utf-8")) results = [] for domain in cfg["domains"]: for i in range(cfg["seeds_per_domain"]): user_msg = cfg["user_template"].format( domain=domain, seed=f"seed_{i}" ) resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": cfg["system_prompt"]}, {"role": "user", "content": user_msg}, ], temperature=cfg["temperature"], max_tokens=cfg["max_tokens"], ) content = resp.choices[0].message.content try: pair = json.loads(content) pair["domain"] = domain results.append(pair) except json.JSONDecodeError: print(f"解析失败,跳过:{content[:50]}") with open(output_path, "w", encoding="utf-8") as f: for r in results: f.write(json.dumps(r, ensure_ascii=False) + "\n") print(f"合成 {len(results)} 条") if __name__ == "__main__": synthesize("synth_config.json", "synth_data.jsonl")3.3 Cline settings.json 接入配置
如果你用 Cline 做数据脚本的辅助编写和调试,可以在 settings.json 里把 TaoToken 配成自定义 provider,这样写清洗脚本时能直接在编辑器里问模型:
{ "cline.apiProvider": "openai", "cline.openAiBaseUrl": "https://taotoken.net/api", "cline.openAiApiKey": "sk-你的key", "cline.openAiModelId": "gpt-4o-mini", "cline.customInstructions": "你是数据工程助手,回答时优先给可运行的 Python 代码。" }配置完重启 Cline,在侧边栏发一条"帮我写一个 JSONL 去重脚本",能正常返回就说明接入成功。注意 base_url 末尾不要加/v1,TaoToken 的接口路径已经处理好了。
4. 验证请求:清洗前后对比与合成数据抽检
4.1 清洗前后样本对比
拿一条真实脏数据看效果。清洗前:
{"instruction": "<p>怎么退款</p>", "output": "退款"}这条被规则层干掉:含 HTML 标签,且回复只有 2 个字。清洗后保留的样本长这样:
{"instruction": "订单超过7天还能退款吗", "output": "超过7天的订单需要联系人工客服,提供订单号和退款原因,客服会在24小时内审核。如果商品未拆封,通常可以退;已拆封的视商品类型而定。"}对比很明显:清洗后的指令有具体场景,回复有可执行信息。你可以写个快速统计脚本看清洗前后的长度分布:
import json def stats(path): lengths = [] for line in open(path, encoding="utf-8"): item = json.loads(line) lengths.append(len(item["output"])) print(f"条数 {len(lengths)},平均回复长度 {sum(lengths)/len(lengths):.0f},最短 {min(lengths)},最长 {max(lengths)}") stats("raw_data.jsonl") stats("clean_data.jsonl")清洗前平均回复长度往往被大量短噪声拉低,清洗后会明显上升。
4.2 合成数据质量抽检
合成数据不能直接用,必须抽检。抽检动作分两步:先随机抽 50 条,再用模型给每条打分。
import json import random from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], ) def sample_and_score(path, n=50): lines = open(path, encoding="utf-8").readlines() samples = random.sample(lines, min(n, len(lines))) scores = [] for line in samples: item = json.loads(line) prompt = f"""给下面这条指令微调样本打分(1-5分)。 指令:{item['instruction']} 回复:{item['output']} 评分标准:指令清晰度、回复完整性、是否有事实错误。 只输出一个数字。""" resp = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0, max_tokens=3, ) try: score = int(resp.choices[0].message.content.strip()) scores.append(score) except ValueError: continue avg = sum(scores) / len(scores) if scores else 0 print(f"抽检 {len(scores)} 条,平均分 {avg:.2f}") return avg if __name__ == "__main__": sample_and_score("synth_data.jsonl")平均分低于 3.5 就说明合成模板需要调,通常是 system_prompt 太宽泛,或者 temperature 太高导致回复发散。把 temperature 从 0.8 降到 0.5 再跑一轮,分数一般能上来。
5. 本篇常见错排查
报错 401 Unauthorized:九成是 Key 没读到环境变量。在脚本里加一行print(os.environ.get("TAOTOKEN_API_KEY")[:8])确认前 8 位,如果是 None 说明 export 没生效,检查是不是在子 shell 里跑的。
合成数据 JSON 解析失败率高:模型返回的内容带了 markdown 代码块标记,比如 ```json 开头。在解析前先 strip 掉:
content = content.strip().removeprefix("```json").removesuffix("```").strip()清洗脚本跑一半卡住:大概率是 API 限流。给 client 加超时和重试:
from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], timeout=30.0, max_retries=3, )Cline 里配置后模型不响应:检查 settings.json 的openAiBaseUrl是不是写成了https://taotoken.net/api/v1。正确写法不带/v1,因为接口路径已经包含在 base_url 里了。改完记得完全重启 Cline,不是 reload window。
清洗后数据量骤降:先看规则层是不是太激进。把rule_filter里每条规则单独打日志,统计各规则干掉多少条。如果某条规则干掉超过 40%,说明阈值设错了,比如len(output) < 10对短回复场景就太严。
6. 把 Key 收拢到一处,数据管线才跑得顺
数据清洗和合成这两段,本质上是同一件事的两面:清洗是减法,合成是加法,中间都要反复调模型。如果 Key 分散在清洗脚本、合成脚本、抽检脚本三个地方,改一次配置要动三个文件,跑批量任务时哪个脚本挂了都难定位。
用 TaoToken 统一 Key 之后,三个脚本共用TAOTOKEN_API_KEY和TAOTOKEN_BASE_URL两个环境变量,换机器只需要重新 export 一次。清洗脚本的规则层和模型层分离、合成模板外置成 JSON、抽检动作固定成 50 条随机采样,这套骨架跑通一次之后,换任何语料都只是改路径和模板的事。
接入文档里有完整的接口参数说明,配 Cline 或写脚本时对着看:
接入文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
如果你要把清洗+合成做成定时任务,或者接进 Agent 自动流转,Coding Plan 的额度比单次调用更适合这种持续跑的场景。先把上面三个脚本跑通,再考虑自动化,顺序别反。