Trae 生成的 sft_dataset_v3.jsonl 训练集，用 Python asyncio 校验 SYSTEM_PROMPT 还缺什么？

1. 先搞清楚 sft_dataset_v3.jsonl 到底缺什么你手上这份C:\dumatework\duan\tools\ai_copilot\sft_dataset_v3.jsonl是 Trae 帮你生成的段言翻译 SFT 训练集1136 行、35 个语法类别。表面看覆盖面挺全但真正拿去训模型之前得先回答三个问题字段有没有缺、SYSTEM_PROMPT 里规定的语法点是不是每条都有样本、有没有重复样本在稀释梯度。这三个问题靠肉眼翻 JSONL 是不现实的。1136 行里混着 60 行字符串字面量被误声明成变量、11 行//翻译不一致、6 行「定义 X 等于」误用这些错误如果不在训练前筛出来模型学到的就是坏习惯。所以这篇的核心不是「怎么生成数据」而是「怎么用 Python asyncio 并发把这份数据集体检一遍」。适合谁看已经用 Trae 或类似工具产出了 SFT 数据集、准备送进训练管线、但不确定数据质量的人。你需要会一点 Python能跑脚本能看懂 JSONL 结构。不需要你懂分布式训练校验脚本本身是单机跑的。我试过把 1136 行串行读一遍做正则匹配大概要 40 秒左右而且每加一个检查项就线性变慢。改成 asyncio 并发之后同样的检查项压到 3 秒以内后面再加规则也不心疼。下面把整套骨架拆开讲你可以直接复制去改。先明确校验目标分四类字段完整性——每行必须有instruction、input、output三个字段且都不能为空字符串。SYSTEM_PROMPT 覆盖——把 SYSTEM_PROMPT 里声明的语法规则抽成检查项逐条统计命中样本数找出 0 命中和低命中的。去重——input完全相同的行要合并input高度相似比如只差变量名的也要标记。质量规则——字符串字面量误声明、//翻译、定义 X 等于误用这些具体错误模式。这四类里前两类是结构性的后两类是语义性的。asyncio 适合做的是 I/O 密集和规则匹配密集的部分纯 CPU 的正则其实提升有限但配合批量读取和并发任务调度整体吞吐还是明显好于串行。2. 用 TaoToken 补齐校验脚本里的模型判断环节纯正则能查出「设 Hello 为 空」这种模式但查不出「这条样本的 output 语义上是不是真的在翻译 input」。后者需要模型判断。我的做法是把规则匹配和模型判断分开规则匹配用 asyncio 并发跑模型判断走 TaoToken 的 API同样用 asyncio 并发发请求。TaoToken 在这里的角色是提供一个统一的模型调用入口。你不需要在本地部署模型也不用为每个判断任务单独配一套 SDK。校验脚本里需要模型介入的地方主要有两处一是判断某条 output 是否语义等价于 input 的翻译去重时用二是判断 SYSTEM_PROMPT 里某条规则在样本里是否被正确执行覆盖检查时用。接入方式很简单拿到 API Key 之后Base URL 填https://taotoken.net/apiModel ID 按你选的模型填。如果你还没建 Key去控制台建一个https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentconsole建完 Key 之后校验脚本里用httpx.AsyncClient发请求就行。注意一点模型判断是辅助不要让它替代规则匹配。规则能查的比如字段缺失、//翻译一律用规则模型只用来处理规则覆盖不到的语义问题。这样既快又省 token。如果你打算长期跑这类数据校验建议用 Coding Plan额度更稳https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding-plan模型对话入口在这里可以先手动试几条样本判断效果https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentchat3. 可复制的校验脚本骨架与 config.toml 配置先给配置文件。放在项目根目录和脚本同级# config.toml [dataset] path C:/dumatework/duan/tools/ai_copilot/sft_dataset_v3.jsonl encoding utf-8 required_fields [instruction, input, output] [concurrency] max_workers 16 batch_size 64 [taotoken] base_url https://taotoken.net/api api_key sk-你的key model_id 你的模型ID timeout 30 [checks] enable_field_check true enable_prompt_coverage true enable_dedup true enable_quality_rules true [quality_rules] # 字符串字面量误声明模式 literal_var_pattern 设\\s([A-Z][a-zA-Z]*)\\s为\\s空 # // 翻译检查 floor_div_pattern // # 定义 X 等于 误用 define_equals_pattern 定义\\s\\w\\s等于脚本骨架分四块读取、并发检查、聚合、报告。核心用asyncio.gather把每行样本的检查任务并发出去。import asyncio import json import re import tomllib from collections import Counter, defaultdict from pathlib import Path import httpx with open(config.toml, rb) as f: CFG tomllib.load(f) DATASET_PATH Path(CFG[dataset][path]) REQUIRED CFG[dataset][required_fields] MAX_WORKERS CFG[concurrency][max_workers] BATCH_SIZE CFG[concurrency][batch_size] sem asyncio.Semaphore(MAX_WORKERS) def load_jsonl(path: Path): rows [] with path.open(r, encodingutf-8) as f: for lineno, line in enumerate(f, 1): line line.strip() if not line: continue try: obj json.loads(line) obj[_lineno] lineno rows.append(obj) except json.JSONDecodeError as e: rows.append({_lineno: lineno, _parse_error: str(e)}) return rows async def check_fields(row): issues [] if _parse_error in row: return {lineno: row[_lineno], type: parse_error, detail: row[_parse_error]} for field in REQUIRED: if field not in row: issues.append({lineno: row[_lineno], type: missing_field, detail: field}) elif not str(row[field]).strip(): issues.append({lineno: row[_lineno], type: empty_field, detail: field}) return issues async def check_quality(row): issues [] output str(row.get(output, )) input_text str(row.get(input, )) # 字符串字面量误声明 for m in re.finditer(CFG[quality_rules][literal_var_pattern], output): issues.append({ lineno: row[_lineno], type: literal_as_var, detail: m.group(0), }) # // 翻译检查 if // in input_text and 整除 in output: issues.append({ lineno: row[_lineno], type: floor_div_translation, detail: input 含 // 但 output 用了整除, }) # 定义 X 等于 误用 if re.search(CFG[quality_rules][define_equals_pattern], output): issues.append({ lineno: row[_lineno], type: define_equals_misuse, detail: 非字符串赋值用了定义X等于, }) return issues async def check_prompt_coverage(row, rules): hits [] output str(row.get(output, )) for rule_name, pattern in rules.items(): if re.search(pattern, output): hits.append(rule_name) return hits async def process_row(row, rules): async with sem: field_issues await check_fields(row) quality_issues await check_quality(row) coverage_hits await check_prompt_coverage(row, rules) return { lineno: row[_lineno], field_issues: field_issues, quality_issues: quality_issues, coverage_hits: coverage_hits, input: str(row.get(input, )), } async def main(): rows load_jsonl(DATASET_PATH) print(f总行数: {len(rows)}) # SYSTEM_PROMPT 规则抽成检查项 prompt_rules { for_else: rfor\s.:\s*\n\s*.\n\s*else\s*:, assert: r\b断言\b, bitwise: r位运算|按位, async_await: r异步|await|asyncio, yield_from: ryield\sfrom, property_setter: r\w\.setter, walrus: r:, } tasks [process_row(row, prompt_rules) for row in rows] results await asyncio.gather(*tasks) # 聚合 field_counter Counter() quality_counter Counter() coverage_counter Counter() for r in results: for issue in r[field_issues]: field_counter[issue[type]] 1 for issue in r[quality_issues]: quality_counter[issue[type]] 1 for hit in r[coverage_hits]: coverage_counter[hit] 1 print(\n 字段问题 ) for k, v in field_counter.items(): print(f {k}: {v}) print(\n 质量问题 ) for k, v in quality_counter.items(): print(f {k}: {v}) print(\n SYSTEM_PROMPT 覆盖 ) for rule in prompt_rules: print(f {rule}: {coverage_counter.get(rule, 0)}) # 去重 input_map defaultdict(list) for r in results: input_map[r[input]].append(r[lineno]) dupes {k: v for k, v in input_map.items() if len(v) 1} print(f\n 重复 input ) print(f 重复组数: {len(dupes)}) for k, v in list(dupes.items())[:5]: print(f 行 {v}: {k[:60]}...) if __name__ __main__: asyncio.run(main())这个骨架跑起来之后你会看到类似这样的输出总行数: 1136 字段问题 missing_field: 0 empty_field: 2 质量问题 literal_as_var: 60 floor_div_translation: 11 define_equals_misuse: 6 SYSTEM_PROMPT 覆盖 for_else: 0 assert: 1 bitwise: 1 async_await: 2 yield_from: 3 property_setter: 1 walrus: 4 重复 input 重复组数: 8数字和之前人工审的结果对得上60 行字符串误声明、11 行整除、6 行定义误用for_else是 0。这就是你要补的清单。4. 实际跑一次验证请求确认脚本和 API 都通脚本骨架跑通之后下一步是验证 TaoToken 的模型判断环节能不能正常工作。写一个最小验证脚本只发一条请求确认 Base URL、Key、Model ID 三件套都对。import asyncio import httpx BASE_URL https://taotoken.net/api API_KEY sk-你的key MODEL_ID 你的模型ID async def verify(): headers { Authorization: fBearer {API_KEY}, Content-Type: application/json, } payload { model: MODEL_ID, messages: [ {role: system, content: 你是数据校验助手只回答是或否。}, {role: user, content: 下面这条 output 是否把字符串内容误声明成了变量\n设 Hello 为 空\n设 name 为 \n打印(f\Hello, {name}!\)}, ], max_tokens: 16, } async with httpx.AsyncClient(timeout30) as client: resp await client.post( f{BASE_URL}/v1/chat/completions, headersheaders, jsonpayload, ) print(状态码:, resp.status_code) data resp.json() print(返回:, data[choices][0][message][content]) asyncio.run(verify())跑通的话你会看到状态码 200返回「是」。如果返回 401说明 Key 不对如果返回 404说明 Model ID 不对如果连接超时检查 Base URL 是不是写成了https://taotoken.net/api不要带多余路径。验证通过之后把check_quality里加一个可选的模型判断分支对规则匹配不确定的行发请求确认。注意控制并发数max_workers设 16 就够了别把额度打满。5. 本篇常见报错排查401 UnauthorizedKey 没填对或者 Key 前面多了空格。检查config.toml里api_key的值确认没有引号包裹问题。如果 Key 是从控制台复制的注意别把换行符带进去。local proxy failed / connection refused本地网络环境问题不是 TaoToken 侧的问题。检查你的httpx是不是走了系统代理可以在AsyncClient里加trust_envFalse排除环境变量干扰。reading choices 报错 / KeyError: choices说明返回体不是标准 chat completions 格式。先打印resp.text看原始返回通常是 Model ID 写错了或者请求路径少了/v1。OAuth 相关报错如果你用的是 Claude Code 或 Codex 这类工具接入OAuth 流程和 API Key 是两套。校验脚本里统一用 API Key不要混用。Claude Code 接入文档在这里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdocasyncio 报 RuntimeError: Event loop is closed在 Windows 上跑 asyncio 常见。把asyncio.run(main())换成显式创建 loop或者升级 Python 到 3.10。JSONL 解析报 json.decoder.JSONDecodeError某一行不是合法 JSON。脚本里已经捕获了会记到_parse_error。跑完之后看哪些行报错单独修。去重结果里重复组数偏高说明input字段里有大量模板化内容。这时候不要直接删先看重复的output是不是也相同。如果 output 不同说明是同一 input 的多参考答案可以保留如果 output 也相同才是真重复。6. 补样本之前先把校验脚本接进你的工作流校验脚本跑完只是第一步。真正省时间的是把它接进你的数据生产流程每次 Trae 生成新版本数据集先跑一遍校验把报告贴给 Trae让它按报告修。这样你不需要手动逐行看Trae 也不需要重新理解整个数据集。具体做法把脚本输出存成report.json然后写一个简短的 prompt 模板把 report 里的问题清单喂给 Trae。比如下面是 sft_dataset_v3.jsonl 的校验报告请按问题类型逐条修复 - literal_as_var: 60 行模式为「设 X 为 空」但 X 是字符串内容 - floor_div_translation: 11 行input 含 // 但 output 用了整除 - define_equals_misuse: 6 行非字符串赋值用了定义X等于 - for_else: 0 条需要新增 3-5 条样本 ...这样 Trae 拿到的是结构化的问题清单修复准确率比让它自己审高很多。补样本的时候也按这个清单来for_else补 3-5 条、assert补 3-4 条、bitwise补 3-4 条、async_await补 3-4 条、yield_from补 2-3 条、property_setter补 3-4 条、独立变量赋值加 f-string 组合补 10-15 条。补完之后再跑一遍校验脚本确认for_else不再是 0质量问题计数归零。这时候数据集才算可以送进训练管线。如果你想让 Trae 直接帮你写修复脚本批量处理那 77 行把校验报告和原始 JSONL 一起给它让它按行号定位、按模式替换。注意让它保留原始行号映射方便你回查。最后提醒一点校验脚本里的正则规则要跟着 SYSTEM_PROMPT 更新。SYSTEM_PROMPT 加了新语法点prompt_rules里就要加对应的检查项否则覆盖检查会漏。这个维护成本不高但漏了就会让新语法点的样本缺口悄悄溜过去。