基于qwen2.5 手把手实战 自定义数据集 微调（llama-factory）
准备工作 1.数据集准备（例:民法典.txt） 2.服务器准备（阿里云 DSW 白嫖） 3.环境配置 pip 升级 模型下载 微调助手 4.数据集处理 脚本文件 4.1文本分割(bert-base-chinese) 4.2数据集生成 4.3.1数据集转换（只有一个数据集）alpaca格式 4.3.2训练集验证集划分（两个数据集）alpaca格式 5.开始微调 5.1配置文件修改 5.2进入微调ui 5.3修改参数即可训练 5.4训练结果 6.测试评估 7.模型对话 8.模型合并 9.模型量化准备工作
1.数据集准备（例:民法典.txt）
《中华人民共和国民法典 》全文免费下载 - 知乎 (zhihu.com)
自行转txt格式（约11万字）
2.服务器准备（阿里云 DSW 白嫖）
创建自己的实例（选择对应环境镜像约24g多）
https://free.aliyun.com/?spm=a2c4g.11174283.0.0.46a0527f6LNsEe&productCode=learn
3.环境配置
pip 升级
!pip install --upgrade pip
模型下载
git 下载
!git lfs install
!git clone https://www.modelscope.cn/Qwen/Qwen2.5-7B-Instruct.git
微调助手
llama-factory
!git clone https://github.com/hiyouga/LLaMA-Factory.git
%cd LLaMA-Factory
!pip install -e .
4.数据集处理
注：若缺失部分环境pip install 安装即可，此处不在强调
脚本文件
4.1文本分割(bert-base-chinese)
bert-base-chinese 下载到本地
git clone https://www.modelscope.cn/tiansz/bert-base-chinese.git
import torch
from transformers import BertTokenizer, BertModel
import re
import os
from scipy.spatial.distance import cosine
def get_sentence_embedding(sentence, model, tokenizer):
"""
获取句子的嵌入表示
参数:
sentence (str): 输入句子
model (BertModel): 预训练的BERT模型
tokenizer (BertTokenizer): BERT分词器
返回:
numpy.ndarray: 句子的嵌入向量
"""
# 使用分词器处理输入句子，并转换为模型输入格式
inputs = tokenizer(sentence, return_tensors="pt", padding=True, truncation=True, max_length=512)
# 使用模型获取输出，不计算梯度
with torch.no_grad():
outputs = model(**inputs)
# 返回最后一层隐藏状态的平均值作为句子嵌入
return outputs.last_hidden_state.mean(dim=1).squeeze().numpy()
def split_text_by_semantic(text, max_length, similarity_threshold=0.5):
"""
基于语义相似度对文本进行分块
参数:
text (str): 输入的长文本
max_length (int): 每个文本块的最大长度（以BERT分词器的token为单位）
similarity_threshold (float): 语义相似度阈值，默认为0.5
返回:
list: 分割后的文本块列表
"""
# 加载BERT模型和分词器
tokenizer = BertTokenizer.from_pretrained('/mnt/workspace/dataset/bert-base-chinese')
model = BertModel.from_pretrained('/mnt/workspace/dataset/bert-base-chinese')
model.eval() # 设置模型为评估模式
# 按句子分割文本（使用常见的中文标点符号）
sentences = re.split(r'(。|！|？|；)', text)
# 重新组合句子和标点
sentences = [s + p for s, p in zip(sentences[::2], sentences[1::2]) if s]
chunks = []
current_chunk = sentences[0]
# 获取当前chunk的嵌入表示
current_embedding = get_sentence_embedding(current_chunk, model, tokenizer)
for sentence in sentences[1:]:
# 获取当前句子的嵌入表示
sentence_embedding = get_sentence_embedding(sentence, model, tokenizer)
# 计算当前chunk和当前句子的余弦相似度
similarity = 1 - cosine(current_embedding, sentence_embedding)
# 如果相似度高于阈值且合并后不超过最大长度，则合并
if similarity > similarity_threshold and len(tokenizer.tokenize(current_chunk + sentence)) <= max_length:
current_chunk += sentence
# 更新当前chunk的嵌入表示
current_embedding = (current_embedding + sentence_embedding) / 2
else:
# 否则，保存当前chunk并开始新的chunk
chunks.append(current_chunk)
current_chunk = sentence
current_embedding = sentence_embedding
# 添加最后一个chunk
if current_chunk:
chunks.append(current_chunk)
return chunks
def read_text_file(file_path):
"""
读取文本文件
参数:
file_path (str): 文件路径
返回:
str: 文件内容
"""
with open(file_path, 'r', encoding='utf-8') as file:
return file.read()
def save_chunks_to_files(chunks, output_dir):
"""
将分割后的文本块保存到文件
参数:
chunks (list): 文本块列表
output_dir (str): 输出目录路径
"""
# 如果输出目录不存在，则创建
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# 将每个文本块保存为单独的文件
for i, chunk in enumerate(chunks):
chunk_file_path = os.path.join(output_dir, f"chunk_{i + 1}.txt")
with open(chunk_file_path, 'w', encoding='utf-8') as file:
file.write(chunk)
print(f"已保存第 {i + 1} 个文本块到 {chunk_file_path}")
# 主程序
# 设置输入和输出路径
input_file_path = './test.txt' # 替换为你的长文本文件路径
output_dir = './saveChunk/' # 替换为你希望保存文本块的目录路径
# 读取长文本
long_text = read_text_file(input_file_path)
# 设置每个文本块的最大分词数量和相似度阈值
max_length = 2048 # 可根据需要调整
similarity_threshold = 0.5 # 可根据需要调整
# 分割长文本
text_chunks = split_text_by_semantic(long_text, max_length, similarity_threshold)
# 保存分割后的文本块到指定目录
save_chunks_to_files(text_chunks, output_dir)
4.2数据集生成
如何获取API-KEY_大模型服务平台百炼(Model Studio)-阿里云帮助中心 (aliyun.com)
获取api 选择所需模型即可
model=“qwen-max-latest”, # 修改自己选择的 模型
import json
import os
import time
import re
from typing import List, Dict
from openai import OpenAI
import logging
import backoff
import pyarrow as pa
import pyarrow.parquet as pq
from concurrent.futures import ThreadPoolExecutor, as_completed
# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 从环境变量中获取 API 密钥
api_key = "？？？"
# 初始化 OpenAI 客户端
client = OpenAI(base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", api_key=api_key)
def read_file(file_path: str) -> str:
with open(file_path, 'r', encoding='utf-8') as file:
return file.read()
@backoff.on_exception(backoff.expo, Exception, max_tries=3)
def generate_single_entry(text: str) -> Dict:
prompt = f"""
基于以下文本，生成1个用于指令数据集的高质量条目。条目应该直接关联到给定的文本内容，提出相关的问题或任务。
请确保生成多样化的指令类型，例如：
- 分析类："分析..."
- 比较类："比较..."
- 解释类："解释..."
- 评价类："评价..."
- 问答类："为什么..."
文本内容：
{text}
请以下面的格式生成条目，确保所有字段都有适当的内容：
{{
"instruction": "使用上述多样化的指令类型之一，提出一个具体的、与文本相关的问题或任务",
"input": "如果需要额外的上下文信息，请在这里提供，否则留空",
"output": "对instruction的详细回答或任务的完成结果"
}}
确保所有生成的内容都与给定的文本直接相关，生成的是有效的JSON格式，并且内容高质量、准确、详细。
"""
try:
response = client.chat.completions.create(
model="qwen-plus-0919", # 尝试使用自己选择的 模型
messages=[{"role": "user", "content": prompt}],
temperature=0.7, # 增加温度以提高多样性
max_tokens=4098
)
logger.info(f"API 响应: {response.choices[0].message.content}")
json_match = re.search(r'\{.*\}', response.choices[0].message.content, re.DOTALL)
if json_match:
entry = json.loads(json_match.group())
required_keys = ['instruction', 'input', 'output']
if isinstance(entry, dict) and all(key in entry for key in required_keys):
# 根据 input 是否为空来设置 text 字段
if entry['input'].strip():
entry[
'text'] = f"Below is an instruction that describes a task, paired with an input that provides further context. Write a response that appropriately completes the request.### Instruction: {entry['instruction']}\n### Input: {entry['input']}\n### Response: {entry['output']}"
else:
entry[
'text'] = f"Below is an instruction that describes a task. Write a response that appropriately completes the request.### Instruction: {entry['instruction']}\n### Input: {entry['input']}\n### Response: {entry['output']}"
logger.info("成功生成完整条目")
return entry
else:
logger.warning("JSON 解析成功，但缺少必要字段")
return {}
else:
logger.error("无法从API响应中提取有效的JSON")
return {}
except Exception as e:
logger.error(f"生成条目时发生错误: {str(e)}")
return {}
def process_file(file_path: str, entries_per_file: int) -> List[Dict]:
dataset = []
try:
text = read_file(file_path)
for j in range(entries_per_file):
logger.info(f" 生成第 {j + 1}/{entries_per_file} 个条目")
entry = generate_single_entry(text)
if entry and all(key in entry for key in ['instruction', 'input', 'output', 'text']):
dataset.append(entry)
logger.info(f" 成功生成 1 个完整条目")
else:
logger.warning(f" 跳过不完整的条目")
time.sleep(2) # 在请求之间增加延迟到2秒
except Exception as e:
logger.error(f"处理文件 {file_path} 时发生未知异常: {str(e)}")
return dataset
def generate_dataset(folder_path: str, entries_per_file: int = 2) -> List[Dict]:
dataset = []
files = [os.path.join(folder_path, filename) for filename in os.listdir(folder_path) if filename.endswith(".txt")]
with ThreadPoolExecutor(max_workers=4) as executor: # 调整 max_workers 数量以适应你的硬件资源
futures = [executor.submit(process_file, file_path, entries_per_file) for file_path in files]
for future in as_completed(futures):
try:
dataset.extend(future.result())
except Exception as e:
logger.error(f"处理未来任务时发生未知异常: {str(e)}")
return dataset
def save_dataset_as_parquet(dataset: List[Dict], output_file: str):
schema = pa.schema([
('instruction', pa.string()),
('input', pa.string()),
('output', pa.string()),
('text', pa.string())
])
arrays = [
pa.array([entry['instruction'] for entry in dataset]),
pa.array([entry['input'] for entry in dataset]),
pa.array([entry['output'] for entry in dataset]),
pa.array([entry['text'] for entry in dataset])
]
table = pa.Table.from_arrays(arrays, schema=schema)
pq.write_table(table, output_file)
if __name__ == "__main__":
input_folder = "./saveChunk" # 指定输入文件夹路径
output_file = "instruction_dataset.parquet"
logger.info("开始生成数据集")
dataset = generate_dataset(input_folder, entries_per_file=10)
save_dataset_as_parquet(dataset, output_file)
logger.info(f"数据集已生成并保存到 {output_file}")
logger.info(f"共生成 {len(dataset)} 个有效条目")
4.3.1数据集转换（只有一个数据集）alpaca格式
import json
import pandas as pd
from rich import print
# Load local Parquet dataset
parquet_file_path = '/mnt/workspace/dataset/instruction_dataset.parquet'
dataset = pd.read_parquet(parquet_file_path)
# Print the columns of the dataset to verify
print("Dataset Columns:", dataset.columns)
# Convert dataset to list of dictionaries
json_data = dataset.to_dict(orient='list')
# Print the first row to verify
print("First Row:", {key: json_data[key][0] for key in json_data})
# Initialize list to store Alpaca format data
alpaca_data = []
# Process each entry in the dataset
for i in range(len(json_data['instruction'])):
instruction = json_data['instruction'][i]
input_text = json_data['input'][i]
original_output = json_data['output'][i]
# Create Alpaca format entry
alpaca_entry = {
"instruction": instruction,
"input": input_text,
"output": original_output
}
alpaca_data.append(alpaca_entry)
# Save in Alpaca format
with open("alpaca_dataset.json", "w") as file:
json.dump(alpaca_data, file, indent=4)
print("Alpaca format dataset saved as 'alpaca_dataset.json'")
4.3.2训练集验证集划分（两个数据集）alpaca格式
import json
import pandas as pd
from rich import print
from sklearn.model_selection import train_test_split
# Load local Parquet dataset
parquet_file_path = '/mnt/workspace/dataset/instruction_dataset.parquet'
dataset = pd.read_parquet(parquet_file_path)
# Print the columns of the dataset to verify
print("Dataset Columns:", dataset.columns)
# Convert dataset to list of dictionaries
json_data = dataset.to_dict(orient='list')
# Print the first row to verify
print("First Row:", {key: json_data[key][0] for key in json_data})
# Initialize lists to store Alpaca format data
alpaca_data = []
# Process each entry in the dataset
for i in range(len(json_data['instruction'])):
instruction = json_data['instruction'][i]
input_text = json_data['input'][i]
original_output = json_data['output'][i]
# Create Alpaca format entry
alpaca_entry = {
"instruction": instruction,
"input": input_text,
"output": original_output
}
alpaca_data.append(alpaca_entry)
# Split the dataset into training and testing sets (80% training, 20% testing)
train_data, test_data = train_test_split(alpaca_data, test_size=0.2, random_state=42)
# Save training set in Alpaca format
with open("alpaca_train_dataset.json", "w") as file:
json.dump(train_data, file, indent=4)
# Save testing set in Alpaca format
with open("alpaca_test_dataset.json", "w") as file:
json.dump(test_data, file, indent=4)
print("Alpaca format training dataset saved as 'alpaca_train_dataset.json'")
print("Alpaca format testing dataset saved as 'alpaca_test_dataset.json'")
5.开始微调
5.1配置文件修改
LLaMA-Factory/data/dataset_info.json
LLaMA-Factory/data/dataset_info.json
同时alpaca_train_dataset.json和alpaca_test_dataset.json放入LLaMA-Factory/data/路径
"train": {
"file_name": "alpaca_train_dataset.json"
},
"test": {
"file_name": "alpaca_test_dataset.json"
},
5.2进入微调ui
需要进入对应文件夹下
%cd LLaMA-Factory
!llamafactory-cli webui
注释：若出现连接失败，尝试配置环境变量
!GRADIO_ROOT_PATH=/${JUPYTER_NAME}/proxy/7860/ \
USE_MODELSCOPE_HUB=1 \
llamafactory-cli webui
5.3修改参数即可训练
模型路径使用绝对路径即可
/mnt/workspace/Qwen2.5-7B-Instruct
注：具体调整参数需自己测试
5.4训练结果
6.测试评估
注：可能需要安装(出现错误根据提示进行修改即可)
pip install rouge-chinese
7.模型对话
chat模式加载模型
若显存不够则使用huggingface推理（vllm加载速度更快）
8.模型合并
合并不能进行量化操作
9.模型量化
需合并了再量化选择所需量化等级即可（24g显存可能不够）
总结
### 文章总结：基于qwen2.5 自定义数据集微调实战（llama-factory）#### 准备工作
1. **数据集准备**：
- 示例：下载《中华人民共和国民法典》并转换为txt格式（约11万字）。
2. **服务器准备**：
- 使用阿里云DSW（数据科学工作坊）创建实例，选择适当的环境镜像（约24GB）。
3. **环境配置**：
- 升级pip。
- 下载模型（如Qwen2.5-7B-Instruct）和微调助手（llama-factory）。
#### 数据集处理
1. **文本分割**：
- 使用BERT-base-chinese模型基于语义相似度将长文本分割成多个文本块。
- 编写脚本实现文本读取、分割和保存。
2. **数据集生成**：
- 利用OpenAI API（通过阿里云百炼平台）生成指令数据集。
- 编写脚本读取文本文件，生成包含指令、输入和输出的JSON条目，并保存为Parquet格式。
3. **数据集转换与划分**：
- 将Parquet数据集转换为Alpaca格式。
- 划分数据集为训练集和验证集（80%训练，20%测试），并保存为JSON文件。
#### 开始微调
1. **配置文件修改**：
- 修改`dataset_info.json`文件，指定训练集和验证集的JSON文件名。
2. **进入微调UI**：
- 使用`llamafactory-cli webui`命令启动微调界面。
- 如有需要，配置环境变量以解决连接问题。
3. **修改参数并训练**：
- 在微调UI中设置模型路径和其他训练参数。
- 开始训练并观察训练结果。
#### 测试评估
- 使用ROUGE-Chinese等工具对模型进行评估。
#### 模型对话
- 加载训练好的模型进行对话测试。
- 若显存不足，可使用Hugging Face进行推理。
#### 模型合并与量化
- 合并多个模型（注意合并后不能进行量化）。
- 对合并后的模型进行量化，选择适当的量化等级（注意显存限制）。
### 注意事项
- 文中提到的脚本和命令需根据实际情况调整路径和参数。
- 在执行过程中，根据错误提示安装缺失的库或调整配置。
- 显存和计算资源是微调大模型时的重要考虑因素。