三分鐘快速入門大模型數據工程

作者：貝塔街的萬事屋 2025-05-07 00:10:00

人工智能

面向自然語言處理的大模型應用，數據工程的核心任務是構建一套高效、可擴展、可控的數據流程，從源頭到可用于模型訓練、推理的高質量文本語料，全鏈路打通。

向自然語言處理的大模型應用，數據工程的核心任務是構建一套高效、可擴展、可控的數據流程，從源頭到可用于模型訓練、推理的高質量文本語料，全鏈路打通。

數據工程的整體目標

構建穩定的數據流轉鏈路，從原始文本數據 → 清洗/標注/審核后的語料 → 符合模型輸入格式的數據集

保證數據可用性、完整性、質量、結構統一性

支持多階段使用場景：訓練/微調、推理RAG、評估測試等

數據來源

最常見的形式是甲方提供，如客戶工單、業務知識庫、客服聊天記錄、FAQ 等。

需盡早明確數據格式、內容范圍、隱私邊界。也需明確數據目標用途、是否可公開訓練、是否需要脫敏。
也可以自主收集，成本高、周期長，但更具通用性和自主性。數據收集有下面幾種方式：

手動采集：（如論壇、PDF、博客文章等）

網頁爬蟲：需遵循 robots 協議與合法邊界

調用開放接口:（如新聞、百科類 API）

AI 生成數據:（如大模型生成問答對，用于訓練增強）

AI 生成數據示例

以微調訓練金融科技領域問答模型為例，通過 AI 生成訓練數據。

流程：

初始化

設置兩種對話風格配置：
小白風格：通俗易懂的比喻解釋，限制術語使用
資深風格：專業術語和技術細節，要求嚴謹準確
每種風格包含：
系統角色設定（system_prompt）
示例對話（examples）
溫度參數（控制回答隨機性）

數據生成循環

構建包含系統提示和示例的對話上下文
從預設問題列表中隨機選取問題
調用 AI 接口獲取風格化回答
結構化存儲問答對（問題、回答、風格標簽）
加入延遲防止 API 限流

結果處理

合并所有生成的問答數據
處理可能的異常情況（如中斷請求）
將最終數據集保存為格式化的JSON文件

示例代碼：

import json
import time
import random
from zhipuai import ZhipuAI
# 初始化模型
client = ZhipuAI(api_key="48b393b3474e43f3b7eafceb662db81a.xxxxxx")
style_config = {
    "小白": {
        "system_prompt": "你是一位區塊鏈科普專家，需要用最簡單易懂的方式向完全不了解技術的普通人解釋區塊鏈概念。要求：\n1. 使用生活化比喻（如'像大家一起記賬本'）\n2. 避免專業術語，必須使用時立即解釋\n3. 回答控制在3句話以內\n4. 可適當使用??????等表情符號",
        "examples": [
            {"role": "user", "content": "區塊鏈到底是什么？"},
            {"role": "assistant",
             "content": "想象一群人在共同維護一個公開賬本??，每筆交易都記錄在所有人的本子上，想作弊就得改掉大部分人的記錄，這幾乎不可能??！"},
            {"role": "user", "content": "比特幣為什么值錢？"},
            {"role": "assistant", "content": "就像限量版球星卡??，總量固定(2100萬枚)，越來越多人想要時價格就上漲啦！"}
        ],
        "temperature": 0.2  # 低溫度值保證回答穩定性
    },
    "資深": {
        "system_prompt": "你是區塊鏈核心技術研究員，需滿足：\n1. 使用專業術語（如'橢圓曲線數字簽名''梅克爾帕特里夏樹'）\n2. 包含技術實現細節\n3. 必要時給出公式/代碼片段\n4. 引用最新論文成果（如2023 IEEE S&P會議）",
        "examples": [
            {"role": "user", "content": "請解釋zk-SNARKs的工作原理"},
            {"role": "assistant",
             "content": "zk-SNARKs依賴QAP(二次算術程序)將計算轉化為多項式關系：\n1) Setup生成CRS(公共參考串)\n2) Prover構造π=(A,B,C)滿足A*B=C+H*Z\n3) Verifier通過雙線性配對驗證e(A,B)=e(C,g)?e(H,g^z)\n參考：Groth16方案優化后驗證僅需3個配對運算。"},
            {"role": "user", "content": "分析EIP-4844對L2的影響"},
            {"role": "assistant",
             "content": "Proto-danksharding通過引入blob攜帶交易(每個blob≈125KB)，使Rollups數據可用性成本降低10-100倍。實測顯示：\n- Optimism交易費從$0.3→$0.007\n- zkSync證明生成時間減少40%"}
        ],
        "temperature": 0.5  # 中等溫度值允許一定創造性
    }
}
def generate_style_data(style_name, num_samples=50):
    config = style_config[style_name]
    data = []
    # 構建對話上下文 = 系統提示 + 示例對話
    messages = [
        {"role": "system", "content": config["system_prompt"]},
        *config["examples"]
    ]
    # 預定義的用戶問題集（可擴展）
    user_inputs = [
        "比特幣和區塊鏈是什么關系？",
        "智能合約真的安全嗎？",
        "普通人怎么參與DeFi？",
        "解釋一下零知識證明",
        "為什么NFT能賣那么貴？",
        "區塊鏈能防止數據造假嗎？"
    ]
    for _ in range(num_samples):
        try:
            # 隨機選擇問題
            user_msg = random.choice(user_inputs)
            # 當前對話 = 歷史上下文 + 新問題
            current_messages = messages + [
                {"role": "user", "content": user_msg}
            ]
            # 調用API（修正模型名稱）
            response = client.chat.completions.create(
                model="glm-4-plus",
                messages=current_messages,
                temperature=config["temperature"],
                max_tokens=150
            )
            # 獲取回復內容
            reply = response.choices[0].message.content
            data.append({
                "user": user_msg,
                "assistant": reply,
                "style": style_name
            })
            print("獲取最新回復內容: ", data[-1])
            time.sleep(0.5)  # 限流
        except Exception as e:
            print(f"生成失敗：{str(e)}")
    return data
if __name__ == '__main__':
    all_data = []
    try:
        print("開始生成小白風格數據...")
        gentle_data = generate_style_data("小白", 10)
        all_data.extend(gentle_data)
        print("開始生成資深風格數據...")
        gentle_data = generate_style_data("資深", 10)
        all_data.extend(gentle_data)
    except KeyboardInterrupt:
        print("\n用戶中斷，保存已生成數據...")
    finally:
        with open("style_chat_data.json", "w", encoding="utf-8") as f:
            json.dump(all_data, f, ensure_ascii=False, indent=2)
        print(f"數據已保存，有效樣本數：{len(all_data)}")

數據清洗與預處理

原始數據往往雜亂無章，需系統清洗處理。

內容清洗
去除 HTML 標簽、表情、特殊符號、水印等無效信息
統一編碼格式，規范文本結構（如換行、縮進）
分段處理
按語義或結構分段，確保模型輸入粒度適當
避免文本過長或過短，提升模型效果
內容篩選
刪除空白內容、重復內容、無意義語段

import json
from typing import List, Dict


def is_meaningless(text: str) -> bool:
    """檢查是否是無意義內容"""
    # 空白或超短內容
    if len(text.strip()) < 5:
        return True

    # 常見無意義模式
    meaningless_phrases = [
        "我不知道", "無法回答",
        "這個問題", "請重新提問"
    ]
    return any(phrase in text for phrase in meaningless_phrases)


def simple_clean(data: List[Dict]) -> List[Dict]:
    """基礎清洗流程"""
    seen = set()  # 用于去重
    cleaned_data = []

    for item in data:
        text = item["assistant"].strip()

        # 檢查空白/重復/無意義
        if (not text or
                text in seen or
                is_meaningless(text)):
            continue

        seen.add(text)
        cleaned_data.append({
            "user": item["user"],
            "assistant": text,
            "style": item["style"]
        })

    return cleaned_data


if __name__ == '__main__':
    # 加載原始數據
    with open('style_chat_data.json', 'r', encoding='utf-8') as f:
        raw_data = json.load(f)

    # 執行清洗
    cleaned_data = simple_clean(raw_data)

    # 保存結果
    with open('cleaned_data.json', 'w', encoding='utf-8') as f:
        json.dump(cleaned_data, f, ensure_ascii=False, indent=2)

    # 打印簡單報告
    print(f"原始數據量: {len(raw_data)}")
    print(f"清洗后保留: {len(cleaned_data)}")
    print(f"過濾數量: {len(raw_data) - len(cleaned_data)}")

過濾掉違法違規、廣告類內容，保障數據合規

數據標注與格式化

根據用途，對數據進行結構化和標注。

標注方式

自動標注：適合結構化文本、已有知識圖譜輔助生成
人工標注：適用于對話意圖、情感分類、多輪對話標記等復雜任務
AI 輔助+人工校對：效率與質量的平衡方式

數據集結構標準化

根據目標任務結構設計
問答：{"question": "", "answer": ""}
多輪對話：{"dialog": [{"role": "user", "content": ""}, {"role": "assistant", "content": ""}, ...]}
分類任務：{"text": "", "label": ""}統一存儲格式
JSONL
CSV
Parquet

數據審查與質檢

保證數據質量，避免模型效果被低質樣本污染。

審查內容

是否存在空文本？
文本長度是否合理？
是否出現亂碼、非法字符？
是否有標簽缺失、格式錯誤？

常規校驗機制

自動規則校驗（正則、長度范圍）
隨機抽樣人工審核
自動評分 + 低分樣本再審

文本去重與相似性篩查

消除重復文本，避免模型過擬合。

實現思路：

使用 Embedding 模型，將文本轉為向量表示
計算文本對之間的相似度（如余弦相似度或歐氏距離）
設置合理相似度閾值（如 > 0.95 即視為重復）
保留更優版本文本，剔除相似冗余樣本

import json
from sentence_transformers import SentenceTransformer
import numpy as np
def deduplicate_data(data_list, threshold=0.9):
    
"""
    語義去重：對 assistant 回復去重
    :param data_list: List[dict], 每個 dict 包含 user / assistant / style
    :param threshold: float, 相似度閾值
    :return: 去重后的 List[dict]
    """
    
model = SentenceTransformer("/root/autodl-tmp/models/thomas/text2vec-base-chinese")
    replies = [item["assistant"] for item in data_list]
    embeddings = model.encode(replies)
    kept = []
    seen = set()
    for i, emb_i in enumerate(embeddings):
        if i in seen:
            continue
        keep = True
        for j, emb_j in enumerate(embeddings):
            if j == i or j in seen:
                continue
            sim = np.dot(emb_i, emb_j) / (np.linalg.norm(emb_i) * np.linalg.norm(emb_j))
            print(f"sim({i}, {j}) = {sim:.4f}")
            if sim > threshold:
                seen.add(j)
        kept.append(data_list[i])
    return kept
if __name__ == '__main__':
    with open("style_chat_data.json", "r", encoding="utf-8") as f:
        all_data = json.loads(f.read())
        cleaned_data = deduplicate_data(all_data, 0.5)

數據統計與版本控制

記錄每輪數據處理中的：樣本數量、保留比例、清洗率、標注覆蓋率等
建立數據集版本控制機制（如 v1、v2、實驗版、正式版）
便于溯源、復現、對比實驗效果

數據安全與隱私保護

尤其是面向企業應用時，必須落實數據安全要求。

敏感字段脫敏（如姓名、電話、身份證、公司信息等）
權限分級，控制訪問范圍
明確數據使用范圍（訓練 or 部署 or 產品化）
加入審計與日志記錄

參考文獻：https://gcn0rm30wzh6.feishu.cn/docx/MsqhdNNkQotaNfxULM5csoqYnec

寫在最后

2025年的今天，AI創新已經噴井，幾乎每天都有新的技術出現。作為親歷三次AI浪潮的技術人，我堅信AI不是替代人類，而是讓我們從重復工作中解放出來，專注于更有創造性的事情，關注我們公眾號口袋大數據，一起探索大模型落地的無限可能！

另外目前我正在做兩個開源項目，一個本地知識庫還有一個是小紅書的智能發布工具，如果有需求的話可以加入下面的群聊中交流。

EasyRAG是我開源的一個本地知識庫項目，未來我會再里面分享使用EasyRAG做各種場景的問答或者文檔生成的最佳實踐文檔

責任編輯：龐桂玉來源：口袋大數據

大模型數據工程