☰
Agentic数据流水线实战:从SFT到RL的高质量训练数据清洗与合成
2026/10/1 4:46:03 网站建设 项目流程

现在做大模型训练,最卡人的往往不是模型结构,也不是训练脚本,而是数据。尤其是SFT、mid-training和RL这三类场景,数据质量直接决定模型上限。我自己这几年一直在做训练数据流水线,最近把越来越多的环节换成了agentic方式来做——让大模型以智能体形态自己规划、自己调工具、自己评审、自己迭代地处理原始语料。这套方法尤其适合手里已经有一堆原始数据、但不知道如何高效清洗和扩增的团队,也适合想用合成数据做SFT但怕被脏数据带偏的个人开发者。

这篇就把我实际跑过的agentic合成 / 清洗训练数据流程完整拆开,包含三阶段需求分析、可落地的pipeline架构、关键prompt与代码,以及踩过的坑。

1. 为什么训练数据开始需要“Agent”

1.1 传统数据流水线的三条死路

先聊点扎心的。以前做数据清洗,大家第一反应就是写脚本:正则过滤、长度过滤、去重、打标签。这套东西对结构化数据很有效,但对预训练语料和指令数据就有点力不从心了。

第一,规则写不全面。你写一个正则想过滤掉“垃圾广告样本”,结果广告换了种说法就漏过去了,或者把正常样本误杀了。第二,人工标注太贵。一条SFT样本从写指令到写答案,经验丰富的标注员平均要几分钟甚至十几分钟,几万条数据就是天文数字。第三,合成数据本身的质量失控。很多人用大模型批量生成SFT数据,一天能生成几十万条,但里面大量是复读机样本、幻觉样本、格式漂移样本。用这些数据训出来的模型,效果差还不能说原因,因为问题出在“数据自己都不干净”上。

我自己最崩溃的一次,是拿一批自动生成的指令数据做SFT,跑完一个7B模型,发现它在回答里频繁重复“作为一个AI语言模型”这句话。一查,原始合成数据里大量样本都以这句话开头,规则清洗完全没拦住。

1.2 核心范式变化:从规则到“Agent自省”

agentic方式解决的不只是“自动化”问题,而是把数据处理从写死规则的脚本,变成了“让模型自己判断怎么处理、处理到什么程度”的闭环系统。

具体说,agentic数据合成/清洗的核心是四件事:

  • 规划:主Agent拿到一批原始数据,先拆任务。是清洗?是扩增?是质检?拆成子任务。
  • 执行:调用子Agent或外部工具逐条处理。比如语义去重工具、质量打分模型、改写模型。
  • 评审:另一个Agent专门负责挑毛病,检查处理后的样本是否合格,不合格就返回重做。
  • 迭代:对不合格样本循环执行“改写-评审”,直到通过或达到最大轮数。

这套模式之所以比规则清洗强,是因为LLM能理解“语义级”的质量问题。比如“这条指令重复啰嗦”是正则判断不了的,但Agent可以。它还能组合多个工具完成复杂流程,比如先查重、再评判、再改写,一套流水线全部自动跑完,你只需要留一个抽检通道。本质上就是把数据工程师的“人工判断力”变成了可批量执行的程序。

2. 三阶段数据需求拆解:SFT、mid-training、RL

2.1 SFT数据:合成、清洗、质检一个都不能少

SFT阶段要的是“高质量的指令-响应对”,核心指标是多样性、正确性、格式一致性。

用agentic方式做SFT数据,我一般分三条线并行:

第一条是合成扩增。给定一批种子指令,主Agent按领域、难度、句式风格拆成主题列表,然后每个主题下批量生成新指令和标准答案。每生成一条,都让评审Agent检查“指令意图是否清晰、答案是否完整、是否与已有样本语义重复”。

第二条是清洗存量。你手上可能有爬虫爬来的、旧模型生成的、开源数据集里筛出来的杂七杂八数据。agentic清洗的典型动作包括:删除无实际语义的废话指令、合并语义重复样本、改写格式错乱的答案、过滤政治与低俗内容(注意:不是简单关键词过滤,而是语义级识别)。

第三条是质检兜底。我习惯在合成流水线末端放一个独立的批判Agent,它不知道前面的生成Agent用的是啥prompt,只管挑刺。只有通过批判Agent的样本才会进入最终训练集。这个“评审与生成分离”设计能明显减少一个Agent自说自话导致的系统性错误。

2.2 mid-training数据:领域续训的清洗与去重

mid-training(也有些人叫domain-adaptive continued pretraining)要解决的是模型在特定领域知识不足的问题。这时候数据以文档为主,比如医疗、法律、矿山设备运维、代码仓库等垂直语料。

这块数据最大的特点是不缺——客户常常给你几个T的PDF和网页,缺的是“干净且高信息密度”的部分。agentic清洗在这里的活儿主要是:

  • 把扫描版PDF识别出来的乱码段落判死;
  • 把重复宣传语、导航文本、版权页从网页正文里剥离;
  • 对专业术语进行实体一致性校验,比如同一个设备型号在一个文档里出现三种写法,Agent会统一成标准表述;
  • 做语义级去重。用embedding向量去重只能处理近乎重复的文本,而Agent能判断“两篇文章虽然句子不同,但讲的是同一个技术方案,保留信息密度更高那篇就行了”。

另外,mid-training阶段也可以做一点合成成分,但要比SFT保守。我的经验是只做“段落级改写”,不做“文档级编造”。比如让Agent把一段专业操作规程改写成包含更多背景解释的延伸段落,同时必须保证不改动核心数值参数。

2.3 RL数据:偏好对与难度挖掘

RL阶段(尤其RLHF/DPO类的偏好优化)需要的数据形态和SFT很不一样:你需要同一个prompt下的好答案和坏答案,也就是chosen和rejected。常见做法是让一个大模型生成多个候选答案,然后由人类或奖励模型打分。

这里agentic方式的玩法更多。首先,可以让一个强Agent把同一个问题拆成多角度答案,再让一个评判Agent按“有用性、安全性、逻辑性”维度排序,自动构造偏好对。其次,可以针对模型的薄弱点做困难样本挖掘:先跑一版弱模型,找出它答错的题目,让Agent针对这些错误生成变体题,从而补齐RL训练集中“简单样本太多、困难样本太少”的问题。

对过程奖励(process reward)类RL训练,Agent还能把解题过程拆成步骤级,标注每一步的对错和原因。这个人工做起来极贵,Agent做起来虽然也有噪声,但胜在量大且可反复跑,配合抽检就能用。

3. 实操:搭建一个可跑的agentic数据流水线

3.1 整体架构:规划、执行、评审、迭代四层

我跑过的agentic数据流水线,最朴素也最能复制的版本是四层结构:

  • 调度层:一个主Agent,负责读取原始数据清单,输出任务计划,维护待处理队列。
  • 执行层:若干子Agent,按任务类型划分。常见的有合成Agent、清洗Agent、改写Agent。它们都会被注册工具集的访问权限。
  • 工具层:模型自身能力之外的外部函数。比如语义查重接口、长度过滤器、敏感词检测、格式校验器。
  • 评审层:一个独立的评审Agent,对执行结果做终审,给出pass或reject并附理由。

每一轮处理的单位是一条样本。整条流水线跑完后,评审通过率、改写轮数、各Agent调用成本、样本分布等全部落表,方便你判断该调prompt还是该调工具逻辑。

这套架构的好处是模块替换方便。后期如果你把评审Agent从小模型换成大模型,或者给执行层加一个“查事实库”工具,都不影响其他模块。

3.2 Prompt与工具设计:让Agent有手有脚

agentic的关键是让Agent能主动决定是否调用工具。我习惯把系统prompt写成带明确流程指引的形式,而不是只丢一句“你是数据工程师”。

下面是一段我常用的清洗Agent系统prompt骨架:

你是训练数据清洗Agent。你会收到一条原始样本,需要执行以下判断: 1. 样本是否符合基本格式要求?不符合则调用 format_fixer。 2. 样本与语义指纹库是否重复?重复则调用 semantic_dedup,并标记remove。 3. 样本是否存在明显事实性风险?存在则调用 fact_checker,若无法修复则标记low_quality。 4. 以上都通过,输出final样本;否则调用rewriter改写后再次评审,最多迭代3轮。 注意:任何一步操作都需要输出理由,禁止静默修改文本内容。

Prompt里那句“禁止静默修改文本内容”非常重要。没有这句,Agent经常会把好好的原文改得面目全非,你还不知道它是怎么改的。

工具层可以简单注册成JSON schema格式,比如:

{ "name": "semantic_dedup", "description": "计算当前样本与现有库中样本的语义相似度,高于阈值则判定为重复", "parameters": { "type": "object", "properties": { "text": {"type": "string"}, "threshold": {"type": "number", "minimum": 0.8} }, "required": ["text"] } }

我的习惯是把“判断逻辑”放在prompt里,“执行逻辑”放在工具里。这样Agent负责决策,工具负责稳定输出,不容易跑偏。

3.3 核心代码与参数清单

下面给一个精简版Python代码框架,核心思路是“调用模型 -> 让模型决定是否调用工具 -> 执行工具 -> 把结果喂回给模型”。模型服务我直接用OpenAI兼容接口写法,方便你换成任意本地部署模型。

from openai import OpenAI import json client = OpenAI() def run_agentic_clean(sample, tools, system_prompt, max_rounds=3): messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": f"待处理样本:\n{sample}"}] for _ in range(max_rounds): resp = client.chat.completions.create( model="your-model-name", messages=messages, tools=tools, tool_choice="auto", temperature=0.2 ) msg = resp.choices[0].message messages.append(msg) # 保留完整上下文 # 模型没有工具调用意图,说明处理完成 if not msg.tool_calls: content = msg.content or "" return {"status": "final", "text": content, "turns": len(messages)} # 执行工具调用 for call in msg.tool_calls: fn_name = call.function.name args = json.loads(call.function.arguments) result = execute_tool(fn_name, args) # 你自己实现的工具分发函数 messages.append({ "role": "tool", "tool_call_id": call.id, "content": json.dumps(result, ensure_ascii=False) }) return {"status": "timeout", "text": None, "turns": max_rounds}

几个关键参数我直接给结论:

  • temperature设0.2左右。数据清洗场景要稳定,温度太高会引入随机改写,太低又可能死板,0.2是我试过的平衡点。
  • max_rounds设3。大多数样本一轮调用工具后就能出结果,最多两轮。允许3轮是给复杂样本留余地,超过3轮直接判timeout,避免Agent陷入循环烧钱。
  • 每次工具调用都保留完整消息记录。很多新手容易犯的错是只把工具结果拼到文本里再发一遍,这样模型会丢失tool_call_id和函数名信息,function calling逻辑就会错乱。

跑完一整批数据后,把status=timeout的样本单独存出来,我通常会对它们跑一次“人工快速评审”,通常几十到几百条,用来判断是prompt问题还是样本本身破坏性太大,然后决定是放宽阈值还是直接丢弃。

3.4 质检闭环:生成数据可用性的最后防线

pipeline跑完不代表数据能用,我最后还有一道质检闭环,分三层:

第一层是自动指标。比如样本长度分布、指令唯一率、改写前后编辑距离、token超限率。这些用脚本统计,几分钟出结果,能挡住大多数格式问题。

第二层是模型打分。用一个评审模型对最终样本做1-5分打分,看各分数段的占比。正常来说5分比例稳定在50%-70%左右;如果某天合成出来的数据全是4分以下,说明生成Agent的prompt崩了,需要回溯检查。

第三层是人工抽检+红队样本。我从每一批数据里随机抽2%,同时混入十几条我知道的“坑样本”(比如含敏感词的、内容重复的、指令含糊的)。如果人工抽检通过率低于95%,整个批次就不进训练集。这套方法成本低、可靠,还能反向检验评审Agent是不是在放水。

4. 常见问题与避坑实录

4.1 典型问题速查表

现象直接原因排查方向
Agent在同一个样本上反复调用改写工具评审标准与改写标准不一致统一两边的prompt评分维度,必要时抽回日志看每一轮变化
生成数据重复度高没有在生成前查重,或查重阈值过低提高语义去重阈值,基于embedding的相似度阈值建议设置到0.85以上
模型偶尔输出完全无关内容工具调用后上下文被截断或丢失检查消息历史是否完整,确认tool call id正确回传
清洗后样本“面目全非”系统prompt没有约束改写范围增加“禁止静默修改,只改确定有问题的部分”描述
成本超预算大量样本进入迭代循环给每个Agent设max_rounds上限;先用便宜小模型初筛,再上大模型精修
人工抽检发现评审Agent漏检评审Agent和生成Agent用同一个模型,思维盲区重叠换成不同规模/不同厂商的模型做评审,或在工具里加独立的规则校验

4.2 高发问题的排查思路

第一个高发坑是Agent自说自话。我遇到最诡异的情况是,清洗Agent把所有样本里的“但是”都改成了“不过”,理由是“表达更自然”。这种偏好很难从单条样本里发现问题,所以我才坚持在pipeline里跑数据分布统计,定期检查高频词是否出现异常波动。

第二个坑是评审Agent太宽松。传统的做法是让评审Agent输出“通过/不通过”,但模型倾向说好话,导致通过率虚高。我的修正办法是强制评审Agent先给出“至少一个可改进点”,没有可改进点时直接判不通过。虽然会牺牲一点通过率,但能逼着它真正思考。

第三个坑是RL偏好对构造失衡。自动构造chosen/rejected样本时,如果两个答案差距太大,模型很快就能学会“选字数多的”这种偷懒策略。我的处理方式是在生成候选答案时,要求每条答案token数量控制在相近区间,评审时才真正按内容质量排序,而不是按长度判断。

另外补充一点,我在SFT数据里遇到过原始语料本身含毒,Agent没识别出来的情况。后来我在工具层加了一个独立的敏感词与低质内容检测接口,所有样本先过一遍规则工具,再走Agent判断。规则工具相当于“粗筛”,Agent负责“精判”,两者结合误杀率下降明显。

最后分享一个小技巧

如果你不想一上来就把整套agentic流水线做重,可以先从“单Agent+单工具”的轻量组合起步。

比如先只做一件事:用一个Agent批量生成SFT候选样本,再调用一个“重复检测工具”过滤近重复项,最后随机抽50条人工看一眼。跑通之后再加评审Agent、再加改写工具,一步一步把闭环补全。不要一开始就追求全自动,因为agentic系统一旦跑偏,调试成本可能比人工清洗还高。先把每一步的产出和rd成本摸清楚,再逐步放开自动化范围。我个人在实际操作中最受益的一个决策,就是每次跑批前都问自己一句:如果这条样本出了问题,我的日志能不能在一分钟内定位到原因?能,就增量扩大规模;不能,就先停下修系统。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询