1. 先搞清楚 sft_dataset_v3.jsonl 到底缺什么你手上這份C:\dumatework\duan\tools\ai_copilot\sft_dataset_v3.jsonl是 Trae 幫你生成的段言翻譯 SFT 訓練集1136 行、35 個語法類別。表面看覆蓋面挺全但真正拿去訓模型之前得先回答三個問題字段有沒有缺、SYSTEM_PROMPT 里規(guī)定的語法點是不是每條都有樣本、有沒有重復樣本在稀釋梯度。這三個問題靠肉眼翻 JSONL 是不現(xiàn)實的。1136 行里混著 60 行字符串字面量被誤聲明成變量、11 行//翻譯不一致、6 行「定義 X 等于」誤用這些錯誤如果不在訓練前篩出來模型學到的就是壞習慣。所以這篇的核心不是「怎么生成數(shù)據(jù)」而是「怎么用 Python asyncio 并發(fā)把這份數(shù)據(jù)集體檢一遍」。適合誰看已經(jīng)用 Trae 或類似工具產出了 SFT 數(shù)據(jù)集、準備送進訓練管線、但不確定數(shù)據(jù)質量的人。你需要會一點 Python能跑腳本能看懂 JSONL 結構。不需要你懂分布式訓練校驗腳本本身是單機跑的。我試過把 1136 行串行讀一遍做正則匹配大概要 40 秒左右而且每加一個檢查項就線性變慢。改成 asyncio 并發(fā)之后同樣的檢查項壓到 3 秒以內后面再加規(guī)則也不心疼。下面把整套骨架拆開講你可以直接復制去改。先明確校驗目標分四類字段完整性——每行必須有instruction、input、output三個字段且都不能為空字符串。SYSTEM_PROMPT 覆蓋——把 SYSTEM_PROMPT 里聲明的語法規(guī)則抽成檢查項逐條統(tǒng)計命中樣本數(shù)找出 0 命中和低命中的。去重——input完全相同的行要合并input高度相似比如只差變量名的也要標記。質量規(guī)則——字符串字面量誤聲明、//翻譯、定義 X 等于誤用這些具體錯誤模式。這四類里前兩類是結構性的后兩類是語義性的。asyncio 適合做的是 I/O 密集和規(guī)則匹配密集的部分純 CPU 的正則其實提升有限但配合批量讀取和并發(fā)任務調度整體吞吐還是明顯好于串行。2. 用 TaoToken 補齊校驗腳本里的模型判斷環(huán)節(jié)純正則能查出「設 Hello 為 空」這種模式但查不出「這條樣本的 output 語義上是不是真的在翻譯 input」。后者需要模型判斷。我的做法是把規(guī)則匹配和模型判斷分開規(guī)則匹配用 asyncio 并發(fā)跑模型判斷走 TaoToken 的 API同樣用 asyncio 并發(fā)發(fā)請求。TaoToken 在這里的角色是提供一個統(tǒng)一的模型調用入口。你不需要在本地部署模型也不用為每個判斷任務單獨配一套 SDK。校驗腳本里需要模型介入的地方主要有兩處一是判斷某條 output 是否語義等價于 input 的翻譯去重時用二是判斷 SYSTEM_PROMPT 里某條規(guī)則在樣本里是否被正確執(zhí)行覆蓋檢查時用。接入方式很簡單拿到 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發(fā)請求就行。注意一點模型判斷是輔助不要讓它替代規(guī)則匹配。規(guī)則能查的比如字段缺失、//翻譯一律用規(guī)則模型只用來處理規(guī)則覆蓋不到的語義問題。這樣既快又省 token。如果你打算長期跑這類數(shù)據(jù)校驗建議用 Coding Plan額度更穩(wěn)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等于腳本骨架分四塊讀取、并發(fā)檢查、聚合、報告。核心用asyncio.gather把每行樣本的檢查任務并發(fā)出去。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總行數(shù): {len(rows)}) # SYSTEM_PROMPT 規(guī)則抽成檢查項 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 重復組數(shù): {len(dupes)}) for k, v in list(dupes.items())[:5]: print(f 行 {v}: {k[:60]}...) if __name__ __main__: asyncio.run(main())這個骨架跑起來之后你會看到類似這樣的輸出總行數(shù): 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 重復組數(shù): 8數(shù)字和之前人工審的結果對得上60 行字符串誤聲明、11 行整除、6 行定義誤用for_else是 0。這就是你要補的清單。4. 實際跑一次驗證請求確認腳本和 API 都通腳本骨架跑通之后下一步是驗證 TaoToken 的模型判斷環(huán)節(jié)能不能正常工作。寫一個最小驗證腳本只發(fā)一條請求確認 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: 你是數(shù)據(jù)校驗助手只回答是或否。}, {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(狀態(tài)碼:, resp.status_code) data resp.json() print(返回:, data[choices][0][message][content]) asyncio.run(verify())跑通的話你會看到狀態(tài)碼 200返回「是」。如果返回 401說明 Key 不對如果返回 404說明 Model ID 不對如果連接超時檢查 Base URL 是不是寫成了https://taotoken.net/api不要帶多余路徑。驗證通過之后把check_quality里加一個可選的模型判斷分支對規(guī)則匹配不確定的行發(fā)請求確認。注意控制并發(fā)數(shù)max_workers設 16 就夠了別把額度打滿。5. 本篇常見報錯排查401 UnauthorizedKey 沒填對或者 Key 前面多了空格。檢查config.toml里api_key的值確認沒有引號包裹問題。如果 Key 是從控制臺復制的注意別把換行符帶進去。local proxy failed / connection refused本地網(wǎng)絡環(huán)境問題不是 TaoToken 側的問題。檢查你的httpx是不是走了系統(tǒng)代理可以在AsyncClient里加trust_envFalse排除環(huán)境變量干擾。reading choices 報錯 / KeyError: choices說明返回體不是標準 chat completions 格式。先打印resp.text看原始返回通常是 Model ID 寫錯了或者請求路徑少了/v1。OAuth 相關報錯如果你用的是 Claude Code 或 Codex 這類工具接入OAuth 流程和 API Key 是兩套。校驗腳本里統(tǒng)一用 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())換成顯式創(chuàng)建 loop或者升級 Python 到 3.10。JSONL 解析報 json.decoder.JSONDecodeError某一行不是合法 JSON。腳本里已經(jīng)捕獲了會記到_parse_error。跑完之后看哪些行報錯單獨修。去重結果里重復組數(shù)偏高說明input字段里有大量模板化內容。這時候不要直接刪先看重復的output是不是也相同。如果 output 不同說明是同一 input 的多參考答案可以保留如果 output 也相同才是真重復。6. 補樣本之前先把校驗腳本接進你的工作流校驗腳本跑完只是第一步。真正省時間的是把它接進你的數(shù)據(jù)生產流程每次 Trae 生成新版本數(shù)據(jù)集先跑一遍校驗把報告貼給 Trae讓它按報告修。這樣你不需要手動逐行看Trae 也不需要重新理解整個數(shù)據(jù)集。具體做法把腳本輸出存成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質量問題計數(shù)歸零。這時候數(shù)據(jù)集才算可以送進訓練管線。如果你想讓 Trae 直接幫你寫修復腳本批量處理那 77 行把校驗報告和原始 JSONL 一起給它讓它按行號定位、按模式替換。注意讓它保留原始行號映射方便你回查。最后提醒一點校驗腳本里的正則規(guī)則要跟著 SYSTEM_PROMPT 更新。SYSTEM_PROMPT 加了新語法點prompt_rules里就要加對應的檢查項否則覆蓋檢查會漏。這個維護成本不高但漏了就會讓新語法點的樣本缺口悄悄溜過去。