Python 实战：用一套“可校验 + 可回流”的流水线生成高质量 JSONL（适用于 SFT 数据工程）

做大模型 SFT，很多人卡在两件事：

1）生成的数据量不小，但质量参差不齐

2）训练完发现模型不听话，却说不清是哪类样本有问题

解决办法不是“再生成一堆数据”，而是用 Python 把数据工程做成流水线：

• 生成（build）
• 校验（validate）
• 修复/剔除（fix/filter）
• 统计（metrics）
• 回流（bad rows → 下一轮增量训练）

本文给你一份通用的工程设计思路（偏落地），你可以直接套在“画像→课程推荐→学习规划”的任务上。

1. 项目目录：先把工程结构搭清楚

推荐一个轻量但可扩展的结构：

sft_data_pipeline/
  data/
    profiles.jsonl
    catalog.json
    out_train.jsonl
    out_test.jsonl
    bad_rows.jsonl
    reports/
      metrics.json
      samples.md
  src/
    build_candidates.py
    synth_answer.py
    validate_and_filter.py
    split_train_test.py
    utils_schema.py
    utils_metrics.py
  run.sh

这样你就能把“生成”和“校验”分离：生成可以快，校验可以严。

2. Python 里最关键的三件事：Schema、验证器、错误分桶

2.1 用“固定 Schema”约束输出结构

SFT 最容易出的问题是：输出 JSON 不可解析、字段多/少、类型错误。

在 Python 侧，建议把“输出允许字段”写死，做硬校验。

你不一定非要上复杂库，用纯 Python 也能做 80% 的事：

• 顶层必须包含：selected
• selected 必须是 list
• 每个 item 必须包含 goods_id
• goods_id 必须在 candidates 的 goods_id 集合中
• 禁止出现未定义字段（防止模型乱加）

2.2 验证器：宁可严一点，不要“将就能用”

在工程里，把“过不了验证”的样本统一写入 bad_rows.jsonl，同时记录 bad_reason，比如：

• json_parse_fail
• out_of_candidates
• missing_field
• extra_field
• empty_selected_when_should_pick
• hallucinated_course_info

2.3 错误分桶：为闭环服务

错误分桶不是为了好看，是为了告诉你下一轮该补什么数据：

• 越界多 → 强化“只能从 candidates 选”的样本
• 结构错多 → 增加严格 JSON 的示例与反例
• 理由空泛 → 强制引用 brief/teacher/direction

3. 一个通用的“校验 + 回流”脚本思路

你可以把每条样本看成：

• input：{user_profile, user_query, candidates, constraints}
• output：模型生成的 JSON 字符串（或 dict）

校验器做的事就是三步：

1）json.loads(output) 能不能成功

2）结构是否符合 schema

3）内容是否符合业务约束（比如 goods_id 必须在 candidates）

然后把结果写入两份文件：

• out_train.jsonl：合格样本
• bad_rows.jsonl：不合格样本（带原因）

这种“先生成、后过滤”的方式，扩到 10k/100k 很好用，因为你不需要生成时就完美。

4. 统计指标：用 Python 给数据集做体检

建议你至少输出这几类指标到 metrics.json：

• total_rows
• valid_rows
• bad_rows
• bad_breakdown（每种 bad_reason 的数量与占比）
• json_parse_success_rate
• out_of_candidates_rate
• avg_selected_len
• empty_selected_rate

同时再抽样输出一些可读样本到 samples.md（比如每种错误各抽 3 条），你会非常省时间。

5. 数据增强：Python 侧要做“可控扩增”，不要随机发散

你要把画像扩到 10k，Python 侧可以做“受控扩增”：

5.1 画像字段扰动（不改变人设逻辑）

• time_per_week_hours 在一个合理范围抖动（如 ±1~2 小时）
• risk_preference 在相邻档位变化（保守↔平衡↔积极）
• goal 用同义表达替换（“稳健增值/长期定投/提升交易体系”）

5.2 候选集合扰动（保持同方向）

• 同方向 candidates 随机替换 10%~30%
• 保留 top1~top2 不动，尾部课程做扰动 这样能让模型学会“在相近候选里做选择”，而不是死记答案。

5.3 负样本扩增（强烈建议）

• candidates 为空
• candidates 只有 1 门
• constraints 要求方向，但 candidates 全不满足
• user_query 与画像矛盾（比如画像保守但问高杠杆）

负样本比例做到 10%~30%，模型会更稳。

6. 并发生成：Python 里怎么把速度拉满

如果你是调用 API 生成 output（teacher model 合成），Python 并发建议：

• I/O 密集：asyncio + aiohttp 或 ThreadPoolExecutor
• 重点控制：concurrency（并发数）与 retry（指数退避）

经验值（按接口限制调）：

• 并发 16~64 通常就很快了
• 一定要做：超时、重试、失败落盘（别丢数据）

7. 最后：把流水线做成“一条命令可复现”

你可以用一个 run.sh 或 Python main.py 串起：

1）profiles 扩增 → profiles_aug.jsonl

2）构造 candidates → inputs.jsonl

3）调用模型生成 output → raw_outputs.jsonl

4）校验过滤 → out_train.jsonl + bad_rows.jsonl

5）统计报告 → metrics.json + samples.md

6）切分 train/test

这就是“可复现数据工程”，也是你博客文章里最能体现专业度的一点。

结语：Python 在 SFT 里最重要的价值是“可验证”

很多人把 Python 只当“胶水”，但在 SFT 工程里，Python 的核心价值是：
把业务规则变成程序能验证的约束，把失败样本变成可迭代的闭环。

当你做到：

• 输出可解析
• ID 不越界
• 错误可分桶
• 数据可回流 微调就会越来越像工程，而不是玄学。