本文基于魔搭社区 ModelScope 平台，介绍如何对 Qwen3-4B-Instruct-2507 进行多数据集独立 LoRA 微调，实现领域能力自由切换

关于 ModelScope： ModelScope 社区是一个模型开源社区及创新平台，由阿里巴巴通义实验室，联合 CCF 开源发展技术委员会，共同作为项目发起创建。（摘自官方文档）

前言

本教程将带完成两个独立的 LoRA 微调任务：

1. 诗词生成模型：基于古诗词数据集训练
2. 金庸小说生成模型：基于金庸武侠小说数据集训练

通过单独训练不同数据集，可以让同一个基础模型获得不同的专业能力，并通过切换 LoRA 权重实现快速领域切换。

由于该教程为进阶教程，故而对基础环境的配置、模型的下载安装、数据集的下载、基础依赖的安装等方面不在进行介绍，如想了解该方面的知识，可以参考本站其它文档。

---

一、环境准备

1.1 环境信息

说明： 以下是博主个人电脑的配置环境，本文档所有操作和测试结果均基于此环境。不同的硬件配置可能会有不同的表现，仅供参考。

项目	配置
操作系统	Windows 10
Python 版本	3.11.9
显卡	NVIDIA GeForce RTX 3060 (12GB 显存)
CUDA 版本	12.4
PyTorch	2.6.0+cu124
模型	Qwen3-4B-Instruct-2507

1.2 虚拟内存配置（Windows 用户推荐）

训练 Qwen3-4B-Instruct-2507 模型需要较大内存，博主个人电脑按以下方式调整虚拟内存（具体参数视硬件情况而定，仅供参考）：

1. 右键"此电脑" → 属性 → 高级系统设置
2. 性能 → 设置 → 高级 → 虚拟内存 → 更改
3. 取消"自动管理"
4. 设置初始大小 16384MB，最大 32768MB
5. 确定并重启

1.3 依赖安装

pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
pip install transformers>=4.37.0
pip install peft>=0.8.0
pip install datasets
pip install modelscope

---

二、项目结构

d:\wwwroot\modelscope\
├── models\
│   └── Qwen3-4B-Instruct-2507\          # 基础模型
├── datasets\
│   ├── chinese-poetry\
│   │   ├── train.json                   # 诗词训练集（388,599条）
│   │   └── test.json                    # 诗词测试集（1,710条）
│   └── jinyong\
│       ├── train_processed.json         # 金庸小说训练集（978段，390万字）
│       ├── 金庸小说-天龙八部.txt
│       ├── 金庸小说-倚天屠龙记.txt
│       └── ...
├── output\
│   ├── qwen3-poetry\final\              # 诗词模型 LoRA 权重
│   └── qwen3-jinyong\final\             # 金庸小说模型 LoRA 权重
├── download_chinese_poetry.py           # 诗词数据下载处理脚本
├── download_jinyong_manual.py           # 金庸小说数据下载处理脚本
├── train_poetry.py                      # 诗词训练脚本
├── train_jinyong.py                     # 金庸小说训练脚本
└── test_models.py                       # 模型测试工具

---

三、数据集准备

本文档使用魔搭社区 ModelScope 提供的 2 个数据集进行测试，分别是诗词数据集、金庸小说数据集， 供仅测试参考。

3.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-2507

3.2 诗词数据处理文件代码

download_chinese_poetry.py将从 ModelScope 社区下载诗词数据集并进行处理。

核心逻辑: 直接从 ModelScope 网页下载 CSV 文件，转换为 JSON 格式

代码说明:

• 下载源: 使用 ModelScope API 直接下载训练集和测试集 CSV 文件
• 数据处理: 使用 pandas 读取 CSV，转换为字典列表格式
• 保存格式: JSON 文件，保留中文编码 (ensure_ascii=False)
• 容错处理: 如果网络下载失败，提供手动下载指引

"""简化版数据集下载脚本 - 直接从网页下载 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}")

3.3 金庸小说处理文件代码

download_jinyong_manual.py将从 ModelScope 社区下载金庸小说数据集并进行处理。

核心逻辑: 长文本切分 + 格式统一

代码说明:

• 数据源: 需要先手动下载 ModelScope 上的 .txt 小说文件
• 文本切分: 每段 4000 字，段落间 200 字重叠保持连贯性
• 过滤机制: 过滤长度小于 500 字的无效段落
• 格式输出: JSON 格式，包含 text 和 source（小说名）字段

"""金庸小说数据集处理脚本 - 手动下载版

说明:
由于 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: 过滤过短无效段落

3.4 诗词数据集下载

诗词数据集已经是标准格式，直接下载即可:

python download_chinese_poetry.py

输出:

• ./datasets/chinese-poetry/train.json (25000 条诗词)
• 格式: {"text1": "春眠不觉晓,处处闻啼鸟..."}

3.5 金庸小说数据集下载

金庸小说数据集需要特殊处理，因为原始数据是大段文本。

步骤 1: 下载小说文本

访问 https://modelscope.cn/datasets/josonfan/jinyong/files

手动下载以下文件到 ./datasets/jinyong/ 目录:

• 金庸-天龙八部.txt
• 金庸-侠客行.txt
• 金庸-笑傲江湖.txt
• 金庸-倚天屠龙记.txt
• 金庸-越女剑.txt

步骤 2: 处理成训练格式

python download_jinyong_manual.py

处理逻辑:

1. 读取每个 .txt 文件的完整内容
2. 切分成 4000 字左右的段落 (段落间有 200 字重叠保持连贯性)
3. 格式化为 JSON: {"text": "...", "source": "天龙八部"}
4. 保存为 ./datasets/jinyong/train_processed.json (978 段)

为什么要切分?

• 原始小说太长 (如天龙八部 180 万字)，模型无法一次处理
• 切分后可以从多个角度学习写作风格
• 适合模型的输入长度限制 (256-512 tokens)

四、训练流程

4.1 训练诗词模型

4.1.1 训练脚本（train_poetry.py）

"""Qwen3-4B-Instruct 诗词训练脚本 - LoRA 微调

功能:
1. 使用 Qwen3-4B-Instruct-2507 作为基础模型
2. 单独训练诗词 LoRA 权重
3. 使用干净的诗词数据集（无HTML编码、无特殊符号）

使用方法:
    python train_poetry.py

训练完成后,LoRA 权重会保存在 OUTPUT_DIR 目录
"""

import os
from dataclasses import dataclass, field
from typing import Optional

from modelscope import AutoModelForCausalLM, AutoTokenizer
from transformers import TrainingArguments, Trainer
from peft import LoraConfig, get_peft_model, TaskType
import torch

# ==================== 训练配置 ====================
# 模型配置
MODEL_PATH = "./models/Qwen3-4B-Instruct-2507"  # 基础模型路径

# 数据集配置
DATASET_NAME = "./datasets/chinese-poetry/train.json"  # 使用本地诗词数据集

# 输出配置
OUTPUT_DIR = "./output/qwen3-poetry"  # 训练输出目录
LOGGING_DIR = "./output/qwen3-poetry/logs"  # 日志目录

# 断点续训配置
RESUME_FROM_CHECKPOINT = False  # 第一次训练设为 False

# LoRA 参数
LORA_R = 64  # LoRA 秩
LORA_ALPHA = 128  # LoRA 缩放系数
LORA_DROPOUT = 0.05  # Dropout 比例

# 训练超参数（快速测试配置）
NUM_EPOCHS = 1  # 训练轮数（快速测试用1轮，正式训练可改为2-3轮）
BATCH_SIZE = 2  # 批次大小
GRADIENT_ACCUMULATION_STEPS = 8  # 梯度累积步数（有效batch_size = 2*8 = 16）
LEARNING_RATE = 1e-4  # 学习率
MAX_LENGTH = 256  # 最大序列长度（诗词较短）
WARMUP_RATIO = 0.1  # 预热比例
SAVE_STEPS = 100  # 每隔多少步保存一次
LOGGING_STEPS = 10  # 每隔多少步记录一次日志

# 数据集大小限制（快速测试）
MAX_SAMPLES = 20000  # 最多使用20K样本进行快速测试

# 其他配置
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(f"最大样本数: {MAX_SAMPLES}")
print("=" * 60)
print()


def load_dataset(dataset_name: str):
    """加载数据集"""
    print(f"[1/5] 正在加载数据集: {dataset_name}")

    try:
        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)

        # 快速测试: 限制样本数
        if len(dataset) > MAX_SAMPLES:
            print(f"⚠️  快速测试模式: 原数据 {len(dataset)} 条, 使用前 {MAX_SAMPLES} 条")
            dataset = dataset.select(range(MAX_SAMPLES))

        print(f"✓ 数据集加载成功,共 {len(dataset)} 条数据")
        return dataset
    except Exception as e:
        print(f"✗ 加载失败: {e}")
        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):
        # 诗词数据集使用 text1 字段
        texts = examples["text1"]

        # 分词
        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="预处理数据集",
    )

    print(f"✓ 数据预处理完成")
    return processed_dataset


def train(model, tokenizer, dataset):
    """训练模型"""
    print(f"\n[5/5] 开始训练")

    # 计算总步数
    total_steps = (
        len(dataset) // (BATCH_SIZE * GRADIENT_ACCUMULATION_STEPS)
    ) * NUM_EPOCHS
    print(f"  - 总步数: {total_steps}")
    print(f"  - 保存间隔: 每 {SAVE_STEPS} 步")

    # 训练参数
    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_steps=LOGGING_STEPS,
        save_steps=SAVE_STEPS,
        save_total_limit=3,  # 只保留最近3个检查点
        fp16=USE_FP16,
        logging_dir=LOGGING_DIR,
        report_to="none",  # 不上报到wandb等
        remove_unused_columns=False,
        dataloader_pin_memory=False,
    )

    # 创建 Trainer
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=dataset,
        tokenizer=tokenizer,
    )

    # 开始训练
    print("\n训练开始...")
    print("-" * 60)
    trainer.train(resume_from_checkpoint=RESUME_FROM_CHECKPOINT)

    # 保存最终模型
    final_dir = os.path.join(OUTPUT_DIR, "final")
    trainer.model.save_pretrained(final_dir)
    tokenizer.save_pretrained(final_dir)

    print("-" * 60)
    print(f"✓ 训练完成!")
    print(f"✓ 模型已保存到: {final_dir}")


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. 预处理数据集
        processed_dataset = preprocess_dataset(dataset, tokenizer)

        # 5. 训练
        train(model, tokenizer, processed_dataset)

        print("\n" + "=" * 60)
        print("所有步骤完成!")
        print("=" * 60)

    except Exception as e:
        print(f"\n✗ 训练失败: {e}")
        import traceback

        traceback.print_exc()


if __name__ == "__main__":
    main()

4.1.2 开始训练

# 切换到项目目录
D:
cd d:\wwwroot\modelscope

# 启动训练
python train_poetry.py

4.1.3 训练结果

• 训练时长：约 2.5 小时（20K 数据，1 轮）
• Loss 变化：0.73 → 0.50
• 总步数：1,250 步
• 保存位置：./output/qwen3-poetry/final/

4.2 训练金庸小说模型

4.2.1 训练脚本（train_jinyong.py）

"""Qwen3-4B-Instruct 金庸小说训练脚本 - LoRA 微调

功能:
1. 使用 Qwen3-4B-Instruct-2507 作为基础模型
2. 单独训练金庸武侠小说 LoRA 权重
3. 使用预处理好的金庸小说数据集

使用方法:
    python train_jinyong.py

训练完成后,LoRA 权重会保存在 OUTPUT_DIR 目录
"""

import os
from dataclasses import dataclass, field
from typing import Optional

from modelscope import AutoModelForCausalLM, AutoTokenizer
from transformers import TrainingArguments, Trainer
from peft import LoraConfig, get_peft_model, TaskType
import torch

# ==================== 训练配置 ====================
# 模型配置
MODEL_PATH = "./models/Qwen3-4B-Instruct-2507"  # 基础模型路径

# 数据集配置
DATASET_NAME = "./datasets/jinyong/train_processed.json"  # 使用预处理后的金庸小说数据

# 输出配置
OUTPUT_DIR = "./output/qwen3-jinyong"  # 训练输出目录
LOGGING_DIR = "./output/qwen3-jinyong/logs"  # 日志目录

# 断点续训配置
RESUME_FROM_CHECKPOINT = False  # 第一次训练设为 False

# LoRA 参数
LORA_R = 64  # LoRA 秩
LORA_ALPHA = 128  # LoRA 缩放系数
LORA_DROPOUT = 0.05  # Dropout 比例

# 训练超参数
NUM_EPOCHS = 3  # 训练轮数（金庸小说数据较少，用3轮）
BATCH_SIZE = 2  # 批次大小
GRADIENT_ACCUMULATION_STEPS = 8  # 梯度累积步数
LEARNING_RATE = 2e-4  # 学习率（比诗词稍高）
MAX_LENGTH = 512  # 最大序列长度（武侠小说句子较长）
WARMUP_RATIO = 0.1  # 预热比例
SAVE_STEPS = 50  # 每隔多少步保存一次（数据少，保存更频繁）
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:
        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)
        print(f"✓ 数据集加载成功,共 {len(dataset)} 条数据")
        return dataset
    except Exception as e:
        print(f"✗ 加载失败: {e}")
        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"],
        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):
        # 金庸小说数据集使用 text 字段
        texts = examples["text"]

        # 分词
        tokenized = tokenizer(
            texts,
            truncation=True,
            max_length=MAX_LENGTH,
            padding="max_length",
            return_tensors="pt",
        )

        # 设置 labels
        tokenized["labels"] = tokenized["input_ids"].clone()

        return tokenized

    # 批量处理
    processed_dataset = dataset.map(
        tokenize_function,
        batched=True,
        remove_columns=dataset.column_names,
        desc="预处理数据集",
    )

    print(f"✓ 数据预处理完成")
    return processed_dataset


def train(model, tokenizer, dataset):
    """训练模型"""
    print(f"\n[5/5] 开始训练")

    # 计算总步数
    total_steps = (
        len(dataset) // (BATCH_SIZE * GRADIENT_ACCUMULATION_STEPS)
    ) * NUM_EPOCHS
    print(f"  - 总步数: {total_steps}")
    print(f"  - 保存间隔: 每 {SAVE_STEPS} 步")

    # 训练参数
    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_steps=LOGGING_STEPS,
        save_steps=SAVE_STEPS,
        save_total_limit=3,
        fp16=USE_FP16,
        logging_dir=LOGGING_DIR,
        report_to="none",
        remove_unused_columns=False,
        dataloader_pin_memory=False,
    )

    # 创建 Trainer
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=dataset,
        tokenizer=tokenizer,
    )

    # 开始训练
    print("\n训练开始...")
    print("-" * 60)
    trainer.train(resume_from_checkpoint=RESUME_FROM_CHECKPOINT)

    # 保存最终模型
    final_dir = os.path.join(OUTPUT_DIR, "final")
    trainer.model.save_pretrained(final_dir)
    tokenizer.save_pretrained(final_dir)

    print("-" * 60)
    print(f"✓ 训练完成!")
    print(f"✓ 模型已保存到: {final_dir}")


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. 预处理数据集
        processed_dataset = preprocess_dataset(dataset, tokenizer)

        # 5. 训练
        train(model, tokenizer, processed_dataset)

        print("\n" + "=" * 60)
        print("所有步骤完成!")
        print("=" * 60)

    except Exception as e:
        print(f"\n✗ 训练失败: {e}")
        import traceback

        traceback.print_exc()


if __name__ == "__main__":
    main()

4.2.2 开始训练

python train_jinyong.py

4.2.3 训练结果

• 训练时长：约 1.5 小时（978 段，3 轮）
• 总步数：约 183 步
• 保存位置：./output/qwen3-jinyong/final/

---

五、模型测试

5.1 测试文件

使用 test_models.py 对 Qwen3-4B-Instruct-2507 模型多数据集 LoRA 微调进行测试。

核心功能：

1. ✅ 基础模型只加载一次（节省时间）
2. ✅ 快速切换 LoRA 权重
3. ✅ 流式生成输出（实时显示）
4. ✅ 针对性测试提示词

5.2 文件代码

"""模型测试脚本 - 对比诗词和金庸小说模型

功能:
1. 加载基础模型（只加载一次）
2. 快速切换不同的LoRA权重
3. 测试生成效果

使用方法:
    python test_models.py
"""

from modelscope import AutoModelForCausalLM, AutoTokenizer
from peft import PeftModel
import torch

# 模型配置
BASE_MODEL_PATH = "./models/Qwen3-4B-Instruct-2507"

# LoRA模型配置
LORA_MODELS = {
    "1": {
        "name": "诗词模型",
        "path": "./output/qwen3-poetry/final",
        "test_prompts": ["春江潮水连海平，", "明月几时有？", "大江东去，浪淘尽、"],
    },
    "2": {
        "name": "金庸小说模型",
        "path": "./output/qwen3-jinyong/final",
        "test_prompts": ["话说当年，乔峰", "少林寺中，", "华山之巅，剑气纵横，"],
    },
}

# 生成参数
MAX_NEW_TOKENS = 200
TEMPERATURE = 0.8
TOP_P = 0.9
TOP_K = 50
DO_SAMPLE = True
REPETITION_PENALTY = 1.2  # 重复惩罚

# 全局缓存
CACHED_BASE_MODEL = None
CACHED_TOKENIZER = None
CURRENT_MODEL = None
CURRENT_LORA_PATH = None


def load_base_model_once():
    """加载基础模型（只执行一次）"""
    global CACHED_BASE_MODEL, CACHED_TOKENIZER

    if CACHED_BASE_MODEL is not None:
        return CACHED_BASE_MODEL, CACHED_TOKENIZER

    print("正在加载基础模型...")
    print(f"  模型: {BASE_MODEL_PATH}")

    CACHED_TOKENIZER = AutoTokenizer.from_pretrained(
        BASE_MODEL_PATH, trust_remote_code=True
    )

    CACHED_BASE_MODEL = AutoModelForCausalLM.from_pretrained(
        BASE_MODEL_PATH,
        torch_dtype=torch.float16,
        device_map="auto",
        trust_remote_code=True,
    )

    print("✓ 基础模型加载完成\n")
    return CACHED_BASE_MODEL, CACHED_TOKENIZER


def switch_lora(lora_path):
    """切换LoRA权重"""
    global CURRENT_MODEL, CURRENT_LORA_PATH

    if CURRENT_LORA_PATH == lora_path and CURRENT_MODEL is not None:
        print(f"✓ 使用已加载的LoRA\n")
        return CURRENT_MODEL, CACHED_TOKENIZER

    base_model, tokenizer = load_base_model_once()

    print(f"正在切换LoRA权重...")
    print(f"  路径: {lora_path}")

    if CURRENT_MODEL is not None:
        CURRENT_MODEL = CURRENT_MODEL.unload()
        del CURRENT_MODEL
        torch.cuda.empty_cache()

    CURRENT_MODEL = PeftModel.from_pretrained(base_model, lora_path)
    CURRENT_LORA_PATH = lora_path

    print("✓ LoRA切换完成\n")
    return CURRENT_MODEL, tokenizer


def generate_text(model, tokenizer, prompt):
    """生成文本（流式输出）"""
    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
    input_length = inputs.input_ids.shape[1]

    print("生成内容: ", end="", flush=True)

    with torch.no_grad():
        for i in range(MAX_NEW_TOKENS):
            outputs = model(
                input_ids=inputs.input_ids,
                attention_mask=inputs.attention_mask,
            )

            next_token_logits = outputs.logits[:, -1, :]
            next_token_logits = next_token_logits / TEMPERATURE

            # 应用重复惩罚
            if REPETITION_PENALTY != 1.0:
                for token_id in set(inputs.input_ids[0].tolist()):
                    next_token_logits[0, token_id] /= REPETITION_PENALTY

            # Top-K 采样
            if TOP_K > 0:
                indices_to_remove = (
                    next_token_logits
                    < torch.topk(next_token_logits, TOP_K)[0][..., -1, None]
                )
                next_token_logits[indices_to_remove] = float("-inf")

            # Top-P 采样
            if TOP_P < 1.0:
                sorted_logits, sorted_indices = torch.sort(
                    next_token_logits, descending=True
                )
                cumulative_probs = torch.cumsum(
                    torch.softmax(sorted_logits, dim=-1), dim=-1
                )
                sorted_indices_to_remove = cumulative_probs > TOP_P
                sorted_indices_to_remove[..., 1:] = sorted_indices_to_remove[
                    ..., :-1
                ].clone()
                sorted_indices_to_remove[..., 0] = 0
                indices_to_remove = sorted_indices_to_remove.scatter(
                    1, sorted_indices, sorted_indices_to_remove
                )
                next_token_logits[indices_to_remove] = float("-inf")

            # 采样
            probs = torch.softmax(next_token_logits, dim=-1)
            if DO_SAMPLE:
                next_token = torch.multinomial(probs, num_samples=1)
            else:
                next_token = torch.argmax(probs, dim=-1, keepdim=True)

            if next_token.item() == tokenizer.eos_token_id:
                break

            # 流式输出当前token
            current_text = tokenizer.decode(
                [next_token.item()], skip_special_tokens=True
            )
            print(current_text, end="", flush=True)

            inputs.input_ids = torch.cat([inputs.input_ids, next_token], dim=-1)
            inputs.attention_mask = torch.cat(
                [
                    inputs.attention_mask,
                    torch.ones((1, 1), device=model.device, dtype=torch.long),
                ],
                dim=-1,
            )

    print()  # 换行
    full_output = tokenizer.decode(
        inputs.input_ids[0][input_length:], skip_special_tokens=True
    )
    return full_output


def test_model(model_key):
    """测试单个模型"""
    model_info = LORA_MODELS[model_key]
    model, tokenizer = switch_lora(model_info["path"])

    print("=" * 60)
    print(f"测试模型: {model_info['name']}")
    print("=" * 60)
    print()

    for i, prompt in enumerate(model_info["test_prompts"], 1):
        print(f"[{i}/{len(model_info['test_prompts'])}] 提示词: {prompt}")
        print("-" * 60)
        result = generate_text(model, tokenizer, prompt)
        print("=" * 60)
        print()


def main():
    """主函数"""
    print("=" * 60)
    print("模型测试工具")
    print("=" * 60)
    print()

    # 预加载基础模型
    load_base_model_once()

    while True:
        print("\n可用的模型：")
        print("-" * 60)
        for key, info in LORA_MODELS.items():
            print(f"[{key}] {info['name']}")
        print("[0] 退出")
        print("-" * 60)

        choice = input("\n请选择模型编号: ").strip()

        if choice == "0":
            print("退出程序")
            break

        if choice in LORA_MODELS:
            test_model(choice)
        else:
            print("⚠️  无效的选择\n")


if __name__ == "__main__":
    main()

5.3 测试效果

运行测试：

python test_models.py

诗词模型测试

提示词 1：春江潮水连海平，

生成内容: 鱼龙相与游戏横。风帆雨楫时出没，万里浮生一梦中。

提示词 2：明月几时有？

生成内容: 把酒问青天。不知天上宫阙，今夕是何年？我欲乘风归去，
又恐琼楼玉宇，高处不胜寒。起舞弄清影，似水绕花回。

提示词 3：大江东去，浪淘尽、

生成内容: 千古英雄。叹青山不老，流水无踪。问谁是、人间故旧？
但黄花篱下孤鸿。空回首，云飞风起，何处江枫。

金庸小说模型测试

提示词 1：话说当年，乔峰

生成内容: 是丐帮大寨主、契丹国萧太后之子，慕容复乃南朝辽国宗室之后。
我原想借你之力，将这两人除去，但今日一见之下，却真叫我生了极大的悔意。

提示词 2：少林寺中，

生成内容: 不识得几个字？这些书上写的都是什么？虚竹道："是教主亲笔所写。
在下不懂武艺武功，只懂得看书写字。"慕容复笑道："那就好了！"...

提示词 3：华山之巅，剑气纵横，

生成内容: 有好几柄长剑飞了出去。这时他和令狐冲都已受了伤，虽不至于一击即毙，
在这生死关头中，谁也不知会先死谁...

---

六、技术要点

6.1 为什么选择单独训练？

优势：

1. 专业能力强：每个模型专注单一领域，生成质量更高
2. 灵活切换：同一基础模型 + 不同 LoRA 权重 = 多种能力
3. 训练简单：避免数据配比问题，无需平衡不同领域数据量
4. 易于调优：可针对每个领域单独调整超参数

对比混合训练：

特性	单独训练	混合训练
专业度	⭐⭐⭐⭐⭐	⭐⭐⭐
灵活性	⭐⭐⭐⭐⭐	⭐⭐
训练复杂度	⭐⭐	⭐⭐⭐⭐
数据配比	无需考虑	需要精细调整

6.2 LoRA 参数说明

LORA_R = 64            # LoRA 秩，控制适配器大小
LORA_ALPHA = 128       # 缩放系数，通常为 r 的 2 倍
LORA_DROPOUT = 0.05    # Dropout 防止过拟合

参数选择建议：

• r=64, alpha=128：平衡性能和显存，推荐用于 4B 模型
• r=32, alpha=64：更省显存，适合大模型或显存紧张
• r=128, alpha=256：更强学习能力，适合复杂任务

6.3 重复惩罚机制

REPETITION_PENALTY = 1.2  # 重复惩罚系数

作用：防止生成重复文本（如"山外青山。山外青山。"）

实现原理：

# 对已出现的 token 降低概率
for token_id in set(inputs.input_ids[0].tolist()):
    next_token_logits[0, token_id] /= REPETITION_PENALTY

6.4 模型缓存与快速切换

# 全局缓存基础模型
CACHED_BASE_MODEL = None
CACHED_TOKENIZER = None

# 只加载一次基础模型
def load_base_model_once():
    if CACHED_BASE_MODEL is not None:
        return CACHED_BASE_MODEL, CACHED_TOKENIZER
    # ... 加载逻辑

# 快速切换 LoRA
def switch_lora(lora_path):
    base_model, tokenizer = load_base_model_once()
    CURRENT_MODEL = PeftModel.from_pretrained(base_model, lora_path)

性能提升：

• 首次加载：约 10 秒
• 切换 LoRA：约 2 秒
• 重选同模型：< 0.1 秒

---

七、常见问题

Q1: 训练时报错"页面文件太小"

原因：Windows 虚拟内存不足

解决方案：

1. 关闭所有占用内存的程序
2. 按照 1.2 节调整虚拟内存设置
3. 重启电脑后再训练

Q2: 生成文本有重复内容

解决方案：

1. 调高 REPETITION_PENALTY（如 1.3）
2. 提高 TEMPERATURE（如 0.9）
3. 调整 TOP_P 和 TOP_K 参数

Q3: 如何使用完整数据集训练？

修改训练脚本中的参数：

MAX_SAMPLES = None  # 或删除这行
NUM_EPOCHS = 2      # 增加训练轮数

注意：完整诗词数据集（38 万条）训练约需 50+ 小时

Q4: 如何继续断点训练？

RESUME_FROM_CHECKPOINT = True  # 改为 True

训练会自动从最近的检查点恢复（每 100 步保存一次）

Q5: 显存不足怎么办？

方案 1：减小 batch size

BATCH_SIZE = 1
GRADIENT_ACCUMULATION_STEPS = 16

方案 2：减小序列长度

MAX_LENGTH = 128  # 从 256 降到 128

---

八、总结

恭喜！到这里你已经掌握了基于魔搭社区 Qwen3-4B-Instruct-2507 进行多数据集独立 LoRA 微调训练的完整流程。

本教程内容回顾

我们从零开始，依次完成了以下关键步骤：

1. 环境准备：配置 Python 环境、虚拟内存及训练专用依赖
2. 项目结构：规划模型、数据集、输出目录的标准化组织方式
3. 数据集准备：下载并预处理诗词和金庸小说两个领域数据集
4. 训练流程：配置 LoRA 参数，分别对两个数据集进行独立训练
5. 模型测试：加载基础模型 + LoRA 权重进行效果验证
6. 技术要点：理解单独训练的优势、LoRA 参数配置、重复惩罚机制
7. 常见问题：处理页面文件不足、显存不足、生成重复等问题
8. 扩展应用：学习添加新数据集、部署 API 服务、模型量化等进阶技巧

学习掌握的能力

完成本教程后，一般来说将能够：

✅ 独立训练多领域模型：掌握一个基础模型 + 多个 LoRA 的训练方案
✅ 使用 ModelScope 平台：熟悉魔搭社区的模型下载和数据集管理方式
✅ 配置 LoRA 微调：理解 LoRA 秩、Alpha、Dropout 等核心参数的作用
✅ 排查常见错误：处理显存不足、训练中断、生成质量等问题
✅ 快速切换模型能力：实现基础模型缓存 + LoRA 权重热切换
✅ 开发专业生成应用：创建支持多领域切换、流式输出的 AI 工具
✅ 控制生成质量：通过重复惩罚、温度调节等技术优化输出效果
✅ 扩展新领域能力：掌握添加新数据集、训练新 LoRA 的完整流程

适用场景与应用方向

基于实际测试，单数据集独立训练方案在不同场景下的表现：

方案特点对比：

特性	单独训练	混合训练	使用建议
专业度	⭐⭐⭐⭐⭐	⭐⭐⭐	追求单领域最佳效果选单独训练
灵活性	⭐⭐⭐⭐⭐	⭐⭐	需要快速切换领域选单独训练
训练复杂度	⭐⭐	⭐⭐⭐⭐	避免数据配比问题选单独训练
存储占用	⭐⭐⭐	⭐⭐⭐⭐⭐	LoRA 权重较小，多个也不占太多空间

适用场景：

• 多领域文本生成（诗词、小说、公文等风格切换）
• 专业知识问答（医疗、法律、金融等垂直领域）
• 风格化内容创作（模仿特定作家、时代风格）

后续学习建议

如果想进一步提升，可以尝试：

• 增加训练数据：使用完整数据集（如 38 万条诗词）进行更充分的训练
• 探索更多数据集：在 ModelScope 上寻找其他领域数据集进行训练
• 学习 QLoRA 微调：使用量化技术进一步降低训练成本
• 部署为 API 服务：使用 FastAPI 封装推理服务，支持多模型切换
• 结合 RAG 技术：将微调模型与知识库检索结合，构建智能问答系统

感谢阅读

希望这篇教程能帮助你顺利掌握多数据集独立 LoRA 微调技术！如果遇到问题，欢迎反馈和交流。

祝你在 AI 领域不断探索，收获满满！ 🚀

---

版权所有

版权归属：参谋带个长

许可证：署名-非商业性-相同方式共享 4.0 国际 (CC-BY-NC-SA-4.0)