Qwen2.5 思维链微调实战：多卡 LoRA 完整代码示例

本文使用的是早年收集的一个 CoT 英文数据集，从质量判断是利用 GPT4+ 提示词生成的。现在网上已经有许多高质量的 CoT 数据集，包括一些数据集专门区分了思考中间过程与返回结果，以及一些将思维步骤使用 Special Token 进行分隔的。

本文考虑到仅仅是练手项目，选用了最基础的数据集，并且使用经过指令微调的 Qwen2.5-7B 将英文数据集进一步翻译成了中文 + 整理成了标准的 Instruction-Input-Output 样式，以方便对 Qwen 模型的微调和练手学习。注意这个数据集仅仅是练手，LLM 翻译的质量并不高，很容易被要翻译的内容带偏了。下图展示了部分数据集和样式：

{
    "instruction": "这家宠物店一共有多少只啮齿动物？宠物店里有 6 个笼子装着啮齿动物。其中 3 个笼子里各有 10 只仓鼠，另外 3 个笼子里各有 5 只豚鼠。那么这家宠物店一共养了多少只啮齿动物呢？",
    "input": "",
    "output": "让我们一步步来思考。\n\n有 3 个笼子，每个笼子里有 10 只豚鼠。\n这意味着总共有 30 只豚鼠。\n\n还有 3 个笼子，每个笼子里有 5 只豚兔。\n这意味着总共有 15 只豚兔。\n\n现在我们可以把 30 和 15 加起来得到答案：\n\n总共有 45 只啮齿动物。\n因此，答案（阿拉伯数字）是 45。"
}

请自行准备 CoT 训练数据集，确保格式符合上述 JSONL 标准。

环境安装

昇腾 NPU + openMind Library 环境安装

国产卡是未来，这里先放昇腾 NPU 环境安装！

使用昇腾 NPU 的话推荐在魔乐社区中找模型，里面能找到完成 NPU 适配的模型。魔乐社区使用的是 openMind Library 工具包，这个包支持在 Nvidia GPU 和 Ascend NPU 上运行，使用起来和 transformers 接口一致。如果说做昇腾 NPU 迁移的话非常推荐使用。

魔乐社区的模型分为 MindSpore 支持和 Pytorch-NPU 支持，这里主要看本地装什么环境，考虑到新手学习的话推荐使用 Pytorch-NPU，和 Pytorch 逻辑基本一致。

驱动安装&验证

首先得确定有 NPU 卡和 NPU 相关驱动，驱动是 8.0.RC3.beta1，具体可以参考软件安装-CANN 商用版 8.0.RC3 开发文档 - 昇腾社区。

安装好后的验证方法是运行下面的命令，该命令作用与 nvidia-smi 类似，这里是查看 NPU 的状态和性能：

npu-smi info

可以看到如下信息的话就表示驱动已经安装完成了。

openMind 环境搭建

openMind 环境安装比较简单，这边列出所需用到的全部安装命令：

# 下载 PyTorch 安装包
wget https://download.pytorch.org/whl/cpu/torch-2.4.0-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl
# 下载 torch_npu 插件包
wget https://gitee.com/ascend/pytorch/releases/download/v6.0.rc3-pytorch2.4.0/torch_npu-2.4.0-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl
# 安装命令
pip3 install torch-2.4.0-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl
pip3 install torch_npu-2.4.0-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl
# 安装 openMind Library
pip install openmind[pt]
pip install transformers accelerate datasets peft
# 部分场景会用到 hf 几个包，干脆全装了
# 安装 SwanLab
pip install swanlab

Nvidia GPU + Transformers 环境安装

这个流程比较简单，首先也是得确保 Nvidia 驱动存在，验证命令：

nvidia-smi

如果没显示同样需要先安装 cuda 环境，这里贴上 CUDA 官方安装链接。网上有大量 cuda 安装教程，这里笔者就不赘述了。同样放出 transformers 环境安装的全部命令：

pip install torch
pip install transformers accelerate datasets peft
# 安装 SwanLab
pip install swanlab

关于提示词模版构建（大坑）

这里需要强调一下，在使用 Qwen2.5 的 Instruct 模型微调时，为了保障效果建议严格按照模型自身的 Instruct 的提示词模版构建。HF Transformers 在 4.3 几的版本开始支持 Chat Templates。Qwen2.5 关于 Instruct 和 Chat 的提示词模版被直接写到了 tokenizers 的设置保存中，这导致了很多人在原始代码中找不到 instruct 提示词格式的构造。很多教程在教微调的时候还用的是 Qwen1 的老提示词模版或者自己构建的提示词模版，这会严重影响使用已经微调的模型做进一步微调时的效果。建议针对模型微调时一定要仔细检查提示词模版的实现部分。尽量使用模型已经定义好的格式和结构。

可以在 Qwen 的 HF 项目中找到提示词模版，点击 HF Qwen 查看 chat_template 设置。chat_template 默认使用的是一种前端模版语言 jinja，并不好看懂，笔者把 qwen2.5 的提示词模版格式化后粘贴在下文：

{%- if tools %}       {{- '<|im_start|>system\n' }}       {%- if messages[0]['role'] == 'system' %}           {{- messages[0]['content'] }}       {%- else %}           {{- 'You are Qwen, created by Alibaba Cloud. You are a helpful assistant.' }}       {%- endif %}       {{- '\n\n# Tools\n\nYou may call one or more functions to assist with the user query.\n\nYou are provided with function signatures within <tools></tools> XML tags:\n<tools>' }}       {%- for tool in tools %}           {{- '\n' }}           {{- tool | tojson }}       {%- endfor %}       {{- '\n</tools>\n\nFor each function call, return a json object with function name and arguments within <tool_call></tool_call> XML tags:\n<tool_call>\n{"name": <function-name>, "arguments": <args-json-object>}\n</tool_call><|im_end|>\n' }}   {%- else %}       {%- if messages[0]['role'] == 'system' %}           {{- '<|im_start|>system\n' + messages[0]['content'] + '<|im_end|>\n' }}       {%- else %}           {{- '<|im_start|>system\nYou are Qwen, created by Alibaba Cloud. You are a helpful assistant.<|im_end|>\n' }}       {%- endif %}   {%- endif %}   {%- for message in messages %}       {%- if (message.role == "user") or (message.role == "system" and not loop.first) or (message.role == "assistant" and not message.tool_calls) %}           {{- '<|im_start|>' + message.role + '\n' + message.content + '<|im_end|>' + '\n' }}       {%- elif message.role == "assistant" %}           {{- '<|im_start|>' + message.role }}           {%- if message.content %}               {{- '\n' + message.content }}           {%- endif %}           {%- for tool_call in message.tool_calls %}               {%- if tool_call.function is defined %}                   {%- set tool_call = tool_call.function %}               {%- endif %}               {{- '\n<tool_call>\n{"name": "' }}               {{- tool_call.name }}               {{- '", "arguments": ' }}               {{- tool_call.arguments | tojson }}               {{- '}\n</tool_call>' }}           {%- endfor %}           {{- '<|im_end|>\n' }}       {%- elif message.role == "tool" %}           {%- if (loop.index0 == 0) or (messages[loop.index0 - 1].role != "tool") %}               {{- '<|im_start|>user' }}           {%- endif %}           {{- '\nresult\n' }}           {{- message.content }}           {{- '\nresult' }}           {%- if loop.last or (messages[loop.index0 + 1].role != "tool") %}               {{- '<|im_end|>\n' }}           {%- endif %}       {%- endif %}   {%- endfor %}   {%- if add_generation_prompt %}       {{- '<|im_start|>assistant\n' }}   {%- endif %}

可以看到超级长，因为定义了好几种情况，包括是否有 system prompt。以及针对 function tools 怎么处理等等等等。如果读不懂（我感觉大多数搞 deep learning 的除了做 LLM Finetune 也很小有机会去学一个前端语言）我建议用大模型给你逐行解释下，这里附上 jinja 的官方文档。

这里笔者简单提供我所使用的 Qwen2.5 简化版 python 模版（下脚本），去除了 Function Calling 和多轮对话的部分。并且只包含对 Instruct 和 Inputs 的处理部分，以及 Assistants 的生成头。这分为带 inputs 的版本和不带 inputs 的版本。我自己专门测试了使用此模版构造的提示词长度上和使用 Qwen 带 chat_template 的 tokenizers 完全一致。你只需要将 outputs 部分增加一个 \n<|im_end|>\n 即可直接拼接成 finetune LLM 模型的 targets 部分。

PROMPT_DICT = {
    "prompt_no_input": """<|im_start|>system\n{instruction}<|im_end|>\n<|im_start|>user\n<|im_end|>\n<|im_start|>assistant\n""",
    "prompt_input": """<|im_start|>system\n{instruction}<|im_end|>\n<|im_start|>user\n{input}<|im_end|>\n<|im_start|>assistant\n""",
}

如果你直接偷懒使用 chat_template 来 tokenizer 仅带 outputs 部分的数据。你会发现由于 Qwen 的 chat template 处理机制，实际上生成的 outputs 部分会默认带上 system prompts。导致最后训练阶段会出现奇怪的内容。Qwen 的 tokenizers 针对未增加 system 角色的对话输入会自动加上如下提示词：

system:You are Qwen, created by Alibaba Cloud. You are a helpful assistant.

更神奇的是，这个 system prompt 居然是个英文的。Qwen 可是个中文模型。。。这个 system prompt 的出现会影响后续的模型微调效果。

可视化工具配置 (SwanLab 使用教程)

SwanLab 可以将微调的许多关键参数自动记录下来并且能够在线或者离线保存 + 查看训练日志。SwanLab 同时支持记录 NVIDIA GPU 和华为昇腾 NPU 设备的日志记录工具。最新版本已经支持对 NPU 的内存使用、功率、温度等进行记录。

关于 SwanLab 的使用方法可以参考 SwanLab 官方文档 - 快速开始。

对于 Huggingface Transformers 或者支持华为昇腾 NPU 的 openMind Library，可以使用 SwanLab Integration 轻松完成实验数据记录：

...   from swanlab.integration.huggingface import SwanLabCallback   swanlab_call = SwanLabCallback( #       "Ascend_finetune_v2",       experiment_name=os.path.basename(os.path.normpath(training_args.output_dir)),       config=asdict(data_args)       | asdict(model_args)       | asdict(training_args)       | asdict(lora_config),       public=True,   )   trainer = openmind.Trainer( # 使用 hf transformers 的话则是把 openmind 替换为 transformers       model=model,       tokenizer=tokenizer,       args=training_args,       callbacks=[swanlab_call],   # callback 加入进去即可       **data_module,   )   ...

使用后不仅能进行多图表对比，更重要的是把一大堆的 huggingface transformers 的训练超参数全部记录下来了，简直调参党福音。

微调代码（多卡，支持华为 Ascend 卡）

下面附上完整的微调代码。在项目目录下创建 finetune.py 文件，并将如下代码粘贴进文件中：

import copy
import os
import io
import json
import logging
from dataclasses import dataclass, field, asdict
from typing import Dict, Optional, Sequence

import torch
from torch.utils.data import Dataset
try:
    import openmind as tf_module
except:
    import transformers as tf_module
import transformers

from peft import LoraConfig, get_peft_model
from swanlab.integration.huggingface import SwanLabCallback

IGNORE_INDEX = -100

PROMPT_DICT = {
    "prompt_no_input": """<|im_start|>system\n{instruction}<|im_end|>\n<|im_start|>user\n<|im_end|>\n<|im_start|>assistant\n""",
    "prompt_input": """<|im_start|>system\n{instruction}<|im_end|>\n<|im_start|>user\n{input}<|im_end|>\n<|im_start|>assistant\n""",
}

@dataclass
class ModelArguments:
    model_name_or_path: Optional[str] = field(
        default="./weights/Qwen/Qwen2.5-7B-Instruct"
    )

@dataclass
class DataArguments:
    data_path: str = field(
        default="./data/cot_train_cn.jsonl",
        metadata={"help": "Path to the training data."},
    )

@dataclass
class TrainingArguments(transformers.TrainingArguments):
    cache_dir: Optional[str] = field(default=None)
    optim: str = field(default="adamw_torch")
    model_max_length: int = field(
        default=512,
        metadata={
            "help": "Maximum sequence length. Sequences will be right padded (and possibly truncated)."
        },
    )

    def _tokenize_fn(strings: Sequence[str], tokenizer) -> Dict:
        """Tokenize a list of strings."""
        tokenized_list = [
            tokenizer(
                text,
                return_tensors="pt",
                padding="longest",
                max_length=tokenizer.model_max_length,
                truncation=True,
            )
            for text in strings
        ]
        input_ids = labels = [tokenized.input_ids[0] for tokenized in tokenized_list]
        input_ids_lens = labels_lens = [
            tokenized.input_ids.ne(tokenizer.pad_token_id).sum().item()
            for tokenized in tokenized_list
        ]
        return dict(
            input_ids=input_ids,
            labels=labels,
            input_ids_lens=input_ids_lens,
            labels_lens=labels_lens,
        )

def jload(f, mode="r", jsonl=True):
    if not isinstance(f, io.IOBase):
        with open(f, mode=mode, encoding="utf-8") as f:
            if jsonl:
                # Parse JSON Lines
                return [json.loads(line) for line in f if line.strip()]
            else:
                # Parse standard JSON
                return json.load(f)
    else:
        if jsonl:
            return [json.loads(line) for line in f if line.strip()]
        else:
            return json.load(f)

def preprocess(
    sources: Sequence[str],
    targets: Sequence[str],
    tokenizer,
) -> Dict:
    """Preprocess the data by tokenizing."""
    examples = [s + t for s, t in zip(sources, targets)]
    examples_tokenized, sources_tokenized = [
        _tokenize_fn(strings, tokenizer) for strings in (examples, sources)
    ]
    input_ids = examples_tokenized["input_ids"]
    labels = copy.deepcopy(input_ids)
    for label, source_len in zip(labels, sources_tokenized["input_ids_lens"]):
        label[:source_len] = IGNORE_INDEX
    return dict(input_ids=input_ids, labels=labels)

class SupervisedDataset(Dataset):
    """Dataset for supervised fine-tuning."""

    def __init__(self, data_path: str, tokenizer):
        super(SupervisedDataset, self).__init__()
        logging.warning("Loading data...")
        list_data_dict = jload(data_path)

        logging.warning("Formatting inputs...")
        prompt_input, prompt_no_input = (
            PROMPT_DICT["prompt_input"],
            PROMPT_DICT["prompt_no_input"],
        )
        sources = [
            (
                prompt_input.format_map(example)
                if example.get("input", "") != ""
                else prompt_no_input.format_map(example)
            )
            for example in list_data_dict
        ]
        targets = [
            f"{example['output']}\n{tokenizer.eos_token}\n"
            for example in list_data_dict
        ]

        logging.warning("Tokenizing inputs... This may take some time...")
        data_dict = preprocess(sources, targets, tokenizer)

        try:
            self.input_ids = data_dict["input_ids"]
        except KeyError as e:
            raise KeyError("input_ids is invalid") from e
        try:
            self.labels = data_dict["labels"]
        except KeyError as e:
            raise KeyError("labels is invalid") from e

    def __len__(self):
        return len(self.input_ids)

    def __getitem__(self, i) -> Dict[str, torch.Tensor]:
        return dict(input_ids=self.input_ids[i], labels=self.labels[i])

@dataclass
class DataCollatorForSupervisedDataset(object):
    """Collate examples for supervised fine-tuning."""

    tokenizer: object

    def __call__(self, instances: Sequence[Dict]) -> Dict[str, torch.Tensor]:
        input_ids, labels = tuple(
            [instance[key] for instance in instances] for key in ("input_ids", "labels")
        )
        input_ids = torch.nn.utils.rnn.pad_sequence(
            input_ids, batch_first=True, padding_value=self.tokenizer.pad_token_id
        )
        labels = torch.nn.utils.rnn.pad_sequence(
            labels, batch_first=True, padding_value=IGNORE_INDEX
        )
        return dict(
            input_ids=input_ids,
            labels=labels,
            attention_mask=input_ids.ne(self.tokenizer.pad_token_id),
        )

def make_supervised_data_module(tokenizer, data_args) -> Dict:
    """Make dataset and collator for supervised fine-tuning."""
    train_dataset = SupervisedDataset(
        tokenizer=tokenizer, data_path=data_args.data_path
    )
    data_collator = DataCollatorForSupervisedDataset(tokenizer=tokenizer)
    return dict(
        train_dataset=train_dataset, eval_dataset=None, data_collator=data_collator
    )

def train():
    parser = transformers.HfArgumentParser(
        (ModelArguments, DataArguments, TrainingArguments)
    )
    model_args, data_args, training_args = parser.parse_args_into_dataclasses()

    model = tf_module.AutoModelForCausalLM.from_pretrained(
        model_args.model_name_or_path,
        cache_dir=training_args.cache_dir,
        trust_remote_code=True,
    )

    # 定义 LoRA 配置
    lora_config = LoraConfig(
        r=16,
        lora_alpha=16,
        target_modules=["q_proj", "v_proj"],
        lora_dropout=0.1,
        bias="none",
    )
    model = get_peft_model(model, lora_config)
    model.print_trainable_parameters()

    tokenizer = tf_module.AutoTokenizer.from_pretrained(
        model_args.model_name_or_path,
        cache_dir=training_args.cache_dir,
        model_max_length=training_args.model_max_length,
        padding_side="right",
        use_fast=False,
        trust_remote_code=True,
    )

    data_module = make_supervised_data_module(tokenizer=tokenizer, data_args=data_args)

    swanlab_call = SwanLabCallback(
        "Ascend_finetune_v2",
        experiment_name=os.path.basename(os.path.normpath(training_args.output_dir)),
        config=asdict(data_args)
        | asdict(model_args)
        | asdict(training_args)
        | asdict(lora_config),
        public=True,
    )

    trainer = tf_module.Trainer(
        model=model,
        tokenizer=tokenizer,
        args=training_args,
        callbacks=[swanlab_call],
        **data_module,
    )
    trainer.train()
    trainer.save_state()
    trainer.save_model(output_dir=training_args.output_dir)

if __name__ == "__main__":
    train()

多卡训练的话可以使用 torchrun，这里附上一个启动多卡的 bash 脚本，在当前目录下创建 finetune.sh，并且粘贴如下脚本：

NPU_NUM=${1:-8}
EXP_NAME=$(basename "$0" .sh)
if [ -d ./output ];then
    rm -rf ./output/$EXP_NAME
    mkdir -p ./output/$EXP_NAME
else
    mkdir -p ./output/$EXP_NAME
fi

# master_port 参数需用户根据实际情况进行配置
torchrun --nproc_per_node=$NPU_NUM --master_port=20248 finetune.py \
    --model_name_or_path "./weights/Qwen/Qwen2.5-7B-Instruct" \
    --data_path data/cot_train_cn.jsonl \
    --bf16 True \
    --output_dir ./output/$EXP_NAME \
    --max_steps 2000 \
    --per_device_train_batch_size 2 \
    --eval_strategy "no" \
    --save_strategy "steps" \
    --save_steps 3000 \
    --save_total_limit 1 \
    --learning_rate 2e-5 \
    --weight_decay 0. \
    --warmup_ratio 0.03 \
    --lr_scheduler_type "cosine" \
    --seed 42 \
    --logging_steps 10

开启多卡训练的方式如下：

bash finetune.sh <使用的 GPU/NPU 数量>

如果提示登录 swanlab，可以在官网完成注册后，使用获取 API KEY 找到对应的登陆密钥并粘贴，这样将能够使用云上看版随时查看训练过程与结果。

微调效果（附上 Gradio 代码）

本来准备了 Ceval 的测试结果，结果不知道为什么 Ascend 服务器连不上了，等过段时间更新下教程文档。

这里放出使用 CoT 数据微调 qwen-7b-instruct、qwen-0.5b-instruct 和使用 qwen-7b-instruct（8NPU）的 loss 结果。可以看到使用 8 个 NPU 能带来更好的训练 loss 表现和稳定性，哪怕在使用同样迭代数据量的情况下，8 个 NPU 依然能带来更好的 loss 结果。可能更大的 batch size 有助于模型稳定下降。

最后展现下使用 gradio 完成的官方 Qwen2.5-7B-Instruct、基于 Qwen2.5-7B 在中文 alpaca 数据集上指令微调、以及 cot 微调后的模型回复对比。可以看到 CoT 微调后模型确实具备了'step by step'的回复模式。

当然许多读者注意到了官方模型也展现出了'step by step'的回答模式，这主要是因为现在较新的模型在 finetune 数据集甚至 pretrain 数据集中就会预先加入 CoT 数据，所以模型在进行问答、尤其是数学题问答时，会展现出'步骤分解'的现象。笔者后续会尝试在较早期的 demo 中更新微调后的模型表现，并进一步验证其在复杂推理任务上的泛化能力。

附上启用 gradio 的 demo 测试代码：

使用 pip install gradio 安装依赖包

import gradio as gr

from openmind import AutoModelForCausalLM, pipeline
from peft import PeftModel

TOTAL_GPU_NUMS = 8
TOKENIZE_PATH = "~/weights/Qwen/Qwen2.5-7B-Instruct"
MODEL_LIST = {
    "office_qwen7b": "~/weights/Qwen/Qwen2.5-7B-Instruct",  # 官方模型
    "alpaca_qwen7b_lora": "./projects/qwen_finietune_cot/output/qwen25-7B-alpaca",  # 7b+alpaca
    "cot_qwen7b_lora": "./projects/qwen_finietune_cot/output/qwen25-7Bi-cot",  # cot 微调
}

model_names = MODEL_LIST.keys()
pipes = dict()
for i, model_name in enumerate(model_names):
    save_path = MODEL_LIST[model_name]
    model = AutoModelForCausalLM.from_pretrained(save_path)
    if model_name[:-5] == "_lora":
        model = PeftModel.from_pretrained(model, save_path)
    pipe = pipeline(
        "text-generation",
        model=model,
        tokenizer=TOKENIZE_PATH,
        framework="pt",
        device=f"npu:{i%TOTAL_GPU_NUMS}",
    )
    pipes[model_name] = pipe

def generate_response(instruct_text, input_text):
    messages = [
        {
            "role": "system",
            "content": instruct_text,
        },
        {
            "role": "user",
            "content": input_text,
        },
    ]
    outputs = [
        pipes[model_name](messages, max_new_tokens=256)[-1]["content"]
        for model_name in model_names
    ]
    return tuple(outputs)

# 创建 Gradio 界面
demo = gr.Interface(
    fn=generate_response,  # 函数名
    inputs=[
        gr.Textbox(label="instruction"),
        gr.Textbox(label="input"),
    ],  # 输入文本框
    outputs=[gr.Textbox(label=model_name) for model_name in model_names],
)

if __name__ == "__main__":
    demo.launch()

总结与最佳实践

通过本教程的实践，我们成功实现了 Qwen2.5 模型的思维链（CoT）微调。以下是几点关键经验总结：

1. 提示词模板至关重要：务必使用模型官方提供的 chat_template，避免自定义模板导致的 System Prompt 缺失或格式错误，这直接影响推理效果。
2. 数据质量决定上限：虽然使用了自动化翻译的数据集进行练手，但在生产环境中，建议使用人工审核的高质量 CoT 数据，特别是针对特定垂直领域的推理任务。
3. 多卡并行加速训练：使用 torchrun 配合 LoRA 技术可以显著降低显存占用并提升训练速度，特别是在处理长序列 CoT 数据时，多卡并行显得尤为重要。
4. 监控与可视化：集成 SwanLab 等工具可以有效追踪训练过程中的 Loss 变化及硬件资源使用情况，便于及时调整超参数。

未来，随着大模型推理能力的进一步提升，CoT 微调将成为提升模型逻辑推理能力的重要手段。开发者应关注数据构建的多样性，并结合实际业务场景进行针对性优化。