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