外观
进阶教程：本地部署 Qwen3-4B-Instruct 进行多数据集混合训练
前言
本教程是多数据集混合训练的进阶实战，介绍如何通过 LoRA 微调让 Qwen3-4B-Instruct 模型同时掌握诗词创作和武侠小说写作能力。
与基础教程不同，本文注重原理讲解和完整代码实现，不再对模型下载、基础环境的安装配置进行说明，如想了解该方面的知识，可以参考本站其它文档。
关于 ModelScope： ModelScope 社区是一个模型开源社区及创新平台，由阿里巴巴通义实验室，联合 CCF 开源发展技术委员会，共同作为项目发起创建。（摘自官方文档）
核心特点:
- 完整可运行的代码流程
- 多数据集混合训练原理
- 关键参数详解与调优策略
- 一次性端到端完成训练
一、原理与架构
1.1 混合训练原理
核心思想: 将不同领域数据统一格式后合并，通过单次训练让模型同时学习多种能力。
数据流:
诗词数据 (text1) ───┐
├──> 统一为 {"text": ...} ──> 合并数据集 ──> LoRA训练
小说数据 (text) ───┘
模型输出:
├─ 诗词创作能力 ✓
└─ 武侠风格能力 ✓关键机制:
- 数据配比: 主领域占 80-90%，辅助领域占 10-20%，避免能力失衡
- 格式统一: 不同格式数据转换为统一的
{"text": "..."}结构 - LoRA 微调: 仅训练 1.16% 参数，保留原模型 98.84% 知识
1.2. 文件结构
d:/wwwroot/modelscope/
├── models/
│ └── Qwen3-4B-Instruct-2507/ # 基础模型
├── datasets/
│ ├── chinese-poetry/
│ │ └── train.json # 诗词数据集
│ ├── jinyong/
│ │ ├── *.txt # 原始小说文本
│ │ └── train_processed.json # 处理后的数据
│ └── merged/
│ └── train.json # 合并后的数据集 ⭐
├── output/
│ └── qwen3-poetry-jinyong/
│ ├── checkpoint-*/ # 训练检查点
│ └── final/ # 最终 LoRA ⭐
├── download_chinese_poetry.py # 诗词数据下载
├── download_jinyong_manual.py # 金庸数据处理
├── merge_datasets.py # 数据集合并 ⭐
├── train_merged_instruct.py # 训练脚本 ⭐
└── text_trained_merged_instruct.py # 测试脚本 ⭐1.2 为什么有效?
- 知识迁移: 诗词的韵律感增强小说的文学性
- 表达丰富: 小说的叙事手法拓展诗词的意境
- 参数共享: LoRA 在共享参数空间中学习多领域特征
1.3 技术栈
| 组件 | 选择 | 原因 |
|---|---|---|
| 基础模型 | Qwen3-4B-Instruct-2507 | 已具备指令理解能力 |
| 微调方法 | LoRA (r=64, α=128) | 参数高效，避免灾难性遗忘 |
| 训练框架 | Transformers + PEFT | 成熟稳定，社区支持好 |
| 数据集 | 诗词 + 金庸小说 | 风格互补，文学性强 |
二、完整代码实现
2.1 环境准备及模型下载
# 安装依赖（一次性安装全部所需库）
pip install modelscope transformers accelerate peft datasets pandas tensorboard -i https://mirrors.aliyun.com/pypi/simple/
# 单独安装 PyTorch CUDA 版本（必须！否则无法使用 GPU）
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu124
# 下载基础模型
D: # 切换到 D 盘（如果已在 D 盘可跳过）
cd d:\wwwroot\modelscope
modelscope download --model Qwen/Qwen3-4B-Instruct-2507 --local_dir ./models/Qwen3-4B-Instruct-25072.2 诗词数据处理文件代码
download_chinese_poetry.py将从 ModelScope 社区下载诗词数据集并进行处理。
"""简化版数据集下载脚本 - 直接从网页下载 CSV
如果 MsDataset.load() 出现路径问题,使用此脚本直接下载 CSV 文件
使用方法:
python download_chinese_poetry.py
"""
import os
import pandas as pd
import json
# 数据集直接下载链接(从 ModelScope 网页获取)
TRAIN_URL = "https://www.modelscope.cn/api/v1/datasets/modelscope/chinese-poetry-collection/repo?Revision=master&FilePath=train.csv"
TEST_URL = "https://www.modelscope.cn/api/v1/datasets/modelscope/chinese-poetry-collection/repo?Revision=master&FilePath=test.csv"
SAVE_DIR = "./datasets/chinese-poetry"
os.makedirs(SAVE_DIR, exist_ok=True)
print("=" * 60)
print("简化版数据集下载")
print("=" * 60)
print("直接从 ModelScope 下载 CSV 文件")
print("=" * 60)
print()
def download_csv(url: str, split_name: str):
"""下载 CSV 并转换为 JSON"""
print(f"[{split_name.upper()}] 正在下载...")
try:
# 读取 CSV
df = pd.read_csv(url)
print(f"✓ 下载完成,共 {len(df)} 条数据")
# 转换为字典列表
data_list = df.to_dict("records")
# 保存为 JSON
save_path = os.path.join(SAVE_DIR, f"{split_name}.json")
with open(save_path, "w", encoding="utf-8") as f:
json.dump(data_list, f, ensure_ascii=False, indent=2)
print(f"✓ 已保存到: {save_path}")
print()
# 显示数据样例
if len(data_list) > 0:
print("数据样例:")
print(json.dumps(data_list[0], ensure_ascii=False, indent=2))
print()
return len(data_list)
except Exception as e:
print(f"✗ 下载失败: {e}")
return 0
try:
# 下载训练集
train_count = download_csv(TRAIN_URL, "train")
# 下载测试集
test_count = download_csv(TEST_URL, "test")
# 统计
print("=" * 60)
print("下载完成!")
print("=" * 60)
print(f"训练集: {train_count} 条")
print(f"测试集: {test_count} 条")
print(f"总计: {train_count + test_count} 条")
print()
print("下一步:")
print(" python train.py")
print("=" * 60)
except Exception as e:
print(f"\n发生错误: {e}")
print("\n如果网络下载失败,请手动下载:")
print(
"1. 访问: https://modelscope.cn/datasets/modelscope/chinese-poetry-collection"
)
print("2. 点击 Files → 下载 train.csv 和 test.csv")
print(f"3. 放到目录: {SAVE_DIR}")2.3 金庸小说处理文件代码
download_jinyong_manual.py将从 ModelScope 社区下载金庸小说数据集并进行处理。
核心逻辑: 长文本切分 + 格式统一
"""金庸小说数据集处理脚本 - 手动下载版
说明:
由于 josonfan/jinyong 数据集的文本存储在单独的 .txt 文件中,
需要先手动下载这些文件,然后用此脚本处理。
手动下载步骤:
1. 访问: https://modelscope.cn/datasets/josonfan/jinyong/files
2. 下载所有 金庸-*.txt 文件到 ./datasets/jinyong/ 目录
3. 运行此脚本: python download_jinyong_manual.py
或者使用ModelScope命令行工具批量下载:
modelscope download --dataset josonfan/jinyong
"""
import os
import json
import glob
# 配置
DATASET_DIR = "./datasets/jinyong"
OUTPUT_FILE = os.path.join(DATASET_DIR, "train_processed.json")
print("=" * 60)
print("金庸小说数据集处理")
print("=" * 60)
print(f"数据目录: {DATASET_DIR}")
print("=" * 60)
print()
# 检查目录是否存在
if not os.path.exists(DATASET_DIR):
os.makedirs(DATASET_DIR, exist_ok=True)
print(f"⚠️ 创建目录: {DATASET_DIR}")
print()
# 查找所有 .txt 文件
txt_files = glob.glob(os.path.join(DATASET_DIR, "*.txt"))
if not txt_files:
print("⚠️ 未找到任何 .txt 文件!")
print()
print("请按以下步骤操作:")
print("=" * 60)
print("1. 访问: https://modelscope.cn/datasets/josonfan/jinyong/files")
print("2. 下载以下文件到 ./datasets/jinyong/ 目录:")
print(" - 金庸-天龙八部.txt")
print(" - 金庸-侠客行.txt")
print(" - 金庸-笑傲江湖.txt")
print(" - 金庸-倚天屠龙记.txt")
print(" - 金庸-越女剑.txt")
print(" - (以及其他你想训练的小说)")
print("3. 下载完成后再次运行此脚本")
print("=" * 60)
exit(0)
print(f"✓ 找到 {len(txt_files)} 个小说文件")
print()
# 读取并处理所有小说
all_texts = []
total_chars = 0
for txt_file in txt_files:
novel_name = os.path.basename(txt_file).replace(".txt", "").replace("金庸-", "")
print(f"正在处理: {novel_name}...")
try:
# 读取文本
with open(txt_file, "r", encoding="utf-8") as f:
content = f.read()
# 清理文本
content = content.strip()
if len(content) < 100:
print(f" ⚠️ 文件太短,跳过")
continue
# 切分成段落 (每段 3000-5000 字,用于训练)
chunk_size = 4000
overlap = 200 # 段落之间有200字重叠,保持连贯性
chunks = []
for i in range(0, len(content), chunk_size - overlap):
chunk = content[i : i + chunk_size].strip()
if len(chunk) > 500: # 只保留有效段落
chunks.append({"text": chunk, "source": novel_name})
all_texts.extend(chunks)
total_chars += len(content)
print(f" ✓ {len(content):,} 字 → {len(chunks)} 个段落")
except Exception as e:
print(f" ✗ 读取失败: {e}")
print()
print("=" * 60)
if not all_texts:
print("⚠️ 没有有效的文本数据!")
print("请检查 .txt 文件是否包含有效内容")
exit(1)
# 保存处理后的数据
print(f"正在保存到: {OUTPUT_FILE}")
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
json.dump(all_texts, f, ensure_ascii=False, indent=2)
print(f"✓ 保存完成!")
print()
# 统计信息
print("=" * 60)
print("数据集统计:")
print("=" * 60)
print(f"小说数量: {len(txt_files)}")
print(f"文本段落: {len(all_texts)}")
print(f"总字符数: {total_chars:,}")
print(f"平均段落: {total_chars // len(all_texts):,} 字/段")
print(f"字段格式: ['text', 'source']")
print("=" * 60)
print()
# 显示样例
print("数据样例:")
print("=" * 60)
if all_texts:
sample = all_texts[0]
print(json.dumps(sample, ensure_ascii=False, indent=2))
print()
print(f"文本预览: {sample['text'][:200]}...")
print("=" * 60)
print()
print("✓ 数据准备完成!")
print()
print("下一步:")
print(f" 使用 merge_datasets.py 合并诗词和金庸数据集")
print(f" 金庸数据集路径: {OUTPUT_FILE}")
print("=" * 60)参数说明:
chunk_size=4000: 每段长度,匹配模型输入限制overlap=200: 段落重叠,保持上下文连贯len(chunk)>500: 过滤过短无效段落
2.3 诗词数据集下载
诗词数据集已经是标准格式，直接下载即可:
python download_chinese_poetry.py输出:
./datasets/chinese-poetry/train.json(25000 条诗词)- 格式:
{"text1": "春眠不觉晓,处处闻啼鸟..."}
2.4 金庸小说数据集下载
金庸数据集需要特殊处理，因为原始数据是大段文本。
步骤 1: 下载小说文本
访问 https://modelscope.cn/datasets/josonfan/jinyong/files
手动下载以下文件到 ./datasets/jinyong/ 目录:
- 金庸-天龙八部.txt
- 金庸-侠客行.txt
- 金庸-笑傲江湖.txt
- 金庸-倚天屠龙记.txt
- 金庸-越女剑.txt
步骤 2: 处理成训练格式
python download_jinyong_manual.py处理逻辑:
- 读取每个
.txt文件的完整内容 - 切分成 4000 字左右的段落 (段落间有 200 字重叠保持连贯性)
- 格式化为 JSON:
{"text": "...", "source": "天龙八部"} - 保存为
./datasets/jinyong/train_processed.json(978 段)
为什么要切分?
- 原始小说太长 (如天龙八部 180 万字)，模型无法一次处理
- 切分后可以从多个角度学习写作风格
- 适合模型的输入长度限制 (256-512 tokens)
2.5 数据合并脚本
核心功能: merge_datasets.py将对多格式数据集进行统一 + 配比控制
"""多数据集合并工具
功能:
将多个数据集合并成一个大数据集,用于训练多领域模型
支持的数据格式:
1. {"text1": "文本内容"} # 纯文本
2. {"instruction": "指令", "output": "回答"} # 指令格式
3. {"text": "文本内容"} # 通用文本
4. 其他格式会自动转换
使用方法:
python merge_datasets.py
配置:
在下方 DATASETS 列表中添加你的数据集配置
"""
import os
import json
from typing import List, Dict, Any
# ==================== 数据集配置 ====================
# 添加你的数据集配置到这里
DATASETS = [
{
"name": "诗词数据集",
"path": "./datasets/chinese-poetry/train.json",
"format": "text1", # 字段名称
"max_samples": 5000, # 使用 5000 条诗词数据 (快速测试)
},
{
"name": "金庸小说",
"path": "./datasets/jinyong/train_processed.json",
"format": "text", # 已经是 {"text": "..."} 格式
"max_samples": 500, # 使用 500 个段落 (快速测试)
},
]
# 输出文件路径
OUTPUT_DIR = "./datasets/merged"
OUTPUT_FILE = os.path.join(OUTPUT_DIR, "train.json")
# ==================================================
print("=" * 60)
print("多数据集合并工具")
print("=" * 60)
print(f"输出文件: {OUTPUT_FILE}")
print("=" * 60)
print()
def load_dataset_file(config: Dict[str, Any]) -> List[Dict[str, str]]:
"""加载单个数据集文件"""
name = config["name"]
path = config["path"]
data_format = config["format"]
max_samples = config.get("max_samples")
print(f"[{name}]")
print(f" 路径: {path}")
if not os.path.exists(path):
print(f" ⚠️ 文件不存在,跳过")
return []
try:
# 读取 JSON 文件
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
# 限制数量
if max_samples and len(data) > max_samples:
data = data[:max_samples]
print(f" ✓ 加载 {len(data)} 条 (限制为 {max_samples} 条)")
else:
print(f" ✓ 加载 {len(data)} 条")
# 转换为统一格式: {"text": "内容"}
normalized_data = []
for item in data:
if data_format == "text1" and "text1" in item:
# 诗词格式
normalized_data.append({"text": item["text1"]})
elif data_format == "instruction-output":
# 指令格式
if "instruction" in item and "output" in item:
text = f"User: {item['instruction']}\nAssistant: {item['output']}"
normalized_data.append({"text": text})
elif data_format == "text" and "text" in item:
# 通用文本格式
normalized_data.append({"text": item["text"]})
else:
# 尝试自动识别
if "text1" in item:
normalized_data.append({"text": item["text1"]})
elif "text" in item:
normalized_data.append({"text": item["text"]})
elif "instruction" in item and "output" in item:
text = f"User: {item['instruction']}\nAssistant: {item['output']}"
normalized_data.append({"text": text})
else:
# 使用第一个字段
first_key = list(item.keys())[0]
normalized_data.append({"text": str(item[first_key])})
print(f" ✓ 格式化完成")
return normalized_data
except Exception as e:
print(f" ✗ 加载失败: {e}")
return []
def merge_datasets(dataset_configs: List[Dict[str, Any]]) -> List[Dict[str, str]]:
"""合并多个数据集"""
print("开始合并数据集...")
print()
all_data = []
stats = {}
for config in dataset_configs:
data = load_dataset_file(config)
all_data.extend(data)
stats[config["name"]] = len(data)
print()
return all_data, stats
def save_merged_dataset(data: List[Dict[str, str]], output_path: str):
"""保存合并后的数据集"""
print("=" * 60)
print("保存合并数据集...")
# 创建输出目录
os.makedirs(os.path.dirname(output_path), exist_ok=True)
# 保存为 JSON
with open(output_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"✓ 已保存到: {output_path}")
print(f"✓ 总计: {len(data)} 条数据")
print()
def main():
"""主函数"""
try:
# 检查是否有数据集配置
if not DATASETS:
print("⚠️ 没有配置数据集")
print("请在脚本中的 DATASETS 列表添加数据集配置")
return
# 合并数据集
merged_data, stats = merge_datasets(DATASETS)
if not merged_data:
print("⚠️ 没有数据可以合并")
return
# 保存合并数据集
save_merged_dataset(merged_data, OUTPUT_FILE)
# 显示统计信息
print("=" * 60)
print("统计信息:")
print("=" * 60)
for name, count in stats.items():
percentage = (count / len(merged_data)) * 100
print(f" {name}: {count} 条 ({percentage:.1f}%)")
print("-" * 60)
print(f" 总计: {len(merged_data)} 条")
print("=" * 60)
print()
# 显示数据样例
print("数据样例 (前 3 条):")
print("=" * 60)
for i, item in enumerate(merged_data[:3], 1):
print(f"\n[样例 {i}]")
print(item["text"][:200] + ("..." if len(item["text"]) > 200 else ""))
print()
print("=" * 60)
print()
print("下一步:")
print(f" 1. 检查合并后的数据集: {OUTPUT_FILE}")
print(f" 2. 修改 train_merged_instruct.py 中的 DATASET_NAME 为: {OUTPUT_FILE}")
print(f" 3. 运行训练: python train_merged_instruct.py")
print("=" * 60)
except Exception as e:
print(f"\n发生错误: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()关键参数:
max_samples: 控制每个数据集的使用量format: 指定数据字段名称 (text1或text)- 配比策略: 主领域 80-90%,辅助领域 10-20%
2.6 训练脚本
train_merged_instruct.py将对合并的数据集进行训练
完整实现:
"""Qwen3-4B-Instruct 诗词训练脚本 - LoRA 微调
功能:
1. 使用 Qwen3-4B-Instruct-2507 (指令版) 作为基础模型
2. 训练诗词 LoRA 权重
3. 快速测试配置 (20K 数据, 1 轮)
使用方法:
python train_merged_instruct.py
训练完成后,LoRA 权重会保存在 OUTPUT_DIR 目录
"""
import os
from dataclasses import dataclass, field
from typing import Optional
from modelscope import AutoModelForCausalLM, AutoTokenizer
from modelscope.msdatasets import MsDataset
from transformers import TrainingArguments, Trainer
from peft import LoraConfig, get_peft_model, TaskType
import torch
# ==================== 训练配置 ====================
# 模型配置
MODEL_PATH = "./models/Qwen3-4B-Instruct-2507" # Instruct 版本模型路径
MODEL_TYPE = "qwen3" # 模型类型
# 数据集配置
DATASET_NAME = "./datasets/merged/train.json" # 使用合并后的数据集
# 输出配置
OUTPUT_DIR = "./output/qwen3-poetry-jinyong" # 训练输出目录
LOGGING_DIR = "./output/qwen3-poetry-jinyong/logs" # 日志目录
# 断点续训配置
RESUME_FROM_CHECKPOINT = False # 第一次训练设为 False,之后改为 True
# LoRA 参数
LORA_R = 64 # LoRA 秩
LORA_ALPHA = 128 # LoRA 缩放系数
LORA_DROPOUT = 0.05 # Dropout 比例
# 训练超参数 (快速测试配置)
NUM_EPOCHS = 2 # 训练轮数 (多领域数据用 2 轮)
BATCH_SIZE = 2 # 批次大小
GRADIENT_ACCUMULATION_STEPS = 8 # 梯度累积步数
LEARNING_RATE = 1e-4 # 学习率
MAX_LENGTH = 256 # 最大序列长度
WARMUP_RATIO = 0.1 # 预热比例
SAVE_STEPS = 100 # 每隔多少步保存一次
LOGGING_STEPS = 10 # 每隔多少步记录一次日志
# 其他配置
USE_FP16 = True # 使用混合精度训练
GRADIENT_CHECKPOINTING = True # 启用梯度检查点
# ==================================================
print("=" * 60)
print("Qwen3-4B-Instruct 诗词+金庸 混合训练 - LoRA 微调")
print("=" * 60)
print(f"模型路径: {MODEL_PATH}")
print(f"数据集: {DATASET_NAME}")
print(f"输出目录: {OUTPUT_DIR}")
print(f"训练轮数: {NUM_EPOCHS}")
print(f"学习率: {LEARNING_RATE}")
print(f"LoRA 秩: {LORA_R}")
print("=" * 60)
print()
def load_dataset(dataset_name: str):
"""加载数据集"""
print(f"[1/5] 正在加载数据集: {dataset_name}")
try:
# 判断是否是本地文件
if dataset_name.endswith(".json"):
# 加载本地 JSON 文件
import json
from datasets import Dataset
with open(dataset_name, "r", encoding="utf-8") as f:
data = json.load(f)
dataset = Dataset.from_list(data)
# 快速测试: 使用前 20000 条数据
if len(dataset) > 20000:
print(f"⚠️ 快速测试模式: 原数据 {len(dataset)} 条, 使用前 20000 条")
dataset = dataset.select(range(20000))
print(f"✓ 数据集加载成功,共 {len(dataset)} 条数据")
return dataset
else:
# 尝试从 ModelScope 加载
dataset = MsDataset.load(dataset_name, split="train")
print(f"✓ 数据集加载成功,共 {len(dataset)} 条数据")
return dataset
except Exception as e:
print(f"✗ 加载失败: {e}")
print("提示: 检查数据集名称是否正确,或使用本地数据集路径")
return None
def load_model_and_tokenizer(model_path: str):
"""加载模型和分词器"""
print(f"\n[2/5] 正在加载模型: {model_path}")
# 加载分词器
tokenizer = AutoTokenizer.from_pretrained(
model_path, trust_remote_code=True, padding_side="right"
)
# 设置 pad_token
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token
# 加载模型
model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.float16 if USE_FP16 else torch.float32,
device_map="auto",
trust_remote_code=True,
)
# 启用梯度检查点
if GRADIENT_CHECKPOINTING:
model.gradient_checkpointing_enable()
print(f"✓ 模型加载成功")
print(f"✓ 参数量: {sum(p.numel() for p in model.parameters()) / 1e9:.2f}B (十亿)")
return model, tokenizer
def setup_lora(model):
"""配置 LoRA"""
print(f"\n[3/5] 正在配置 LoRA")
print(f" - LoRA 秩 (r): {LORA_R}")
print(f" - LoRA Alpha: {LORA_ALPHA}")
print(f" - Dropout: {LORA_DROPOUT}")
# LoRA 配置
lora_config = LoraConfig(
task_type=TaskType.CAUSAL_LM,
r=LORA_R,
lora_alpha=LORA_ALPHA,
lora_dropout=LORA_DROPOUT,
target_modules=["q_proj", "k_proj", "v_proj", "o_proj"], # Qwen3 的注意力层
bias="none",
)
# 应用 LoRA
model = get_peft_model(model, lora_config)
# 打印可训练参数
trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
total_params = sum(p.numel() for p in model.parameters())
print(f"✓ LoRA 配置完成")
print(f" - 可训练参数: {trainable_params / 1e6:.2f}M")
print(f" - 总参数: {total_params / 1e9:.2f}B")
print(f" - 可训练比例: {100 * trainable_params / total_params:.2f}% (节省显存!)")
return model
def preprocess_dataset(dataset, tokenizer):
"""预处理数据集"""
print(f"\n[4/5] 正在预处理数据集")
def tokenize_function(examples):
# 检查数据集字段类型
if "text1" in examples:
# 纯文本数据集(如中文诗词)
texts = examples["text1"]
elif "instruction" in examples and "output" in examples:
# 指令数据集
texts = []
for instruction, output in zip(examples["instruction"], examples["output"]):
text = f"User: {instruction}\nAssistant: {output}"
texts.append(text)
elif "text" in examples:
# 通用文本字段
texts = examples["text"]
else:
# 其他格式,尝试获取第一个字段
first_key = list(examples.keys())[0]
texts = examples[first_key]
# 分词
tokenized = tokenizer(
texts,
truncation=True,
max_length=MAX_LENGTH,
padding="max_length",
return_tensors="pt",
)
# 设置 labels (用于计算 loss)
tokenized["labels"] = tokenized["input_ids"].clone()
return tokenized
# 应用预处理
processed_dataset = dataset.map(
tokenize_function,
batched=True,
remove_columns=dataset.column_names,
desc="Tokenizing dataset",
)
print(f"✓ 数据集预处理完成")
return processed_dataset
def train(model, tokenizer, train_dataset):
"""训练模型"""
print(f"\n[5/5] 开始训练")
# 训练参数
training_args = TrainingArguments(
output_dir=OUTPUT_DIR,
num_train_epochs=NUM_EPOCHS,
per_device_train_batch_size=BATCH_SIZE,
gradient_accumulation_steps=GRADIENT_ACCUMULATION_STEPS,
learning_rate=LEARNING_RATE,
warmup_ratio=WARMUP_RATIO,
logging_dir=LOGGING_DIR,
logging_steps=LOGGING_STEPS,
save_steps=SAVE_STEPS,
save_total_limit=3, # 只保留最近 3 个检查点
fp16=USE_FP16,
gradient_checkpointing=GRADIENT_CHECKPOINTING,
report_to="tensorboard", # 使用 TensorBoard 记录日志
load_best_model_at_end=False,
remove_unused_columns=False,
)
# 创建 Trainer
trainer = Trainer(
model=model,
args=training_args,
train_dataset=train_dataset,
tokenizer=tokenizer,
)
# 开始训练
print()
print("-" * 60)
print("训练开始...")
print(
f"总训练步数: {len(train_dataset) // (BATCH_SIZE * GRADIENT_ACCUMULATION_STEPS) * NUM_EPOCHS}"
)
print(f"检查点保存间隔: 每 {SAVE_STEPS} 步")
if RESUME_FROM_CHECKPOINT:
print(f"断点续训: 开启 (将从最新检查点续训)")
print("-" * 60)
print()
trainer.train(
resume_from_checkpoint=(
RESUME_FROM_CHECKPOINT if RESUME_FROM_CHECKPOINT else None
)
)
# 保存最终模型
final_output_dir = os.path.join(OUTPUT_DIR, "final")
trainer.save_model(final_output_dir)
tokenizer.save_pretrained(final_output_dir)
print()
print("=" * 60)
print("✓ 训练完成!")
print(f"✓ LoRA 权重已保存到: {final_output_dir}")
print("=" * 60)
print()
print("下一步:")
print(f"1. 查看训练日志: tensorboard --logdir {LOGGING_DIR}")
print(f"2. 使用训练后的模型: python text_trained_merged_instruct.py")
print()
def main():
"""主函数"""
try:
# 1. 加载数据集
dataset = load_dataset(DATASET_NAME)
if dataset is None:
return
# 2. 加载模型和分词器
model, tokenizer = load_model_and_tokenizer(MODEL_PATH)
# 3. 配置 LoRA
model = setup_lora(model)
# 4. 预处理数据集
train_dataset = preprocess_dataset(dataset, tokenizer)
# 5. 训练
train(model, tokenizer, train_dataset)
except KeyboardInterrupt:
print("\n\n训练被用户中断")
except Exception as e:
print(f"\n\n训练出错: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()关键参数详解:
| 参数 | 值 | 作用 | 调优建议 |
|---|---|---|---|
LORA_R | 64 | LoRA 秩,控制可训练参数量 | 简单任务 32,复杂任务 64-128 |
LORA_ALPHA | 128 | 缩放系数,影响更新强度 | 通常为 r 的 2 倍 |
NUM_EPOCHS | 2 | 训练轮数 | 数据多用 1-2 轮,数据少用 3-5 轮 |
BATCH_SIZE | 2 | 批次大小 | 显存大可增加到 4-8 |
GRADIENT_ACCUMULATION | 8 | 梯度累积 | 等效批次 = BATCH_SIZE × 此值 |
LEARNING_RATE | 1e-4 | 学习率 | 多领域用 1e-4,单领域可用 2e-4 |
MAX_LENGTH | 256 | 序列最大长度 | 诗词用 256,小说用 512 |
训练资源:
- 5500 条 × 2 轮 ≈ 2-3 小时 (RTX 3090)
- 显存: 10-12 GB
2.7 测试脚本
text_trained_merged_instruct.py 将对训练后的混合数据集实际效果进行测试
"""Qwen3-4B-Instruct 模型混合数据集训练对话脚本
功能:
使用训练好的 Qwen3-4B-Instruct + 混合数据集 LoRA 进行对话
使用方法:
python text_trained_merged_instruct.py
"""
from modelscope import AutoModelForCausalLM, AutoTokenizer
from peft import PeftModel
from transformers import TextIteratorStreamer
import torch
# 配置
BASE_MODEL_PATH = "./models/Qwen3-4B-Instruct-2507" # 基础模型路径
LORA_PATH = "./output/qwen3-poetry-jinyong/final" # LoRA 权重路径 (诗词+金庸)
# 生成参数
MAX_NEW_TOKENS = 512 # 最大生成长度
TEMPERATURE = 0.8 # 温度 (越高越随机)
TOP_P = 0.9 # Top-p 采样
TOP_K = 50 # Top-k 采样
print("=" * 60)
print()
print("加载训练后的模型...")
print()
print("=" * 60)
print(f"基础模型: {BASE_MODEL_PATH}")
print(f"LoRA 权重: {LORA_PATH}")
print("=" * 60)
print()
# 加载分词器
print("[1/3] 正在加载分词器...")
tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL_PATH, trust_remote_code=True)
print("✓ 分词器加载完成")
print()
# 加载基础模型
print("[2/3] 正在加载基础模型...")
base_model = AutoModelForCausalLM.from_pretrained(
BASE_MODEL_PATH,
torch_dtype=torch.float16,
device_map="auto",
trust_remote_code=True,
)
print("✓ 基础模型加载完成")
print()
# 加载 LoRA 权重
print("[3/3] 正在加载 LoRA 权重...")
model = PeftModel.from_pretrained(base_model, LORA_PATH)
model.eval() # 设置为评估模式
print("✓ LoRA 权重加载完成")
print()
print("✓ 准备就绪!")
print()
print("=" * 60)
print("命令: quit(退出) | clear(清空历史) | history(查看历史)")
print("=" * 60)
print()
# 对话历史
conversation_history = [
{
"role": "system",
"content": "你是一位精通古诗词创作和武侠小说写作的 AI 助手。",
}
]
def chat_stream(messages: list):
"""流式生成回复"""
# 应用聊天模板
text = tokenizer.apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)
# 编码
inputs = tokenizer([text], return_tensors="pt").to(model.device)
# 生成
with torch.no_grad():
streamer = TextIteratorStreamer(
tokenizer, skip_prompt=True, skip_special_tokens=True
)
generation_kwargs = {
**inputs,
"max_new_tokens": MAX_NEW_TOKENS,
"temperature": TEMPERATURE,
"top_p": TOP_P,
"top_k": TOP_K,
"do_sample": True,
"streamer": streamer,
}
from threading import Thread
thread = Thread(target=model.generate, kwargs=generation_kwargs)
thread.start()
response = ""
for text in streamer:
print(text, end="", flush=True)
response += text
print() # 换行
return response.strip()
# 主循环
while True:
try:
# 获取用户输入
user_input = input("\n你: ")
# 处理命令
if user_input.lower() == "quit":
print("再见!")
break
elif user_input.lower() == "clear":
conversation_history = [
{
"role": "system",
"content": "你是一位精通古诗词创作的 AI 助手,擅长创作优美的古体诗词。",
}
]
print("✓ 对话历史已清空")
continue
elif user_input.lower() == "history":
print("\n对话历史:")
for i, msg in enumerate(conversation_history[1:], 1): # 跳过 system
role = "你" if msg["role"] == "user" else "AI"
print(f"{i}. {role}: {msg['content'][:50]}...")
continue
if not user_input.strip():
continue
# 添加用户消息
conversation_history.append({"role": "user", "content": user_input})
# 生成回复
print("\nAI: ", end="")
response = chat_stream(conversation_history)
# 添加助手回复
conversation_history.append({"role": "assistant", "content": response})
except KeyboardInterrupt:
print("\n\n再见!")
break
except Exception as e:
print(f"\n错误: {e}")
import traceback
traceback.print_exc()三、执行流程
3.1 完整执行步骤
# 步骤 1: 下载诗词数据集
python download_chinese_poetry.py
# 步骤 2: 下载金庸小说(需手动下载 .txt 文件)
# 访问 https://modelscope.cn/datasets/josonfan/jinyong/files
# 下载 5 个 .txt 文件到 ./datasets/jinyong/
python download_jinyong_manual.py
# 步骤 3: 合并数据集
python merge_datasets.py
# 步骤 4: 训练模型 (2-3 小时)
python train_merged_instruct.py
# 步骤 5: 测试模型
python text_trained_merged_instruct.py3.2 预期输出
测试 1 - 诗词创作:
你: 写一首关于春天的诗
AI: 《春思》
燕剪晴丝织昼长，东风吹雨湿花裳。
玉皇欲遣天孙织，万朵胭脂锁海棠。✅ 对仗工整、韵律准确、意象优美
测试 2 - 武侠风格:
你: 写一段武侠小说，描述一位侠客行侠仗义
AI: 暮色四合，江上孤舟一叶，船头立着个白衣人...
[完整的金庸风格武侠描写]✅ 环境细腻、人物鲜明、完全是金庸风格
测试 3 - 融合创作:
你: 用古诗的形式描述一位江湖侠客
AI: 《侠客吟》
长剑倚天秋水寒，孤舟一叶泛沧澜。
江声不语行人远，落日孤城雁数残。✅ 完美融合诗词格式和武侠意象
四、效果评估
4.1 评估维度
训练完成后，建议从以下维度验证效果：
| 维度 | 评估标准 | 测试方法 |
|---|---|---|
| 诗词能力 | 格式正确、韵律基本符合 | 输入“写一首关于 XX 的诗”，检查格式 |
| 武侠风格 | 叙事流畅、有武侠元素 | 输入“写一段武侠描写”，检查风格 |
| 风格切换 | 请求不同风格时能正确响应 | 交替测试诗词和小说请求 |
| 融合能力 | 能结合两种风格创作 | 输入“用古诗描述侠客”，检查效果 |
4.2 验证方法
基本验证：
- 诗词测试: 输入“写一首关于春天的七言绝句”，检查是否四句、每句七字
- 武侠测试: 输入“写一段武侠小说开头”，检查是否有场景描写、氛围渲染
- 对比测试: 同样提示词分别用原始模型和微调模型生成，对比差异
注意: 实际效果取决于数据质量、训练参数等因素，结果因人而异。
4.3 潜在问题与解决
| 问题 | 原因 | 解决方案 |
|---|---|---|
| 只会一种风格 | 数据配比失衡 | 调整 max_samples 平衡比例 |
| 风格混乱 | 训练不充分 | 增加 NUM_EPOCHS 或数据量 |
| 质量下降 | 学习率过高 | 降低 LEARNING_RATE |
| 过拟合 | 训练过度 | 减少 NUM_EPOCHS 或增加数据 |
五、进阶优化
5.1 增加数据量
如果效果满意，可以增加数据量获得更好效果:
# merge_datasets.py
DATASETS = [
{
"name": "诗词数据集",
"max_samples": 15000, # 从 5000 增加到 15000
},
{
"name": "金庸小说",
"max_samples": 1500, # 从 500 增加到 1500
},
]5.2 调整训练参数
# train_merged_instruct.py
NUM_EPOCHS = 3 # 增加训练轮数
LEARNING_RATE = 5e-5 # 降低学习率（更稳定）
MAX_LENGTH = 512 # 增加序列长度（小说更长）5.3 添加更多领域
可以加入第三个、第四个数据集:
DATASETS = [
{"name": "诗词", "max_samples": 10000},
{"name": "金庸", "max_samples": 1000},
{"name": "成语", "max_samples": 2000}, # 新增
{"name": "对联", "max_samples": 1000}, # 新增
]六、常见问题
Q1: 为什么诗词用 5000 条，金庸只用 500 段?
A: 数据配比遵循“主领域优先”原则:
- 本教程以诗词为主要训练目标，占主导 (90%)
- 金庸作为辅助领域，提供额外风格能力 (10%)
- 配比可根据实际需求调整，如需强化武侠能力可提高金庸比例
Q2: 能不能先训练诗词，再训练金庸?
A: 可以，但不推荐:
- 问题: 后训练的数据会"覆盖"先训练的知识（灾难性遗忘）
- 方案 1 (本教程): 混合训练，避免遗忘
- 方案 2: 多 LoRA 分别训练，需要切换使用
Q3: 训练时显存不够怎么办?
A: 调整以下参数降低显存:
BATCH_SIZE = 1 # 减小批次
MAX_LENGTH = 128 # 减小序列长度
GRADIENT_ACCUMULATION_STEPS = 16 # 增加梯度累积Q4: 如何判断训练是否成功?
A: 观察以下指标:
- Loss 趋势: 训练过程中 Loss 持续下降并趋于稳定
- 能力验证: 诗词和武侠两种风格都能正常生成
- 风格独立: 请求诗词时不会输出小说，反之亦然
- 对比测试: 同样的提示词，微调后效果明显优于原始模型
Q5: 混合训练会不会降低单个领域的能力?
A: 配比合理则不会，配比失衡则会:
- 配比合理时：主领域能力保持，辅助领域为额外收获
- 配比失衡时：数据量小的领域会被“稀释”，能力可能不如单独训练
- 本教程策略：诗词 90% + 金庸 10%，确保主领域不受影响
- 验证方法：训练后分别测试两个领域，对比原始模型效果
七、总结
恭喜！到这里你已经掌握了基于魔搭社区 Qwen3-4B-Instruct-2507 进行多数据集混合 LoRA 微调训练的完整流程。
本教程内容回顾
我们从零开始，依次完成了以下关键步骤：
- 原理讲解：理解混合训练的核心思想、数据配比策略和技术架构
- 数据集准备：下载并处理诗词数据集（25000 条）和金庸小说数据集（978 段）
- 数据合并：配置数据比例（5000 诗词 + 500 金庸），统一格式后合并为 5500 条训练数据
- 模型训练：配置 LoRA 参数，执行 2 轮训练（约 2-3 小时）
- 效果测试：加载基础模型 + LoRA 权重，验证诗词、武侠和融合创作能力
- 效果评估：通过多维度指标科学评估训练效果
- 常见问题：处理数据配比、显存不足、训练效果等问题
学习掌握的能力
完成本教程后，一般来说将能够：
✅ 独立完成混合训练：掌握多数据集合并 + LoRA 微调的完整流程
✅ 使用 ModelScope 平台：熟悉魔搭社区的模型下载和数据集管理方式
✅ 配置数据配比策略：理解主领域与辅助领域的数据量平衡方法
✅ 排查常见错误：处理数据格式不一致、显存不足、训练效果差等问题
✅ 评估训练效果：通过多维度指标科学验证模型能力
✅ 开发专业生成应用：创建支持多领域融合能力的 AI 工具
✅ 控制生成质量：通过训练参数调节优化输出效果
✅ 扩展新领域能力：掌握添加新数据集、调整配比的完整流程
适用场景与应用方向
基于实际测试，多数据集混合训练方案在不同场景下的表现：
方案特点对比：
| 特性 | 混合训练 | 单独训练 | 使用建议 |
|---|---|---|---|
| 融合能力 | ⭐⭐⭐⭐⭐ | ⭐⭐ | 需要风格融合创作选混合训练 |
| 训练效率 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | 一次训练获得多领域能力选混合训练 |
| 专业深度 | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ | 追求单领域极致效果选单独训练 |
| 存储占用 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | 只需一个 LoRA 文件，存储更节省 |
适用场景：
- 文学创作辅助（诗词创作、小说续写、风格融合）
- 教育场景（古诗词教学、文学鉴赏、创意写作）
- 游戏开发（武侠游戏剧情、诗词答题、角色对白）
- AI 助手（文学类聊天机器人、创意内容生成）
后续学习建议
如果想进一步提升，可以尝试：
- 扩展数据集：加入更多文学领域数据（对联、成语、古文等）
- 尝试单独训练：训练多个独立 LoRA，体验不同训练策略的差异
- 学习 QLoRA 微调：使用量化技术进一步降低训练成本
- 部署为 API 服务：使用 FastAPI 封装推理服务，支持 Web 应用集成
- 结合 RAG 技术：将微调模型与知识库检索结合，构建智能问答系统
感谢阅读
希望这篇教程能帮助你顺利掌握多数据集混合 LoRA 微调技术！如果遇到问题，欢迎反馈和交流。
祝你在 AI 领域不断探索，收获满满！