在不少公司的研发组织里AI 数据建设已经从临时支持逐渐变成独立团队。围绕这个团队经常出现一个问题数据工作是不是应该直接交给数据科学家来做现实中即使 AI 数据部门的组织层级提高了基础数据的采集、清洗、标注、格式转换、质量检查和版本管理往往仍然由数据工程师接手而不是科学家直接维护。这个现象不是谁更聪明、谁更重要的问题而是基础数据工作越来越表现出工业流水线的特征。下面从工程分工的角度解释背后的原因并给出一条从关系数据库到模型训练集的最小实现路径让读者可以自己跑通一遍理解数据团队为什么需要独立的数据生产线。1. AI 数据工作为什么需要工程化而不是直接交给科学家1.1 科学家解决的问题是模型实验数据部门解决的问题是流水线数据科学家或算法工程师的核心工作对象是模型设计网络结构、选择损失函数、调整超参数、分析训练曲线、给出评测结论。模型实验天然是迭代式、短周期、探索性的一次实验跑完结论出来接下来可能换一个思路再跑一次。数据部门要解决的是另一类问题某一类数据如何稳定、持续、按统一格式进入训练和评测流程。这包含数据库表里的业务数据如何取出文本如何清洗敏感字段如何脱敏指令样本如何构成切分比例如何固定版本如何记录。这些问题一旦写进实验里就会把模型实验变成手工搬数据的过程效率很低也很难复现。所以“数据工作没有交给科学家”并不是因为科学家不擅长处理数据而是因为模型实验要求快速改变假设数据生产却要求稳定、可重复、可监控。两者节奏不同放在同一个角色里短期能跑通长期会互相牵制。1.2 一条完整数据链路上哪些环节需要工程能力以最常见的文本指令数据为例从原始业务库里的一份工单变成大模型可以读的 JSONL 文件通常要经过下面这些环节数据抽取连接数据库按时间、状态、业务类型等条件取数。数据清洗去掉空白字符、统一编码、过滤无效内容、处理缺失值。数据去重按业务主键或文本相似度去掉重复样本。敏感信息处理手机号、身份证、地址等字段需要在生成训练集前脱敏。格式转换把业务字段构造成 instruction、input、output 结构的 JSON 行。质量检查统计空值、重复率、长度分布、类别分布。数据切分按固定随机种子切出训练集、验证集、测试集。版本管理记录数据集生成时间、数据源范围、清洗规则版本。这些环节没有一个依赖模型结构但每一个都会直接影响模型效果。数据重复会导致训练集和测试集泄漏清洗规则不一致会导致线上效果波动敏感字段没有处理会导致合规风险。把这些问题交给一个每次都手工处理的科学家技术上行不通流程上也不可控。1.3 数据工程师和数据科学家的职责边界一个比较合理的分工是数据工程师负责数据和特征的“可生产性”科学家负责数据和评测的“定义权”。工程师保证数据能按时、按量、按规则产出科学家定义数据应该包含什么能力、评测集应该覆盖什么场景、哪些坏样本必须剔除。环节数据工程师数据科学家/算法工程师数据源接入负责建表、取数、增量同步提出业务范围和数据范围清洗规则实现规则并验证效果定义什么算干净数据标注规范搭建标注流程和质检工具定义标注标准和处理原则样本构造实现 instruction 构造和格式转换设计任务模板和评估维度切分策略固化随机种子和切分代码确认测试集隔离原则数据版本管理版本和血缘关系记录实验评测使用的是哪版数据这个边界不是绝对的团队小的时候一个人可以身兼数职。但一旦数据量涨到需要脚本、调度、监控和回滚工程化就是必然选择。2. 最小可运行目标从 MySQL 业务表生成大模型可读 JSONL2.1 先确定本轮数据加工的范围和验收标准数据工程最容易犯的错是还没明确验收标准就开始写脚本。下面用一个最小案例跑通全流程范围限定在三件事从 MySQL 表中读取符合条件的客服工单数据。对文本做基础清洗去掉重复工单和空内容。生成一个可供微调或评测框架读取的 JSONL 文件。验收标准可以定为输出文件中每一行都是一个合法 JSON 对象。字段包含 instruction、input、output以及 meta 元信息。不再出现原始库中的空值工单和重复工单。中文内容在文件中直接可读不出现 \uXXXX 转义。数据集使用固定随机种子切分为训练集、验证集、测试集。这个案例虽然简单但已经包含了数据抽取、清洗、转换、质检、切分五个关键动作。理解了这一条链路后续换成用户行为数据、商品评论数据或图像标注数据思路是一致的。2.2 环境准备与依赖安装本机需要安装 Python 3.9 及以上版本MySQL 数据库可以使用本地服务或 Docker 临时启动。Python 侧只需要两个核心依赖pip install pymysql pandas scikit-learnpymysql 负责连接 MySQLpandas 负责批量读取和统计scikit-learn 用来做数据切分。如果原始库数据量非常大后续可以换成 SQLAlchemy 加分批读取的方式但当前最小案例用 pandas 足够。数据库连接这一段要注意连接字符串里的 charset 必须显式设置为 utf8mb4否则后续读取中文数据很容易出现乱码。下面是一个统一的连接参数示例conn pymysql.connect( host127.0.0.1, port3306, userroot, passwordyour_password, databaseai_data_demo, charsetutf8mb4 )2.3 用客服工单表作为示例数据源案例背景设定为一个客服工单系统。每一条记录代表一个用户提交的工单包含工单号、用户姓名、手机号、分类、标题、内容、处理状态、处理人和创建时间。先创建数据库和表CREATE DATABASE IF NOT EXISTS ai_data_demo DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE ai_data_demo; CREATE TABLE customer_ticket ( id BIGINT PRIMARY KEY AUTO_INCREMENT, ticket_no VARCHAR(32) NOT NULL, customer_name VARCHAR(64), phone VARCHAR(20), category VARCHAR(32), title VARCHAR(255), content TEXT, status VARCHAR(16), handler VARCHAR(64), created_at DATETIME, updated_at DATETIME );插入几条示例数据INSERT INTO customer_ticket (ticket_no, customer_name, phone, category, title, content, status, handler, created_at, updated_at) VALUES (TK20250101001, 张三, 13800000001, 账单, 重复扣费问题, 我的余额在昨天被扣了两次请帮忙查一下原因。, CLOSED, 李客服, 2025-01-01 10:00:00, 2025-01-01 11:00:00), (TK20250101002, 李四, 13900000002, 登录, 验证码收不到, 登录的时候一直收不到短信验证码换了手机也不行。, CLOSED, 王客服, 2025-01-01 10:30:00, 2025-01-01 12:00:00), (TK20250101003, 张三, 13800000001, 退款, 退款进度咨询, 申请退款已经三天了什么时候能到账, OPEN, 赵客服, 2025-01-02 09:00:00, 2025-01-02 09:30:00);这里故意保留了 OPEN 状态的工单和内容字段方便后续演示过滤逻辑。真实项目里由于同一用户可能有多个工单phone 字段必须脱敏不能原样写进训练集。3. 从关系数据库到模型训练集的四个加工步骤3.1 数据抽取先决定全量还是增量数据抽取的第一步是明确数据范围。模型训练通常只关心已经完结的工单因为未完结工单没有最终处理结果无法构造稳定的回答文本。所以抽取条件可以写成SELECT id, ticket_no, category, title, content, status, created_at FROM customer_ticket WHERE status CLOSED AND content IS NOT NULL AND content ! ;如果是一次性构建数据集全量抽取没有问题。如果数据每天新增就必须使用增量抽取否则同一个工单会被反复读入。常见做法是记录上一次任务执行到的最大 id 或最大 updated_at下次任务从该位置继续取数。增量场景下重复数据不能只靠抽选取数阶段解决还要在清洗阶段保留按 ticket_no 去重的兜底逻辑。这样即使上游同步任务出现重复推送数据生产线也不会把重复样本送进训练集。3.2 数据清洗去重、归一化和敏感信息处理清洗不是把数据变“漂亮”而是让数据满足模型训练的最低要求。以下是客服工单数据必须处理的四类问题。第一是空白字符归一化。用户输入里可能包含连续空格、制表符、换行、全角空格统一替换成单空格避免同一个语义的句子因为空白字符不同被当成不同样本。第二是缺失值处理。content 为空时这条样本没有训练价值直接过滤掉。如果 title 缺失可以用 category 和 content 拼接出一个占位标题但要在质量检查阶段统计缺失比例。第三是重复数据识别。最可靠的是按业务主键 ticket_no 去重。如果业务主键缺失可以退一步使用内容去重但内容完全相同的用户反馈并不一定是重复工单需要结合时间窗口判断。第四是敏感信息脱敏。原始表里的 phone、customer_name 不能直接写进 JSONL。施工单内容中的手机号、地址也要用正则替换成占位符。下面是一个简单脱敏函数import re def desensitize(text): if not text: return text text re.sub(r1[3-9]\d{9}, [PHONE], text) text re.sub(r\d{17}[\dXx], [ID_CARD], text) return text脱敏逻辑在样本构造之前执行确保输出文件里不会残留线上用户信息。3.3 指令构造和 JSONL 格式转换清洗完成的数据还是关系表结构大模型训练框架通常期望的是每行一个 JSON 对象。指令微调场景里最常见的结构是{ instruction: 请根据客服工单内容输出一段简洁的工单处理摘要。, input: 工单内容我的余额在昨天被扣了两次请帮忙查一下原因。, output: 当前工单已处理处理分类为账单。工单标题重复扣费问题。, meta: { ticket_no: TK20250101001, category: 账单, status: CLOSED } }instruction 描述任务input 存放模型需要阅读的输入内容output 存放期望答案meta 里保存业务标识。不要把业务字段直接平铺进训练文本否则后续做数据回滚和问题定位时会非常困难。JSONL 与 JSON 的区别是每一行都是一个独立的 JSON 对象而不是一个包含所有对象的数组。这种格式适合逐行读取不需要一次性载入全部数据大文件场景下也更友好。3.4 质量统计和异常样本识别转换完成不代表数据合格还要跑一次质量统计。最少需要看六个指标总样本数。清洗后剩余样本数。空内容行数。重复工单数。过短文本数量。各分类样本分布。统计的目的不是追求某类样本越多越好而是发现问题。例如清洗后样本数骤降很可能是过滤条件写得太严格某个分类占比超过 90%说明原始数据分布已经严重倾斜模型很可能学到偏置。4. 用 Python 实现一条最小数据加工流水线4.1 项目目录和核心依赖建议按下面的目录组织代码把“取数”“清洗”“转换”“质检”拆成不同函数避免把全部逻辑写在一个脚本里ai_data_pipeline/ ├── requirements.txt ├── extract.py ├── clean.py ├── convert.py ├── quality_check.py └── run_pipeline.py对应依赖文件pymysql1.1.1 pandas2.2.2 scikit-learn1.5.0版本号表示本文示例使用的版本实际项目中以你自己环境可用的版本为准。拆分文件的意义是取数逻辑变了只改 extract清洗规则变了只改 clean不会动了清洗又带出取数问题。4.2 读取数据库并生成清洗后的数据集先定义一个取数函数返回 pandas DataFrame。连接信息不建议硬编码可以先放在脚本顶部后续接入生产环境时改用配置中心或环境变量。import pymysql import pandas as pd def load_closed_tickets(host, port, user, password, database): conn pymysql.connect( hosthost, portport, useruser, passwordpassword, databasedatabase, charsetutf8mb4 ) sql SELECT id, ticket_no, category, title, content, status, created_at FROM customer_ticket WHERE status CLOSED AND content IS NOT NULL AND content ! df pd.read_sql(sql, conn) conn.close() return df读取成功后进入清洗阶段。清洗函数把文本归一化、脱敏、去重三个动作串起来import re def normalize_whitespace(text): if text is None: return return re.sub(r\s, , str(text)).strip() def desensitize(text): if not text: return text text re.sub(r1[3-9]\d{9}, [PHONE], text) text re.sub(r\d{17}[\dXx], [ID_CARD], text) return text def clean_df(df): df[clean_content] df[content].apply(normalize_whitespace) df[clean_title] df[title].apply(normalize_whitespace) df[clean_content] df[clean_content].apply(desensitize) df[clean_title] df[clean_title].apply(desensitize) df df[df[clean_content] ! ] df df.drop_duplicates(subset[ticket_no]) return df注意一个坑先过滤再脱敏会比先脱敏再过滤多处理一些无效文本。在数据量大的场景推荐先做空值过滤再做脱敏减少正则替换次数。4.3 构造指令样本并输出 JSONL清洗完成后把行数据转换成指令样本。下面这个函数把客服工单的标题和内容组装成一条简单的摘要任务样本import json def build_sample(row): instruction 请根据客服工单内容输出一段简洁的工单处理摘要。 input_text f工单标题{row[clean_title]}\n工单内容{row[clean_content]} output_text f处理分类{row[category]}。工单标题{row[clean_title]}。 meta { ticket_no: row[ticket_no], category: row[category], status: row[status], created_at: str(row[created_at]) } return { instruction: instruction, input: input_text, output: output_text, meta: meta } def write_jsonl(df, output_path): with open(output_path, w, encodingutf-8) as f: for _, row in df.iterrows(): sample build_sample(row) f.write(json.dumps(sample, ensure_asciiFalse) \n)这里把输出文本模板写得比较简单目的是演示链路。真实项目中 output 应来自人工标注、客服处理结果或线上业务结论不能凭空拼一段内容当成正确答案。生成 JSONL 时json.dumps 必须显式传入 ensure_asciiFalse否则中文字符会全部变成 \uXXXX文件体积变大且人工排查困难。4.4 运行流程与预期结果在 run_pipeline.py 里组合各步骤def run(): df load_closed_tickets( host127.0.0.1, port3306, userroot, passwordyour_password, databaseai_data_demo ) df clean_df(df) print(清洗后样本数:, len(df)) write_jsonl(df, ticket_samples.jsonl) if __name__ __main__: run()正常运行会在控制台输出一行样本数。打开生成的 ticket_samples.jsonl每行应是一个中文可读的 JSON 对象。如果文件中出现 \uXXXX 转义说明 ensure_ascii 没有设置为 False。4.5 训练集、验证集、测试集切分数据切分要遵循一个原则测试集必须尽可能接近真实线上场景并且不能反向渗透到训练集。最简单可靠的做法是固定随机种子按比例切分from sklearn.model_selection import train_test_split df_train, df_temp train_test_split(df, test_size0.2, random_state42) df_val, df_test train_test_split(df_temp, test_size0.5, random_state42) print(train:, len(df_train)) print(val:, len(df_val)) print(test:, len(df_test))这里 test_size0.2 表示先从总量中抽出 20% 作为评估预留数据然后把这 20% 再等分为验证集和测试集最终比例是训练 80%、验证 10%、测试 10%。random_state 固定为 42保证每次运行切分结果一致这是实验可复现的前提。5. 把数据集交给模型前先完成这几项质量检查5.1 一份可以直接复用的质量检查清单构造完数据集后不要急着启动训练。建议先执行一份固定清单任何一项不达标都要停下来查原因。检查项合理范围不达标的处理方式清洗前后样本量不应出现数量级跳变检查过滤条件和清洗逻辑重复工单数0确认 ticket_no 去重是否生效空内容样本0检查取数 SQL 是否过滤空串文本长度分布无明显长尾统计 min、max、p50、p95分类分布不出现单一分类超过 90%评估数据来源是否偏差敏感信息残留无手机号、身份证号用正则扫描全文件JSONL 单行格式每行都是合法 JSON用 json.loads 逐行解析这份清单可以用脚本自动执行也可以作为人工评审的首页清单。数据版本越频繁自动化的收益越大。5.2 用规则快速识别脏数据质量检查不一定要依赖大模型。规则扫描在早期阶段更高效。一个通用方案是扫描 JSONL 文件统计常见问题import json def scan_jsonl(path): total 0 invalid 0 lengths [] with open(path, r, encodingutf-8) as f: for line in f: line line.strip() if not line: continue total 1 try: obj json.loads(line) lengths.append(len(obj.get(output, ))) except json.JSONDecodeError: invalid 1 print(总行数:, total) print(非法JSON行数:, invalid) print(output长度中位数:, sorted(lengths)[len(lengths) // 2] if lengths else 0)非法 JSON 行数只要不是 0就说明格式转换代码有 bug。output 长度中位数过低说明输出模板可能太短需要补充更完整的人工标注数据。5.3 数据版本、血缘和回滚管理数据文件一旦生成就具有了不可变属性。训练实验需要能精确回答一个问题当前模型用的训练集是哪一版。因此每条数据产线都应该记录元信息。推荐在每个 JSONL 文件同级目录生成一份 manifest.json{ dataset_name: customer_ticket_summary_v1, created_at: 2025-01-10 12:00:00, sql_filter: status CLOSED, clean_rule_version: v1.2, sample_count: 1000, train_count: 800, val_count: 100, test_count: 100, raw_source: customer_ticket, checksum: 生成的哈希值 }有了这个文件实验报错时就能快速定位是哪次清洗规则改变导致数据分布异常发布新版本数据后如果模型效果下降也能直接回滚到上一版数据文件而不是把所有代码推到重来。6. 常见问题与排查路径6.1 读取数据库出现中文乱码现象是 pandas 读出来的中文全是问号或乱码。常见原因有三个数据库连接字符集没有设置成 utf8mb4数据表本身字符集不是 utf8mb4终端或脚本输出环境不支持中文显示。排查顺序是先看数据库连接参数确认 charsetutf8mb4再查表字符集执行SHOW CREATE TABLE customer_ticket;最后用最简单的 Python print 输出一条 content 字段确认是读取乱码还是显示乱码。修复方式是把连接字符串和表字符集都统一到 utf8mb4。6.2 清洗后样本量骤减现象是 MySQL 里查出来有 10 万条经过清洗脚本后只剩几百条。重点检查三个条件取数 SQL 里的 status 过滤是否过于严格比如把未完结工单也排除掉了clean_text 后空字符串判断是否把包含标点但内容很短的文本也删掉了drop_duplicates 的 subset 是否选错了列导致大量不同工单因为 customer_name 相同而合并。建议把每一步过滤的样本量都打印出来不要只打印最终数量。通过中间数字可以快速定位是抽取问题、清洗问题还是去重问题。6.3 生成的 JSONL 无法被训练框架读取现象是训练脚本读取文件时报 JSON 解析错误或者读出来的字段全是空对象。原因通常是三种某行 JSON 出现了尾逗号说明代码里用了手工拼接而不是 json.dumps文件编码不是 UTF-8某一行数据包含特殊控制字符json.loads 无法解析。排查命令可以用 Python 逐行读取定位到具体行号python -c import json; [json.loads(l) for l in open(ticket_samples.jsonl, encodingutf-8) if l.strip()]如果报错定位到某一行直接打印该行前 200 个字符重点看是否有不可见控制字符。解决方案是统一使用 json.dumps 生成不要手工拼接字符串。遇到不可见字符在清洗阶段用正则过滤掉控制字符。6.4 增量更新后数据重复现象是第二次运行任务后输出文件中的样本条数比预期多训练时发现同一工单多次出现。原因是增量任务没有记录上次处理的游标每次都是全量重跑或者全量重跑后没有执行按 ticket_no 去重。正确做法是把数据表的自增主键 id 或 updated_at 作为增量游标每次任务启动时先查询上次的最大游标值取数 SQL 加上AND id last_max_id。同时保留清洗阶段的幂等逻辑即使游标记录丢失重新全量跑一遍最终输出的数据也不应该产生重复样本。7. 数据团队落地的分工建议与扩展方向7.1 工程团队和算法团队各守什么边界回到开头的问题AI 数据部门升级后为什么数据工作没有交给科学家从团队运作角度答案很清晰。数据工程师应守住数据生产的稳定性。他们要保证每天定时取数、清洗、转换、质检、发布数据版本出现失败能快速告警数据质量变化能追溯到具体规则变更。科学家应守住数据规格的定义权。他们要回答什么数据值得采集、任务模板如何设计、评估集需要覆盖哪些边界情况、什么样的样本算坏样本。科学家不需要写调度脚本但必须能说清楚评测结果和训练数据之间的关系。如果这两类工作混在一个角色身上最常见的后果是模型实验结果好但难以复现数据文件版本混乱换个人接手就不知道数据是怎么来的。公司越大、数据越复杂这个问题越明显。7.2 生产环境下的数据流水线还要补齐什么最小案例跑通后如果要进入生产环境至少还需要补齐六项能力。任务调度建议使用 Airflow、DolphinScheduler 或内部调度平台把取数、清洗、质检、发布拆成可重跑的定时任务。数据监控对输出文件的行数、重复率、空值率、分类分布设置阈值超过阈值自动告警。元数据中心记录每份数据集的来源、生成时间、SQL、清洗版本和校验和。权限管理原始库的敏感字段和脱敏后的训练集要区分权限不能所有研发成员都能直接访问原始手机号。回滚机制保留历史版本数据文件使用对象存储或 NAS 路径保存并通过 manifest 文件检索。质量门禁生成训练集后自动执行质量扫描质量指标不达标时阻止发布避免脏数据流入训练任务。7.3 如何从文本数据扩展到图像检测数据数据处理流水线的思想同样适用于图像任务。YOLO 这类目标检测模型需要的数据集也可以从业务系统的图片表和标注表抽取原始图片读取出来按统一目录存放标注坐标从数据库或标注平台导出然后转换成 YOLO 格式的 txt 文件并同步生成 train.txt、val.txt、test.txt 三个路径清单。与文本数据加工类似图像数据也需要处理重复图片、错误标注、类别不平衡、文件名冲突等问题。核心还是那条思路工程团队保证标注数据稳定流转算法团队定义类别和质检标准。7.4 给新手的练习建议如果刚开始接触 AI 数据工程不建议直接搭一套大数据平台。可以先按本文的步骤手工造 50 条客服数据完整走通 MySQL 到 JSONL 的流程。然后自己增加两个难度给数据加一个 deduplicate 字段做增量抽取把清洗规则写进版本控制尝试回滚到旧版本。接下来可以尝试把同样的数据加工成两种不同任务格式对比模型在两种数据上的表现差异。这个过程会比直接跑一个大模型更有价值因为它能训练一个最基本的工程直觉数据文件的可复现性远比单次实验的惊艳结果更重要。