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

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

{
    "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。"
}

npu-smi info

# 下载 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-smi

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

{%- 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 %}

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""",
}

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

...   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,   )   ...

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()

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 数量>

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 思维链微调实战：多卡 LoRA 完整代码示例

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

思维链技术介绍

Qwen2.5 思维链微调实操

CoT 数据集准备

更多推荐文章

相关免费在线工具

环境安装

昇腾 NPU + openMind Library 环境安装

驱动安装&验证

openMind 环境搭建

Nvidia GPU + Transformers 环境安装

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

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

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

微调效果（附上 Gradio 代码）

总结与最佳实践

更多推荐文章

相关免费在线工具

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

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

思维链技术介绍

Qwen2.5 思维链微调实操

CoT 数据集准备

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具

环境安装

昇腾 NPU + openMind Library 环境安装

驱动安装&验证

openMind 环境搭建

Nvidia GPU + Transformers 环境安装

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

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

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

微调效果（附上 Gradio 代码）

总结与最佳实践

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具