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項目時釋放價值。最后提醒一點數據工作往往是枯燥的不像調參和看模型結構那么“性感”。但正是這些枯燥的清洗、去重、標注、版本控制工作最終決定了你的模型是聰明還是僵硬。字節用一級部門告訴行業數據值得被認真對待。接下來你也可以在自己的項目里認真對待它。