资深软件工程师AI转型之路 | 原理篇03：数据工程一——高质量大模型训练文本语料准备

m0_46187008

676人浏览 · 2026-03-18 00:01:09

m0_46187008 · 2026-03-18 00:01:09 发布

一、前言

系列承接：原理篇01（CPU训练全流程）、原理篇02（大模型架构体系）。本篇定位：从零手搓大模型 · 数据工程第一步训练文本语料源头获取与合规治理。

数据工程相关语料文件、代码等都将跟随博客展开，而统一归档到如下地址：https://github.com/coder-sswdg/data_engineering

从传统后端、客户端、架构师等软件工程师，转向大模型算法工程与LLM开发，直接上手模型架构，完全忽略数据工程，会带来不少问题。

在真实工业界，有一条被无数项目验证的铁律：数据决定模型上限，工程与算法只是不断逼近这个上限。

尤其在大模型预训练、微调、指令学习中：

语料不干净 → 模型输出杂乱、逻辑混乱
语料不合规 → 企业面临版权风险、法律纠纷
语料不均衡 → 模型偏科、泛化能力差
语料低质量 → 训练再多轮次也毫无意义

作为大模型数据工程开篇第一篇，本篇聚焦高质量训练文本从哪里来、如何安全获取、如何选择、各自特点是什么，一次性把语料源头讲解清楚。

将包括：

✅ 全种类语料来源深度解析（开源/商业/合成/自建/爬取）

✅ 中英文无版权、可商用、可教学语料完整方案

✅ 企业级选型策略与风险避坑

✅ 一键可运行代码，直接产出训练底座

✅ 标准化工程目录，承接后续全流程数据链路

二、高质量大模型训练语料

在进入语料来源之前，先明确工业界对训练文本的硬性标准：

合规无版权风险：可商用、可分发、无法律隐患
文本质量高：语句通顺、逻辑正常、无乱码无噪声
领域均衡：通用/知识/专业内容比例合理
无冗余重复：避免模型死记硬背、过拟合
语言适配：支持中英文、多语言统一格式

本篇所有内容，都围绕以上5点展开。

三、个人练习项目 vs 企业级大模型项目

本项目基于 data_engineering 构建，定位为个人学习、开源演示、轻量级CPU验证，和企业商用预训练有明确差异，便于大家自我定位：

维度	个人练习项目（本项目）	企业级商用大模型项目
语料规模	小而精（百万字符级）	超大（TB~PB级、万亿Token）
语料来源	开源无版权 + 原创生成	商业授权合规语料
版权要求	可公开分享、可教学	可商用、可追责、全授权
数据质量	满足学习验证即可	深度清洗、质检、质量打分
领域多样性	通用知识为主	全领域、多体裁、多风格
算力依赖	单机CPU即可运行	分布式集群、GPU/AI芯片
整体成本	0	极高（数据+清洗+人力+授权）
使用场景	学习、练手、博客开源	正式商用、对外服务

一句话总结

个人项目追求：安全、轻量、可跑通全流程

企业项目追求：合规、高质量、可落地商用

四、业界主流训练语料来源全解析

下面是企业真实选型会用到的完整对比，作者把每一类语料的来源、特点、质量、风险、适用场景、成本全部展开说明。

1. WikiText / 维基百科类开源语料

代表：WikiText-2、WikiText-103、维基百科中文/英文dump

协议：CC BY-SA / MIT / CC BY 4.0（可商用、可修改）

语言：中英文齐全

规模：小～中等

特点：

内容以百科知识为主，结构规范、逻辑性强
几乎无广告、无乱码、无低质内容
句式正式，非常适合大模型基础预训练

优点：

完全开源免费
版权清晰，教学/商用都安全
质量稳定，噪音极低

缺点：

内容偏向知识科普，领域单一
口语化、对话类内容极少

适合人群：

个人学习、教学演示、小模型训练、开源项目

2. 国内开源中文语料（百度/中科院/高校）

代表：CLUECorpus、THUCNews、百度百科开放语料、WuDao开源语料

协议：Apache 2.0 / 可商用协议

语言：中文为主，部分支持中英

规模：中等

特点：

专为中文NLP优化，更符合中文表达习惯
包含新闻、百科、社区、文章等多种体裁
格式统一，清洗成本低

优点：

免费、可商用
中文友好，比国外语料更适配国内模型

缺点：

覆盖领域有限，多样性不足
不适合超大规模工业级预训练

适合：

中文小模型、垂直领域微调、学习研究

3. 商业机构授权售卖语料（工业级首选）

代表：

国内：澜舟、智源、海天、数美、第三方数据服务商
海外：The Pile、C4、BooksCorpus等授权版

授权方式：年度授权、永久授权、按语种授权

语言：全语种覆盖

规模：100B～10T tokens（超大规模）

特点（重点展开）：

全领域覆盖：新闻、书籍、论文、百科、问答、博客、说明书、文学、专业教材全覆盖。
深度清洗去重：已经完成去重、过滤、格式标准化、质量打分。
版权完全干净：全部来自正规授权出版物、公开授权平台、合作机构。
可定制：可按行业、领域、语言、质量等级定制。
附带质检报告：提供重复率、质量分、领域分布、语言分布等指标。

优点：

质量最高，直接用于企业级预训练
无法律风险，可商用可落地
节省大量数据清洗人力成本
多样性强，模型泛化能力最好

缺点：

价格极高（几万～数百万不等）
授权流程严格，需要签署合同
部分数据不允许二次转售

适合：

企业正式预训练、商用大模型、私有化部署、ToB项目

4. 大模型AI合成语料

方式：GPT-4、Claude、通义千问、文心一言批量生成

版权：自行生成 → 自有版权

语言：任意

规模：理论无限

特点：

完全可控，可指定格式、领域、难度、风格
可生成指令数据、对话数据、思维链数据
生成速度极快

优点：

低成本、高灵活
无版权风险
特别适合微调、指令学习

缺点：

存在幻觉，内容可能错误
同质化严重，多样性不足
不适合作为预训练主力语料

适合：

微调数据增强、小样本学习、垂直领域指令集

5. 自建原创语料（100%安全）

方式：自行撰写、团队整理、人工创作

版权：完全自有

特点：

绝对干净、绝对安全
可完全控制内容质量

缺点：

规模极小，成本高

适合：

教学演示、博客开源、小样本测试

6. 公开网页爬取语料（极高风险，不推荐）

方式：爬取知乎、公众号、新闻网站、百科、小说网站

风险：

严重侵犯版权，企业绝对不能用
内容噪声极大：广告、乱码、重复、排版混乱
低质内容多，清洗成本极高

仅适合：

内部学术研究，不可商用、不可开源、不可发布

完整版语料对比总表

语料来源	版权合规	语种	规模	成本	质量	风险	适合场景
WikiText/维基百科	极高	中英	中	0	高	无	学习/教学/小模型
百度/中科院开源	高	中文	中	0	中高	无	中文NLP/小模型
商业授权语料	极高	全语种	超大	高	顶级	无	企业预训练/商用
AI合成语料	高	全语种	无限	低	中	低	微调/增强
自建原创	极高	自定义	小	低	高	无	教学/演示
网页爬取	极低	全语种	超大	低	低	极高	内部研究

五、个人练习推荐：真正无风险开源中文语料

为保证 data_engineering 项目100% 安全可发布，推荐几类可直接使用、无版权风险的中文开源语料：

维基百科中文子集（CC BY-SA 4.0）：可商用、干净、规范，最适合个人练习。
CLUECorpus Small（Apache 2.0）：百度官方开源，中文友好，社区认可度高。
公版图书/古籍/现代公版文章：完全无版权，可随意使用。
悟道 / 智源开源小样本（标注可商用）：只使用官方明确开放的安全子集。

六、本篇最终选型：安全、无版权、可直接发布

为了保证本博客可发布、代码可运行、商用无风险，在本系列中最终选择语料来源如下：

中文：开源安全中文语料 + 100%原创文本（完全自有版权，大模型合成文本）
中文开源语料：中文维基百科（zhwiki）

两者结合，形成高质量、小而精、干净安全的训练语料底座。

七、无版权中英文语料一键获取脚本（完整版可运行）

7.1 项目目录结构（统一根目录 data_engineering）

data_engineering/
├── corpus/
│   ├── raw/           # 原始语料
│   ├── clean/         # 清洗后（下篇使用）
│   ├── tokenizer/     # 分词与词表
│   └── synthetic/     # 合成数据
└── 01_data_collection/
    └── get_txt_corpus_from_wiki.py  # 本篇脚本

7.2 安装依赖

pip install requests tqdm pathlib2

7.3 完整代码（目录创建 + 英文语料下载 + 中文原创生成）

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
一站式维基百科语料处理脚本
流程：下载bz2文件 → 解压 → 提取JSON → 转换为TXT
过程文件：corpus/processing
最终TXT：corpus/raw
"""

import os
import sys
import json
import re
import time
import requests
import bz2
import shutil
from tqdm import tqdm
import subprocess
from pathlib import Path

# ===================== 全局配置 =====================
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
RAW_DIR = os.path.join(PROJECT_ROOT, "corpus", "raw")          # 最终TXT目录
PROCESSING_DIR = os.path.join(PROJECT_ROOT, "corpus", "processing")  # 过程文件目录
ENCODING = "utf-8"
MAX_SIZE_MB = 50                                              # 每个文件最大50M
MAX_SIZE = MAX_SIZE_MB * 1024 * 1024                          # 字节数
WIKI_URL = "https://dumps.wikimedia.org/zhwiki/latest/zhwiki-latest-pages-articles.xml.bz2"
WIKI_BZ2_FILENAME = "zhwiki_latest.xml.bz2"
WIKI_XML_FILENAME = "zhwiki_latest.xml"

# ===================== 通用工具函数 =====================
def ensure_dirs():
    """确保所有目录存在"""
    os.makedirs(RAW_DIR, exist_ok=True)
    os.makedirs(PROCESSING_DIR, exist_ok=True)
    print(f"📁 目录准备完成")
    print(f"   - 过程文件: {PROCESSING_DIR}")
    print(f"   - 最终TXT: {RAW_DIR}")

def download_file(url, save_path, min_size=1*1024*1024):
    """下载文件（带进度条、完整性检查）"""
    if os.path.exists(save_path):
        if os.path.getsize(save_path) >= min_size:
            print(f"✅ {os.path.basename(save_path)} 已存在，跳过下载")
            return True
        else:
            print(f"⚠️ {os.path.basename(save_path)} 文件残缺，重新下载")
            os.remove(save_path)

    print(f"🔽 开始下载: {os.path.basename(save_path)}")
    try:
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
        }
        resp = requests.get(url, stream=True, headers=headers, timeout=1200)
        resp.raise_for_status()

        total_size = int(resp.headers.get("content-length", 0))
        with open(save_path, "wb") as f, tqdm(
            total=total_size, unit="B", unit_scale=True, desc=os.path.basename(save_path)
        ) as bar:
            for chunk in resp.iter_content(chunk_size=1024*1024):
                if chunk:
                    f.write(chunk)
                    bar.update(len(chunk))

        final_size = os.path.getsize(save_path)
        if final_size < min_size:
            raise Exception(f"文件太小({final_size} bytes)，小于最小要求({min_size} bytes)")
        
        print(f"✅ 下载完成: {os.path.basename(save_path)} ({final_size/1024/1024:.1f}MB)")
        return True
    except Exception as e:
        if os.path.exists(save_path):
            os.remove(save_path)
        print(f"❌ 下载失败: {str(e)}")
        return False

def decompress_bz2(bz2_path, output_path):
    """解压bz2文件（带进度条）"""
    if os.path.exists(output_path):
        print(f"✅ {os.path.basename(output_path)} 已解压，跳过")
        return True
    
    print(f"📦 开始解压: {os.path.basename(bz2_path)}")
    try:
        total_size = os.path.getsize(bz2_path)
        with open(bz2_path, 'rb') as f_in, open(output_path, 'wb') as f_out, tqdm(
            total=total_size, unit="B", unit_scale=True, desc="解压进度"
        ) as bar:
            decompressor = bz2.BZ2Decompressor()
            while True:
                chunk = f_in.read(1024*1024)  # 1MB chunks
                if not chunk:
                    break
                data = decompressor.decompress(chunk)
                f_out.write(data)
                bar.update(len(chunk))
        
        print(f"✅ 解压完成: {os.path.basename(output_path)}")
        return True
    except Exception as e:
        if os.path.exists(output_path):
            os.remove(output_path)
        print(f"❌ 解压失败: {str(e)}")
        return False

# ===================== Wiki提取JSON函数 =====================
def extract_title(text):
    """提取页面标题"""
    match = re.search(r"<title>(.*?)</title>", text)
    return match.group(1).strip() if match else "Untitled"

def extract_text(text):
    """提取并清理页面内容"""
    match = re.search(r"<text.*?>(.*?)</text>", text, re.DOTALL)
    if not match:
        return ""
    t = match.group(1)
    # 清理维基百科标记
    t = re.sub(r"{{.*?}}", "", t)
    t = re.sub(r"\[\[.*?\]\]", "", t)
    t = re.sub(r"<.*?>", "", t)
    t = re.sub(r"\s+", " ", t)
    return t.strip()

def process_wiki_to_json(xml_file, output_dir, json_mode=True):
    """流式处理XML文件生成JSON（不爆内存）"""
    print(f"🔧 开始提取Wiki内容到JSON: {os.path.basename(xml_file)}")
    
    buffer = []
    file_count = 1
    max_size = MAX_SIZE  # 50M per file
    current_size = 0
    json_path = os.path.join(output_dir, f"wiki_{file_count:04d}.json")
    out = open(json_path, "w", encoding=ENCODING)

    try:
        with open(xml_file, "rb") as f:
            for line_bytes in f:
                try:
                    line = line_bytes.decode(ENCODING, errors="ignore")
                except:
                    continue

                # 捕获页面边界
                if "<page>" in line:
                    buffer = []
                elif "</page>" in line:
                    text = "".join(buffer)
                    title = extract_title(text)
                    content = extract_text(text)

                    if len(content) > 200:  # 过滤短内容
                        item = json.dumps({"title": title, "text": content}, ensure_ascii=False) + "\n"
                        item_size = len(item.encode(ENCODING))

                        # 超过大小限制则新建文件
                        if current_size + item_size > max_size:
                            out.close()
                            file_count += 1
                            json_path = os.path.join(output_dir, f"wiki_{file_count:04d}.json")
                            out = open(json_path, "w", encoding=ENCODING)
                            current_size = 0

                        out.write(item)
                        current_size += item_size
                else:
                    buffer.append(line)
        
        out.close()
        print(f"✅ JSON提取完成，生成 {file_count} 个JSON文件")
        return True
    except Exception as e:
        print(f"❌ JSON提取失败: {str(e)}")
        if out:
            out.close()
        return False

# ===================== JSON转TXT函数 =====================
def convert_json_to_txt(json_dir, output_dir):
    """将JSON文件转换为TXT文件"""
    print(f"📝 开始转换JSON到TXT，每个文件最大 {MAX_SIZE_MB}MB")
    
    file_index = 1
    current_size = 0
    out = None
    start_time = time.time()

    # 新建TXT文件函数
    def new_file():
        nonlocal out, current_size
        if out:
            out.close()
        filename = f"wiki_{file_index:02d}_{MAX_SIZE_MB}M.txt"
        path = os.path.join(output_dir, filename)
        out = open(path, "w", encoding=ENCODING)
        current_size = 0
        print(f"   新建TXT文件: {filename}")

    new_file()

    # 遍历所有JSON文件
    json_files = [f for f in sorted(os.listdir(json_dir)) if f.endswith(".json")]
    if not json_files:
        print("⚠️  未找到JSON文件")
        return False

    for fname in json_files:
        fpath = os.path.join(json_dir, fname)
        print(f"   处理JSON: {fname}")

        with open(fpath, "r", encoding=ENCODING) as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue

                try:
                    data = json.loads(line)
                    text = data.get("text", "")
                    if len(text) < 100:  # 过滤过短内容
                        continue

                    # 准备输出内容
                    output_text = text + "\n\n"
                    byte_size = len(output_text.encode(ENCODING))

                    # 超过大小限制则新建文件
                    if current_size + byte_size > MAX_SIZE:
                        file_index += 1
                        new_file()

                    out.write(output_text)
                    current_size += byte_size

                except Exception as e:
                    continue

    if out:
        out.close()

    # 统计结果
    elapsed = time.time() - start_time
    print(f"\n✅ TXT转换完成！")
    print(f"   📂 输出目录: {output_dir}")
    print(f"   ⏱  耗时: {elapsed:.2f}s")
    print(f"   📄 生成文件数: {file_index} 个")
    return True

# ===================== 主流程函数 =====================
def main():
    print("================================================")
    print("      一站式维基百科语料处理脚本")
    print("      流程：下载 → 解压 → 提取JSON → 转换TXT")
    print("================================================")
    
    # 1. 准备目录
    ensure_dirs()
    
    # 2. 下载维基百科bz2文件
    bz2_path = os.path.join(PROCESSING_DIR, WIKI_BZ2_FILENAME)
    if not download_file(WIKI_URL, bz2_path, min_size=300*1024*1024):  # 最小300MB
        sys.exit(1)
    
    # 3. 解压bz2文件
    xml_path = os.path.join(PROCESSING_DIR, WIKI_XML_FILENAME)
    if not decompress_bz2(bz2_path, xml_path):
        sys.exit(1)
    
    # 4. 提取Wiki内容到JSON
    if not process_wiki_to_json(xml_path, PROCESSING_DIR):
        sys.exit(1)
    
    # 5. 转换JSON到TXT（最终输出到raw目录）
    if not convert_json_to_txt(PROCESSING_DIR, RAW_DIR):
        sys.exit(1)
    
    # 6. 完成提示
    print("\n🎉 所有处理完成！")
    print(f"📁 最终TXT文件位置: {RAW_DIR}")
    print(f"📁 过程文件位置: {PROCESSING_DIR}（可按需删除）")

if __name__ == "__main__":
    main()

7.4 最终效果

可一站式完成维基百科中文语料的下载、JSON提取以及最终的txt文本语料生成，为后续的语料清洗、质量评估、分词等奠定基础。

八、企业级语料选型建议（仅供拓展）

如果你是工程师给公司做技术方案，直接用下面结论：

预训练主力：商业授权高质量语料（安全、高质量、省人力）
中文补充：百度/中科院开源中文语料
微调增强：AI合成指令数据 + 人工标注
学习测试：WikiText + 自建原创（本篇方案）
绝对禁止：爬取互联网数据

九、本篇总结

作为大模型数据工程的第一篇——语料获取与合规治理，我们完成了最重要的“地基工作”：

系统梳理了6大类训练语料的来源、特点、风险与适用场景
重点展开了商业语料的工业级价值与企业选型逻辑
提供了100%安全无版权、可直接发布的中英文语料方案
给出了可直接运行、工程化、CPU友好的完整代码
搭建了可承接后续3篇内容的标准数据目录

数据工程没有捷径，源头干净合规，后面的所有步骤才有意义。

本文是《资深软件工程师AI转型之路》系列的第 [5] 篇。‌

‌上一篇‌：https://blog.csdn.net/m0_46187008/article/details/158841354?spm=1001.2014.3001.5501
‌下一篇‌：敬请期待
‌系列目录‌：https://blog .csdn.net/m0_46187008/category_13135461.html?spm=1001.2014.3001.5482

欢迎交流‌：如果本文对你有帮助或启发，欢迎‌点赞、收藏、关注‌。对于文中的内容有任何疑问或独到见解，也欢迎在评论区留言讨论，我们一起进步！

‌转载须知‌：转载、节选或引用本文内容，请务必注明出自本系列及作者，并保持内容完整性。商业用途请联系作者获得授权。

💡 提示：本篇为系列文章，建议收藏专栏，持续追更不迷路～