拓冰网站定制 
2025年之后AI大模型领域的竞争已经从“拼模型结构”转向“拼数据质量”。字节跳动近期将AI业务的组织架构进一步收紧继Seed大模型团队、FlowAI应用团队之后又成立了一个专门面向数据的一级部门直接回应了“数据是AI核心竞争力”这一关键命题。这件事表面看是大厂组织调整背后其实是整个行业对数据处理、数据治理和数据集构建方式的重新定义。这篇文章不谈八卦只谈技术判断。我会从字节这个动作出发拆解三个问题为什么数据部门要独立成一级部门数据工程在AI链路里到底卡在哪些环节普通开发者和团队能从中借鉴哪些可落地的数据基础设施方法论如果你正在做大模型应用、RAG系统、模型微调或Agent开发这篇文章里提到的数据清洗、数据版本管理、评测集构建、数据飞轮等思路可以直接拿回自己的项目里用。1. 字节成立AI数据部门背后是对AI竞争本质的一次重新定价很多人看到“字节成立AI数据部门”的第一反应是字节又在搞组织架构调整。但从技术视角看这次调整的信息量远大于组织本身。Seed负责基础模型Flow负责AI原生应用而数据部门独立出来说明字节已经意识到数据不是模型的附属品而是和模型、应用并列的第三极。过去几年大模型的发展逻辑是“规模法则”模型越大、数据越多效果越好。但到了2024年以后大家发现一个问题公开互联网的高质量文本数据快被“挖空”了。各家训练数据集的重复率越来越高模型之间的差异越来越小。这时候谁能构建更高质量、更独特、更结构化、更贴合业务场景的数据集谁就能在模型效果上拉开差距。字节这个动作等于把“数据”从后台职能提升到了战略级部门。这背后的技术逻辑是数据已经不只是训练时的“燃料”而是贯穿模型预训练、监督微调、对齐、评测、知识更新、Agent工具调用的全链路基础设施。数据部门的独立意味着字节要用专业团队专门处理数据采集、清洗、去重、标注、配比、版本管理、质量评估这些脏活累活。对我们普通开发者的启示是如果你还在靠“爬点公开数据直接灌给模型”的方式做应用很快会被数据工程能力拉开差距。这不是说个人需要做一个庞大的数据团队而是要在自己的项目里建立“数据治理”的最小闭环。2. 大模型时代的数据不再只是“存起来”的数据先厘清一个概念。传统软件工程里数据管理关注的是结构化数据、事务一致性、查询性能技术栈是MySQL、Redis、Kafka、Hadoop这些。大模型时代的数据至少包含四种形态每一种都有完全不同的工程技术难点。第一种是预训练语料。这类数据量级最大动辄几十TB甚至PB级别来源是网页、书籍、论文、代码、音视频字幕。难点不是存而是清洗和去重。一个网页里有大量导航栏、广告、重复段落如果直接喂给模型模型学到的是噪声而不是知识。第二种是指令数据。这是微调和对齐的关键。比如“请帮我总结这篇文章”对应的期望输出是“这篇文章主要讲了……”。这类数据通常是人工标注或从用户反馈中挖掘量级比预训练语料小很多但质量要求极高。一条坏的指令数据可能让模型学会错误行为。第三种是评测数据。这是最容易被忽视的一类。没有评测集你就无法量化模型升级前后的效果变化。很多团队做模型微调时觉得loss下降了就是效果好结果一上线实测模型反而变笨了。原因就是评测集覆盖度不够或者评测集本身和训练集有重叠。第四种是业务数据/知识数据。这是企业私有数据也是RAG检索增强生成的核心。包括产品文档、客服记录、数据库内容、结构化知识图谱等。难点在于如何把非结构化数据拆成适合检索的块如何做向量化如何保持数据更新。字节独立数据部门本质上就是把这四类数据的能力统一收拢。而在我们自己的项目里也应该按这四个维度去规划数据工作而不是笼统地说“我们有很多数据”。3. 数据质量为什么成为模型能力的天花板2023年的时候业界流行一句话数据和模型是“垃圾进垃圾出”。现在这个说法已经不够准确了更准确的说法是数据质量决定了模型能力的上限模型架构决定的是逼近这个上限的效率。为什么要强调这一点因为很多团队在微调模型时花了大量时间调学习率、调LoRA rank、试各种prompt模板但最终效果不如别人一个干净的数据集。原因在于模型参数只是把数据里的规律固化下来数据里没有的规律模型再怎么调参也学不出来。举个例子。如果你做客服场景的模型微调收集了10000条客服对话。但里面的用户问题大多雷同真实场景中的长尾问题只占5%。模型训练完之后常见问题回答得很好一旦用户换个说法模型就答非所问。这不是模型不够聪明而是数据覆盖度不够。再比如做RAG系统时知识库文档里有大量重复段落、过时内容、格式混乱的表格。检索模块把相关片段捞出来后生成模块基于这些噪声内容做总结结果自然不理想。很多人以为是embedding模型选得不好实际是源数据没有做清洗和治理。字节数据部门的独立传递的一个重要信号就是以后评判一个AI团队的能力不能只看模型榜单分数还要看这个团队的数据构建、清洗、评测、迭代能力。而数据能力是可以积累的一旦形成“数据飞轮”效应后来者很难追上。4. 从组织调整看技术趋势数据工程正在成为AI的核心岗位如果你关注招聘市场会发现“数据工程师”和“AI数据工程师”的需求量在持续上升。传统的BI数据工程师偏向报表、数仓、ETL而AI方向的数据工程师需要理解模型训练逻辑、tokenizer、embedding、数据去重算法、指令数据构建、评估集设计。字节把数据部门提升为一级部门会带来一个连锁反应其他大厂也会跟进行业对数据工程岗位的定价会水涨船高。对开发者来说现在学习AI数据工程相当于2008年学习移动开发属于提前卡位。AI数据工程师需要掌握的技能包括大规模文本处理如使用Spark、Dask处理TB级数据、数据去重算法如MinHash、SimHash、语义去重基于embedding的聚类、数据标注流程设计、质量评估指标体系、数据版本管理如DVC、数据流水线编排如Airflow、Prefect、以及面向大模型的数据配比实验方法。这些技能和传统后端开发、算法工程师有交叉但侧重点不同。传统算法工程师关心模型结构AI数据工程师关心数据如何影响模型行为。这种“数据即代码”的思维方式正是字节这次调整想强调的。5. 构建你自己的AI数据基础设施从清洗到版本管理不管你是不是字节员工这套数据方法论都可以落到自己的项目里。下面我以一个典型的大模型知识库项目为例演示如何从零构建一个最小可用的AI数据流水线。5.1 数据采集与归一化假设你要做一个企业内部知识库问答机器人。数据来源可能有内部Wiki页面HTML产品文档Markdown/PDF历史客服工单Excel/CSV工单留言JSON第一步把所有格式统一转成纯文本或Markdown并保留元数据来源、更新时间、作者、权限。下面是一个用Python做最小归一化的示例# normalize_docs.py import json import hashlib from pathlib import Path from bs4 import BeautifulSoup import markdownify def html_to_markdown(html_content: str) - str: soup BeautifulSoup(html_content, html.parser) # 去掉 script 和 style 中的内容 for tag in soup([script, style, nav, footer]): tag.decompose() return markdownify.MarkdownConverter().convert_html(soup.prettify()) def process_document(raw_path: Path, output_path: Path) - None: suffix raw_path.suffix.lower() if suffix .html: text html_to_markdown(raw_path.read_text(encodingutf-8)) elif suffix .md: text raw_path.read_text(encodingutf-8) elif suffix .json: data json.loads(raw_path.read_text(encodingutf-8)) # 假设 JSON 里有一个 content 字段 text data.get(content, ) else: text raw_path.read_text(encodingutf-8) doc { source: str(raw_path), text: text, chars: len(text), sha256: hashlib.sha256(text.encode(utf-8)).hexdigest(), } output_path.write_text(json.dumps(doc, ensure_asciiFalse, indent2), encodingutf-8) if __name__ __main__: raw_dir Path(./raw_docs) out_dir Path(./normalized_docs) out_dir.mkdir(exist_okTrue) for raw_file in raw_dir.rglob(*): if raw_file.is_file() and raw_file.suffix.lower() in (.html, .md, .json, .txt): output_file out_dir / f{raw_file.stem}.json process_document(raw_file, output_file) print(归一化完成输出目录:, out_dir)这段代码做的事情很简单但解决了大问题不同来源的文档有了统一的JSON结构后续清洗、切块、向量化都用这个统一格式而不是每种数据写一套解析逻辑。5.2 清洗规则去掉噪声保留语义单元清洗不是简单地把空行删掉。对于LLM场景清洗要考虑以下几点去掉页眉页脚、导航、版权声明等重复性字符串。去除HTML标签中的隐藏内容。合并被截断的段落。去重包括精确去重和语义去重。过滤质量过低的内容如字符数过短、乱码、无标点段落。一个实用的清洗函数如下# clean_text.py import re def clean_text(text: str) - str: # 去掉不可见字符和乱码 text re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f], , text) # 把多个空行压缩为一个 text re.sub(r\n{3,}, \n\n, text) # 去掉常见的页眉页脚模式这里按需调整 text re.sub(r(?m)^\s*(版权所有|Copyright|隐私政策|上一篇|下一篇)\s*$, , text) # 去掉多余空格 text re.sub(r[ \t]{2,}, , text) # 保留中文、英文、数字、常见标点其他换成空格 text re.sub(r[^\u4e00-\u9fff\u0030-\u0039\u0041-\u005a\u0061-\u007a\u3000-\u303f\uff00-\uffef\s.,!?;:()\-], , text) # 再次压缩空白 text re.sub(r\s, , text).strip() return text if __name__ __main__: sample 这是测试。\n\n\n\n版权所有XXX\n\n应该保留的内容。 print(clean_text(sample))注意清洗规则不能一刀切。比如代码文档里可能包含-、、{}这样的符号直接正则替换会把代码语义破坏。所以针对不同来源的数据应该维护不同的清洗规则并保留清洗前版本方便回溯。5.3 数据去重MinHash与语义去重大模型训练和RAG检索对重复数据都非常敏感。训练集重复度过高会导致模型记忆严重、泛化能力下降RAG知识库里重复文档会导致检索结果冗余浪费上下文窗口。精确去重可以用哈希但很多重复是“近似重复”比如两篇文章只有几个词不同。推荐用MinHash LSH做大规模去重这里给一个简化版# minhash_dedup.py import hashlib from datasketch import MinHashLSH, MinHash def tokenize(text: str) - set: # 简单中文分词按字符 n-gram或使用 jieba import jieba return set(jieba.cut_for_search(text)) def build_lsh(docs, threshold0.8, num_perm128): lsh MinHashLSH(thresholdthreshold, num_permnum_perm) minhashes {} for idx, doc in enumerate(docs): m MinHash(num_permnum_perm) for token in tokenize(doc): m.update(token.encode(utf-8)) lsh.insert(fdoc_{idx}, m) minhashes[fdoc_{idx}] m return lsh, minhashes if __name__ __main__: docs [ 大模型训练需要高质量数据, 大模型训练需要高质量数据集, 今天的天气真不错, ] lsh, minhashes build_lsh(docs) # 查询每个文档的近似重复 for key, m in minhashes.items(): result lsh.query(m) if len(result) 1: print(f{key} 与 {result} 近似重复)对于精读项目可以进一步做embedding级别的语义去重用CLIP、bge或text-embedding模型把文本向量化然后计算相似度矩阵将相似度超过阈值的文档合并或丢弃。5.4 数据切块Chunking的策略RAG系统里切块大小直接影响检索效果。常见误区是全库统一用固定长度切分。实际上切块策略应该结合文档结构判断。比如有标题结构的文档按标题层级切分。表格数据按行/列语义合并。代码文件按函数/类切分。长对话按轮次切分。一个混合切块示例# chunker.py from typing import List, Dict def chunk_by_headings(text: str, max_chunk_size: int 1000) - List[Dict[str, str]]: lines text.split(\n) chunks [] current_chunk [] current_title section for line in lines: if line.startswith(#) and current_chunk: content \n.join(current_chunk).strip() if content: chunks.append({title: current_title, content: content}) current_title line.lstrip(#).strip() current_chunk [line] else: current_chunk.append(line) if len(\n.join(current_chunk)) max_chunk_size: content \n.join(current_chunk).strip() if content: chunks.append({title: current_title, content: content}) current_chunk [] if current_chunk: content \n.join(current_chunk).strip() if content: chunks.append({title: current_title, content: content}) return chunks if __name__ __main__: markdown_text # 第一章 什么是数据治理 数据治理是一个长期工程。 ## 1.1 元数据管理 元数据是数据的数据。 # 第二章 数据质量 数据质量包括准确性、完整性、一致性。 for chunk in chunk_by_headings(markdown_text): print(chunk[title], -, chunk[content][:30])生产环境中还需要考虑chunk之间的重叠避免切块切断了关键上下文。通常重叠1-2个句子即可。5.5 数据版本管理与血缘数据一旦进入训练流程任何修改都可能影响模型效果。如果没有版本管理你很难回答“这个模型是用哪一版数据训练的”这个问题。推荐使用DVCData Version Control对数据集进行版本管理它类似Git但针对的是大文件。基本流程# 初始化仓库和 DVC git init dvc init # 添加数据目录生成 .dvc 文件 dvc add data/raw_docs git add data/raw_docs.dvc .gitignore git commit -m add raw docs v1 # 切换分支或回滚到旧版本 git checkout commit_id -- data/raw_docs.dvc dvc checkout同时在每次数据清洗处理时建议在输出数据集中写入一行 provenance来源信息{ version: 2025.06.01, source_files: [normalized_docs/abc.json, normalized_docs/def.json], clean_pipeline: clean_text.py:v1.2, dedup: minhash_dedup.py:v0.9, created_by: data_team }这样当模型表现异常时你能快速定位是哪一次数据处理改动导致的。6. 数据飞轮让数据越来越值钱的工程机制字节独立数据部门目标绝不只是做一次性数据集而是要建立数据飞轮。数据飞轮的核心逻辑是AI系统在使用过程中会产生新的交互数据这些数据经过筛选、清洗和标注之后重新进入训练集持续提升模型能力模型能力提升后又带来更多用户产生更多数据。飞轮听起来很美好落地却很难。难点在于不能把用户的原始输入直接当训练数据需要去除个人隐私信息。需要设计采样策略只收集那些能纠正模型错误的高价值数据。需要建立人工或自动标注流程把原始交互转换成指令格式。需要控制数据腐烂即旧数据可能过时需要定期清理。一个实际的飞轮管道可能长这样用户提问 - 模型回答 - 用户反馈点赞/点踩/修改 - 筛选出低质量回答 - 重新生成正确答案 - 人工抽检 - 写入微调数据集 - 定期微调模型 - 新模型上线 - 产生新反馈。实现这个管道你需要一个日志系统记录交互一个消息队列异步处理一个标注/审核平台以及模型版本管理。对于小团队可以先用Notion或Airtable做标注用Python脚本定期从数据库导出数据并格式化。这里我给出一个从用户反馈中筛选数据的最小示例# build_finetune_data.py import json from datetime import datetime, timedelta def extract_feedback_rows(db_connection, since: datetime): query SELECT id, user_query, bot_response, user_rating, revised_answer FROM ai_interaction_logs WHERE created_at %s AND user_query IS NOT NULL AND (user_rating bad OR revised_answer IS NOT NULL) with db_connection.cursor() as cursor: cursor.execute(query, (since,)) return cursor.fetchall() def convert_to_instruction(rows): instructions [] for row in rows: question row[user_query] if row[revised_answer]: answer row[revised_answer] else: # 这里应该调用一个更强的模型或者人工重写答案不能使用原错误回答 answer # 占位后续走人工标注队列 instructions.append({ instruction: 请回答用户的问题。, input: question, output: answer, }) return instructions if __name__ __main__: # 伪代码实际连接数据库 rows [] with open(feedback_log.json, r, encodingutf-8) as f: rows json.load(f) instructions convert_to_instruction(rows) with open(finetune_data.jsonl, w, encodingutf-8) as f: for item in instructions: if item[output]: # 只保留有正确答案的样本 f.write(json.dumps(item, ensure_asciiFalse) \n)注意从用户反馈里直接拿答案是有风险的。用户修改后的回答不一定正确所以必须经过抽检。数据飞轮的关键不是自动而是“有质量门槛的自动”。7. 数据评测没有评测集一切优化都是空谈很多团队投入大量精力构建训练数据却忽略了评测数据。字节数据部门如果只是建数据和清洗数据而不定义评测标准那模型方向就难以控制。评测数据应该是独立的、静态的、与训练数据隔离的。评测集至少要覆盖四类能力领域知识问答检验模型对业务知识的掌握。指令遵循检验模型能否按复杂指令执行。格式输出检验模型能否稳定输出JSON、Markdown等格式。对抗样本检验模型对诱导、歧义、噪声的鲁棒性。一个简单但有效的做法是用Git维护评测集版本并在每次模型迭代后记录评测结果。# evaluate.py import json from openai import OpenAI # 以 OpenAI 兼容接口为例实际项目请替换成你的模型 endpoint client OpenAI(base_urlhttp://localhost:8000/v1, api_keyEMPTY) def evaluate_model(eval_path: str, model: str) - dict: with open(eval_path, r, encodingutf-8) as f: cases json.load(f) hit 0 for case in cases: prompt case[prompt] expected case[expected] response client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], temperature0, ) answer response.choices[0].message.content.strip() # 简单判断期望关键词是否出现在答案中 if expected in answer: hit 1 return {total: len(cases), hit: hit, accuracy: hit / len(cases)} if __name__ __main__: result evaluate_model(eval_sets/v1.0.json, your-model-name) print(评测结果:, result)生产级评测还需要考虑答案公平性、多选判断、模糊匹配等但核心思想一致评测集必须固定才能对比不同版本。8. 常见数据工程问题与排查思路在搭建AI数据管道时常见问题其实非常集中。下面整理成一张表方便快速检索问题现象可能原因排查方式解决方案模型微调后效果不如base训练集包含劣质数据或标签噪声抽样查看训练集计算每条数据与标签的相关性清洗数据剔除错误标签提高标注一致性RAG检索结果相关但生成答案差检索出的chunk内容互相冲突或过时打印检索结果检查chunk是否包含矛盾信息更新源数据增加文档有效期字段过滤过期内容训练loss下降但评测集分数不变训练集与评测集分布不一致分析训练集和评测集的词汇/主题分布补充多样化数据扩大评测集覆盖语义去重把不同文档误删embedding模型不适合该领域抽样查看被去重文档计算人工相似度调整相似度阈值或换领域微调的embedding模型处理大数据集OOM一次性读入内存查看代码是否使用read().splitlines()等内存大的操作改用line-by-line或Spark/Dask数据版本混乱手动复制覆盖数据集文件检查是否有多个data副本目录使用DVC 对象存储做版本管理标注结果不一致标注规范不清晰统计标注员之间的重合率编写标注手册定期校准用投票机制决定最终标签排查数据问题最核心的方法是“先定位到样本”。不要只看loss或accuracy要抽几条错误样例反向分析是数据本身的问题还是模型的问题。9. 给不同规模团队的数据工程建议9.1 个人开发者和独立项目如果你只是自己做个AI应用或研究项目不需要完整的平台但至少要养成三个习惯数据不直接覆盖每次清洗前保留一份raw数据。数据集版本化哪怕只是用git管理jsonl文件也要能追溯。实验记录每次微调或RAG改动时记录用的什么数据、怎么清洗、怎么切块。可以用一个简单的目录结构project/ ├── raw/ # 原始数据只读 ├── normalized/ # 归一化后的数据 ├── cleansed/ # 清洗去重后的数据 ├── chunks/ # 切块后的数据 ├── eval_sets/ # 评测集 └── configs/ # 数据处理配置9.2 中小型团队建议成立一个“数据小组”或让专人负责数据工程。不需要搞很大的平台优先打通三件事数据采集和归一化的自动化定时任务 统一schema。数据质量监控字符数、重复率、样本量、字段完整性。生成数据集的流程化清洗 - 切块 - 向量化 - 入库每一步输出版本号。工具选型上如果不是超大规模可以先用Python HuggingFace Datasets DVC Airflow。这些工具学习成本相对低社区资料多。9.3 大型团队字节成立一级数据部门说明大型团队确实需要独立的组织来定义数据标准和工具链。但大团队容易落入“重平台轻内容”的误区。真正有价值的是数据集本身而不是平台。所以即便组织庞大也要确保每条数据都能追溯到源头每个模型都能关联到具体数据版本。10. 总结像字节一样思考数据字节成立AI数据一级部门不是简单的组织比拼。它背后是一个技术判断数据工程是大模型下半场的基础设施谁能把数据质量、数据版本、数据飞轮做扎实谁就能持续迭代出更好的模型和应用。对普通开发者而言现在正是把“数据”当成第一等公民来对待的时候。不需要等到你有几TB数据才去搞治理从第一个RAG项目开始就应该把数据清洗、去重、评测集、版本管理做到位。这些能力会在你进入更复杂的AI项目时释放价值。最后提醒一点数据工作往往是枯燥的不像调参和看模型结构那么“性感”。但正是这些枯燥的清洗、去重、标注、版本控制工作最终决定了你的模型是聪明还是僵硬。字节用一级部门告诉行业数据值得被认真对待。接下来你也可以在自己的项目里认真对待它。 字节成立AI数据部门，数据工程成模型能力新天花板
2025年之后AI大模型领域的竞争已经从“拼模型结构”转向“拼数据质量”。字节跳动近期将AI业务的组织架构进一步收紧继Seed大模型团队、FlowAI应用团队之后又成立了一个专门面向数据的一级部门直接回应了“数据是AI核心竞争力”这一关键命题。这件事表面看是大厂组织调整背后其实是整个行业对数据处理、数据治理和数据集构建方式的重新定义。这篇文章不谈八卦只谈技术判断。我会从字节这个动作出发拆解三个问题为什么数据部门要独立成一级部门数据工程在AI链路里到底卡在哪些环节普通开发者和团队能从中借鉴哪些可落地的数据基础设施方法论如果你正在做大模型应用、RAG系统、模型微调或Agent开发这篇文章里提到的数据清洗、数据版本管理、评测集构建、数据飞轮等思路可以直接拿回自己的项目里用。1. 字节成立AI数据部门背后是对AI竞争本质的一次重新定价很多人看到“字节成立AI数据部门”的第一反应是字节又在搞组织架构调整。但从技术视角看这次调整的信息量远大于组织本身。Seed负责基础模型Flow负责AI原生应用而数据部门独立出来说明字节已经意识到数据不是模型的附属品而是和模型、应用并列的第三极。过去几年大模型的发展逻辑是“规模法则”模型越大、数据越多效果越好。但到了2024年以后大家发现一个问题公开互联网的高质量文本数据快被“挖空”了。各家训练数据集的重复率越来越高模型之间的差异越来越小。这时候谁能构建更高质量、更独特、更结构化、更贴合业务场景的数据集谁就能在模型效果上拉开差距。字节这个动作等于把“数据”从后台职能提升到了战略级部门。这背后的技术逻辑是数据已经不只是训练时的“燃料”而是贯穿模型预训练、监督微调、对齐、评测、知识更新、Agent工具调用的全链路基础设施。数据部门的独立意味着字节要用专业团队专门处理数据采集、清洗、去重、标注、配比、版本管理、质量评估这些脏活累活。对我们普通开发者的启示是如果你还在靠“爬点公开数据直接灌给模型”的方式做应用很快会被数据工程能力拉开差距。这不是说个人需要做一个庞大的数据团队而是要在自己的项目里建立“数据治理”的最小闭环。2. 大模型时代的数据不再只是“存起来”的数据先厘清一个概念。传统软件工程里数据管理关注的是结构化数据、事务一致性、查询性能技术栈是MySQL、Redis、Kafka、Hadoop这些。大模型时代的数据至少包含四种形态每一种都有完全不同的工程技术难点。第一种是预训练语料。这类数据量级最大动辄几十TB甚至PB级别来源是网页、书籍、论文、代码、音视频字幕。难点不是存而是清洗和去重。一个网页里有大量导航栏、广告、重复段落如果直接喂给模型模型学到的是噪声而不是知识。第二种是指令数据。这是微调和对齐的关键。比如“请帮我总结这篇文章”对应的期望输出是“这篇文章主要讲了……”。这类数据通常是人工标注或从用户反馈中挖掘量级比预训练语料小很多但质量要求极高。一条坏的指令数据可能让模型学会错误行为。第三种是评测数据。这是最容易被忽视的一类。没有评测集你就无法量化模型升级前后的效果变化。很多团队做模型微调时觉得loss下降了就是效果好结果一上线实测模型反而变笨了。原因就是评测集覆盖度不够或者评测集本身和训练集有重叠。第四种是业务数据/知识数据。这是企业私有数据也是RAG检索增强生成的核心。包括产品文档、客服记录、数据库内容、结构化知识图谱等。难点在于如何把非结构化数据拆成适合检索的块如何做向量化如何保持数据更新。字节独立数据部门本质上就是把这四类数据的能力统一收拢。而在我们自己的项目里也应该按这四个维度去规划数据工作而不是笼统地说“我们有很多数据”。3. 数据质量为什么成为模型能力的天花板2023年的时候业界流行一句话数据和模型是“垃圾进垃圾出”。现在这个说法已经不够准确了更准确的说法是数据质量决定了模型能力的上限模型架构决定的是逼近这个上限的效率。为什么要强调这一点因为很多团队在微调模型时花了大量时间调学习率、调LoRA rank、试各种prompt模板但最终效果不如别人一个干净的数据集。原因在于模型参数只是把数据里的规律固化下来数据里没有的规律模型再怎么调参也学不出来。举个例子。如果你做客服场景的模型微调收集了10000条客服对话。但里面的用户问题大多雷同真实场景中的长尾问题只占5%。模型训练完之后常见问题回答得很好一旦用户换个说法模型就答非所问。这不是模型不够聪明而是数据覆盖度不够。再比如做RAG系统时知识库文档里有大量重复段落、过时内容、格式混乱的表格。检索模块把相关片段捞出来后生成模块基于这些噪声内容做总结结果自然不理想。很多人以为是embedding模型选得不好实际是源数据没有做清洗和治理。字节数据部门的独立传递的一个重要信号就是以后评判一个AI团队的能力不能只看模型榜单分数还要看这个团队的数据构建、清洗、评测、迭代能力。而数据能力是可以积累的一旦形成“数据飞轮”效应后来者很难追上。4. 从组织调整看技术趋势数据工程正在成为AI的核心岗位如果你关注招聘市场会发现“数据工程师”和“AI数据工程师”的需求量在持续上升。传统的BI数据工程师偏向报表、数仓、ETL而AI方向的数据工程师需要理解模型训练逻辑、tokenizer、embedding、数据去重算法、指令数据构建、评估集设计。字节把数据部门提升为一级部门会带来一个连锁反应其他大厂也会跟进行业对数据工程岗位的定价会水涨船高。对开发者来说现在学习AI数据工程相当于2008年学习移动开发属于提前卡位。AI数据工程师需要掌握的技能包括大规模文本处理如使用Spark、Dask处理TB级数据、数据去重算法如MinHash、SimHash、语义去重基于embedding的聚类、数据标注流程设计、质量评估指标体系、数据版本管理如DVC、数据流水线编排如Airflow、Prefect、以及面向大模型的数据配比实验方法。这些技能和传统后端开发、算法工程师有交叉但侧重点不同。传统算法工程师关心模型结构AI数据工程师关心数据如何影响模型行为。这种“数据即代码”的思维方式正是字节这次调整想强调的。5. 构建你自己的AI数据基础设施从清洗到版本管理不管你是不是字节员工这套数据方法论都可以落到自己的项目里。下面我以一个典型的大模型知识库项目为例演示如何从零构建一个最小可用的AI数据流水线。5.1 数据采集与归一化假设你要做一个企业内部知识库问答机器人。数据来源可能有内部Wiki页面HTML产品文档Markdown/PDF历史客服工单Excel/CSV工单留言JSON第一步把所有格式统一转成纯文本或Markdown并保留元数据来源、更新时间、作者、权限。下面是一个用Python做最小归一化的示例# normalize_docs.py import json import hashlib from pathlib import Path from bs4 import BeautifulSoup import markdownify def html_to_markdown(html_content: str) - str: soup BeautifulSoup(html_content, html.parser) # 去掉 script 和 style 中的内容 for tag in soup([script, style, nav, footer]): tag.decompose() return markdownify.MarkdownConverter().convert_html(soup.prettify()) def process_document(raw_path: Path, output_path: Path) - None: suffix raw_path.suffix.lower() if suffix .html: text html_to_markdown(raw_path.read_text(encodingutf-8)) elif suffix .md: text raw_path.read_text(encodingutf-8) elif suffix .json: data json.loads(raw_path.read_text(encodingutf-8)) # 假设 JSON 里有一个 content 字段 text data.get(content, ) else: text raw_path.read_text(encodingutf-8) doc { source: str(raw_path), text: text, chars: len(text), sha256: hashlib.sha256(text.encode(utf-8)).hexdigest(), } output_path.write_text(json.dumps(doc, ensure_asciiFalse, indent2), encodingutf-8) if __name__ __main__: raw_dir Path(./raw_docs) out_dir Path(./normalized_docs) out_dir.mkdir(exist_okTrue) for raw_file in raw_dir.rglob(*): if raw_file.is_file() and raw_file.suffix.lower() in (.html, .md, .json, .txt): output_file out_dir / f{raw_file.stem}.json process_document(raw_file, output_file) print(归一化完成输出目录:, out_dir)这段代码做的事情很简单但解决了大问题不同来源的文档有了统一的JSON结构后续清洗、切块、向量化都用这个统一格式而不是每种数据写一套解析逻辑。5.2 清洗规则去掉噪声保留语义单元清洗不是简单地把空行删掉。对于LLM场景清洗要考虑以下几点去掉页眉页脚、导航、版权声明等重复性字符串。去除HTML标签中的隐藏内容。合并被截断的段落。去重包括精确去重和语义去重。过滤质量过低的内容如字符数过短、乱码、无标点段落。一个实用的清洗函数如下# clean_text.py import re def clean_text(text: str) - str: # 去掉不可见字符和乱码 text re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f], , text) # 把多个空行压缩为一个 text re.sub(r\n{3,}, \n\n, text) # 去掉常见的页眉页脚模式这里按需调整 text re.sub(r(?m)^\s*(版权所有|Copyright|隐私政策|上一篇|下一篇)\s*$, , text) # 去掉多余空格 text re.sub(r[ \t]{2,}, , text) # 保留中文、英文、数字、常见标点其他换成空格 text re.sub(r[^\u4e00-\u9fff\u0030-\u0039\u0041-\u005a\u0061-\u007a\u3000-\u303f\uff00-\uffef\s.,!?;:()\-], , text) # 再次压缩空白 text re.sub(r\s, , text).strip() return text if __name__ __main__: sample 这是测试。\n\n\n\n版权所有XXX\n\n应该保留的内容。 print(clean_text(sample))注意清洗规则不能一刀切。比如代码文档里可能包含-、、{}这样的符号直接正则替换会把代码语义破坏。所以针对不同来源的数据应该维护不同的清洗规则并保留清洗前版本方便回溯。5.3 数据去重MinHash与语义去重大模型训练和RAG检索对重复数据都非常敏感。训练集重复度过高会导致模型记忆严重、泛化能力下降RAG知识库里重复文档会导致检索结果冗余浪费上下文窗口。精确去重可以用哈希但很多重复是“近似重复”比如两篇文章只有几个词不同。推荐用MinHash LSH做大规模去重这里给一个简化版# minhash_dedup.py import hashlib from datasketch import MinHashLSH, MinHash def tokenize(text: str) - set: # 简单中文分词按字符 n-gram或使用 jieba import jieba return set(jieba.cut_for_search(text)) def build_lsh(docs, threshold0.8, num_perm128): lsh MinHashLSH(thresholdthreshold, num_permnum_perm) minhashes {} for idx, doc in enumerate(docs): m MinHash(num_permnum_perm) for token in tokenize(doc): m.update(token.encode(utf-8)) lsh.insert(fdoc_{idx}, m) minhashes[fdoc_{idx}] m return lsh, minhashes if __name__ __main__: docs [ 大模型训练需要高质量数据, 大模型训练需要高质量数据集, 今天的天气真不错, ] lsh, minhashes build_lsh(docs) # 查询每个文档的近似重复 for key, m in minhashes.items(): result lsh.query(m) if len(result) 1: print(f{key} 与 {result} 近似重复)对于精读项目可以进一步做embedding级别的语义去重用CLIP、bge或text-embedding模型把文本向量化然后计算相似度矩阵将相似度超过阈值的文档合并或丢弃。5.4 数据切块Chunking的策略RAG系统里切块大小直接影响检索效果。常见误区是全库统一用固定长度切分。实际上切块策略应该结合文档结构判断。比如有标题结构的文档按标题层级切分。表格数据按行/列语义合并。代码文件按函数/类切分。长对话按轮次切分。一个混合切块示例# chunker.py from typing import List, Dict def chunk_by_headings(text: str, max_chunk_size: int 1000) - List[Dict[str, str]]: lines text.split(\n) chunks [] current_chunk [] current_title section for line in lines: if line.startswith(#) and current_chunk: content \n.join(current_chunk).strip() if content: chunks.append({title: current_title, content: content}) current_title line.lstrip(#).strip() current_chunk [line] else: current_chunk.append(line) if len(\n.join(current_chunk)) max_chunk_size: content \n.join(current_chunk).strip() if content: chunks.append({title: current_title, content: content}) current_chunk [] if current_chunk: content \n.join(current_chunk).strip() if content: chunks.append({title: current_title, content: content}) return chunks if __name__ __main__: markdown_text # 第一章 什么是数据治理 数据治理是一个长期工程。 ## 1.1 元数据管理 元数据是数据的数据。 # 第二章 数据质量 数据质量包括准确性、完整性、一致性。 for chunk in chunk_by_headings(markdown_text): print(chunk[title], -, chunk[content][:30])生产环境中还需要考虑chunk之间的重叠避免切块切断了关键上下文。通常重叠1-2个句子即可。5.5 数据版本管理与血缘数据一旦进入训练流程任何修改都可能影响模型效果。如果没有版本管理你很难回答“这个模型是用哪一版数据训练的”这个问题。推荐使用DVCData Version Control对数据集进行版本管理它类似Git但针对的是大文件。基本流程# 初始化仓库和 DVC git init dvc init # 添加数据目录生成 .dvc 文件 dvc add data/raw_docs git add data/raw_docs.dvc .gitignore git commit -m add raw docs v1 # 切换分支或回滚到旧版本 git checkout commit_id -- data/raw_docs.dvc dvc checkout同时在每次数据清洗处理时建议在输出数据集中写入一行 provenance来源信息{ version: 2025.06.01, source_files: [normalized_docs/abc.json, normalized_docs/def.json], clean_pipeline: clean_text.py:v1.2, dedup: minhash_dedup.py:v0.9, created_by: data_team }这样当模型表现异常时你能快速定位是哪一次数据处理改动导致的。6. 数据飞轮让数据越来越值钱的工程机制字节独立数据部门目标绝不只是做一次性数据集而是要建立数据飞轮。数据飞轮的核心逻辑是AI系统在使用过程中会产生新的交互数据这些数据经过筛选、清洗和标注之后重新进入训练集持续提升模型能力模型能力提升后又带来更多用户产生更多数据。飞轮听起来很美好落地却很难。难点在于不能把用户的原始输入直接当训练数据需要去除个人隐私信息。需要设计采样策略只收集那些能纠正模型错误的高价值数据。需要建立人工或自动标注流程把原始交互转换成指令格式。需要控制数据腐烂即旧数据可能过时需要定期清理。一个实际的飞轮管道可能长这样用户提问 - 模型回答 - 用户反馈点赞/点踩/修改 - 筛选出低质量回答 - 重新生成正确答案 - 人工抽检 - 写入微调数据集 - 定期微调模型 - 新模型上线 - 产生新反馈。实现这个管道你需要一个日志系统记录交互一个消息队列异步处理一个标注/审核平台以及模型版本管理。对于小团队可以先用Notion或Airtable做标注用Python脚本定期从数据库导出数据并格式化。这里我给出一个从用户反馈中筛选数据的最小示例# build_finetune_data.py import json from datetime import datetime, timedelta def extract_feedback_rows(db_connection, since: datetime): query SELECT id, user_query, bot_response, user_rating, revised_answer FROM ai_interaction_logs WHERE created_at %s AND user_query IS NOT NULL AND (user_rating bad OR revised_answer IS NOT NULL) with db_connection.cursor() as cursor: cursor.execute(query, (since,)) return cursor.fetchall() def convert_to_instruction(rows): instructions [] for row in rows: question row[user_query] if row[revised_answer]: answer row[revised_answer] else: # 这里应该调用一个更强的模型或者人工重写答案不能使用原错误回答 answer # 占位后续走人工标注队列 instructions.append({ instruction: 请回答用户的问题。, input: question, output: answer, }) return instructions if __name__ __main__: # 伪代码实际连接数据库 rows [] with open(feedback_log.json, r, encodingutf-8) as f: rows json.load(f) instructions convert_to_instruction(rows) with open(finetune_data.jsonl, w, encodingutf-8) as f: for item in instructions: if item[output]: # 只保留有正确答案的样本 f.write(json.dumps(item, ensure_asciiFalse) \n)注意从用户反馈里直接拿答案是有风险的。用户修改后的回答不一定正确所以必须经过抽检。数据飞轮的关键不是自动而是“有质量门槛的自动”。7. 数据评测没有评测集一切优化都是空谈很多团队投入大量精力构建训练数据却忽略了评测数据。字节数据部门如果只是建数据和清洗数据而不定义评测标准那模型方向就难以控制。评测数据应该是独立的、静态的、与训练数据隔离的。评测集至少要覆盖四类能力领域知识问答检验模型对业务知识的掌握。指令遵循检验模型能否按复杂指令执行。格式输出检验模型能否稳定输出JSON、Markdown等格式。对抗样本检验模型对诱导、歧义、噪声的鲁棒性。一个简单但有效的做法是用Git维护评测集版本并在每次模型迭代后记录评测结果。# evaluate.py import json from openai import OpenAI # 以 OpenAI 兼容接口为例实际项目请替换成你的模型 endpoint client OpenAI(base_urlhttp://localhost:8000/v1, api_keyEMPTY) def evaluate_model(eval_path: str, model: str) - dict: with open(eval_path, r, encodingutf-8) as f: cases json.load(f) hit 0 for case in cases: prompt case[prompt] expected case[expected] response client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], temperature0, ) answer response.choices[0].message.content.strip() # 简单判断期望关键词是否出现在答案中 if expected in answer: hit 1 return {total: len(cases), hit: hit, accuracy: hit / len(cases)} if __name__ __main__: result evaluate_model(eval_sets/v1.0.json, your-model-name) print(评测结果:, result)生产级评测还需要考虑答案公平性、多选判断、模糊匹配等但核心思想一致评测集必须固定才能对比不同版本。8. 常见数据工程问题与排查思路在搭建AI数据管道时常见问题其实非常集中。下面整理成一张表方便快速检索问题现象可能原因排查方式解决方案模型微调后效果不如base训练集包含劣质数据或标签噪声抽样查看训练集计算每条数据与标签的相关性清洗数据剔除错误标签提高标注一致性RAG检索结果相关但生成答案差检索出的chunk内容互相冲突或过时打印检索结果检查chunk是否包含矛盾信息更新源数据增加文档有效期字段过滤过期内容训练loss下降但评测集分数不变训练集与评测集分布不一致分析训练集和评测集的词汇/主题分布补充多样化数据扩大评测集覆盖语义去重把不同文档误删embedding模型不适合该领域抽样查看被去重文档计算人工相似度调整相似度阈值或换领域微调的embedding模型处理大数据集OOM一次性读入内存查看代码是否使用read().splitlines()等内存大的操作改用line-by-line或Spark/Dask数据版本混乱手动复制覆盖数据集文件检查是否有多个data副本目录使用DVC 对象存储做版本管理标注结果不一致标注规范不清晰统计标注员之间的重合率编写标注手册定期校准用投票机制决定最终标签排查数据问题最核心的方法是“先定位到样本”。不要只看loss或accuracy要抽几条错误样例反向分析是数据本身的问题还是模型的问题。9. 给不同规模团队的数据工程建议9.1 个人开发者和独立项目如果你只是自己做个AI应用或研究项目不需要完整的平台但至少要养成三个习惯数据不直接覆盖每次清洗前保留一份raw数据。数据集版本化哪怕只是用git管理jsonl文件也要能追溯。实验记录每次微调或RAG改动时记录用的什么数据、怎么清洗、怎么切块。可以用一个简单的目录结构project/ ├── raw/ # 原始数据只读 ├── normalized/ # 归一化后的数据 ├── cleansed/ # 清洗去重后的数据 ├── chunks/ # 切块后的数据 ├── eval_sets/ # 评测集 └── configs/ # 数据处理配置9.2 中小型团队建议成立一个“数据小组”或让专人负责数据工程。不需要搞很大的平台优先打通三件事数据采集和归一化的自动化定时任务 统一schema。数据质量监控字符数、重复率、样本量、字段完整性。生成数据集的流程化清洗 - 切块 - 向量化 - 入库每一步输出版本号。工具选型上如果不是超大规模可以先用Python HuggingFace Datasets DVC Airflow。这些工具学习成本相对低社区资料多。9.3 大型团队字节成立一级数据部门说明大型团队确实需要独立的组织来定义数据标准和工具链。但大团队容易落入“重平台轻内容”的误区。真正有价值的是数据集本身而不是平台。所以即便组织庞大也要确保每条数据都能追溯到源头每个模型都能关联到具体数据版本。10. 总结像字节一样思考数据字节成立AI数据一级部门不是简单的组织比拼。它背后是一个技术判断数据工程是大模型下半场的基础设施谁能把数据质量、数据版本、数据飞轮做扎实谁就能持续迭代出更好的模型和应用。对普通开发者而言现在正是把“数据”当成第一等公民来对待的时候。不需要等到你有几TB数据才去搞治理从第一个RAG项目开始就应该把数据清洗、去重、评测集、版本管理做到位。这些能力会在你进入更复杂的AI项目时释放价值。最后提醒一点数据工作往往是枯燥的不像调参和看模型结构那么“性感”。但正是这些枯燥的清洗、去重、标注、版本控制工作最终决定了你的模型是聪明还是僵硬。字节用一级部门告诉行业数据值得被认真对待。接下来你也可以在自己的项目里认真对待它。