使用Qwen2.5 微调私有数据

1
2
3
4
5
6

<|im_start|>system
You are a helpful assistant.<|im_end|>
<|im_start|>user
你好，请介绍一下你自己。<|im_end|>
<|im_start|>assistant
你好，我是通义千问...<|im_end|>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig

# 定义模型 ID
model_id = "Qwen/Qwen2.5-7B-Instruct"

# 加载分词器
tokenizer = AutoTokenizer.from_pretrained(model_id)

# 定义4-bit量化配置
bnb_config = BitsAndBytesConfig(load_in_4bit=True)

# 加载模型
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    quantization_config=bnb_config,
    device_map="cuda:0",
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

Qwen2ForCausalLM(
  (model): Qwen2Model(
    (embed_tokens): Embedding(152064, 3584)
    (layers): ModuleList(
      (0-27): 28 x Qwen2DecoderLayer(
        (self_attn): Qwen2Attention(
          (q_proj): Linear4bit(in_features=3584, out_features=3584, bias=True)
          (k_proj): Linear4bit(in_features=3584, out_features=512, bias=True)
          (v_proj): Linear4bit(in_features=3584, out_features=512, bias=True)
          (o_proj): Linear4bit(in_features=3584, out_features=3584, bias=False)
        )
        (mlp): Qwen2MLP(
          (gate_proj): Linear4bit(in_features=3584, out_features=18944, bias=False)
          (up_proj): Linear4bit(in_features=3584, out_features=18944, bias=False)
          (down_proj): Linear4bit(in_features=18944, out_features=3584, bias=False)
          (act_fn): SiLUActivation()
        )
        (input_layernorm): Qwen2RMSNorm((3584,), eps=1e-06)
        (post_attention_layernorm): Qwen2RMSNorm((3584,), eps=1e-06)
      )
    )
    (norm): Qwen2RMSNorm((3584,), eps=1e-06)
    (rotary_emb): Qwen2RotaryEmbedding()
  )
  (lm_head): Linear(in_features=3584, out_features=152064, bias=False)
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

def chat(user_message, system_message="你是《黑神话：悟空》领域助手，回答准确、简明。"):
    messages = [
        {"role": "system", "content": system_message},
        {"role": "user", "content": user_message}
    ]
    
    # 应用对话模板
    text = tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True
    )
    
    model_inputs = tokenizer([text], return_tensors="pt").to(model.device)
    
    # 模型生成
    generated_ids = model.generate(
        input_ids=model_inputs.input_ids,
        attention_mask=model_inputs.attention_mask,
        max_new_tokens=256
    )
    
    # 解码时跳过 prompt 部分
    generated_ids = [
        output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
    ]
    
    response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
    
    return response

1
2
3

question_1 = "我该怎么成为天命人？"
answer_1 = chat(question_1)
print(f"问题: {question_1}\n回答:\n{answer_1}")

1
2
3
4
5
6
7
8

问题: 我该怎么成为天命人？
回答:
在《黑神话：悟空》的设定中，成为天命人需要满足特定条件和经历一系列考验。具体步骤包括：

1. **选对道路**：游戏中选择正确的道路至关重要，不同道路会影响角色的发展方向。
2. **完成任务**：跟随游戏主线和支线任务，完成各种挑战和任务。
...
最终是否能成为天命人取决于玩家的选择和表现。请注意，《黑神话：悟空》尚未正式发布，上述内容基于已知的游戏设定进行推测。实际游戏中可能有所不同。

1
2
3

question_2 = "如何获得并合成出云棍？"
answer_2 = chat(question_2)
print(f"问题: {question_2}\n回答:\n{answer_2}")

1
2
3
4
5
6
7
8

问题: 如何获得并合成出云棍？
回答:
在《黑神话：悟空》中，云棍可以通过完成特定任务和击败特定 boss 获得材料，然后在炼器炉中合成。

具体步骤如下：
1. **获取材料**：首先需要通过游戏中的特定任务或击败特定 boss 来获得制作云棍所需的材料。
...
请注意，《黑神话：悟空》目前尚未正式发布，上述信息基于游戏设定和预告片内容，具体细节可能会有所调整。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import os, re, json, datetime, time, random, glob
from openai import OpenAI

# 路径与配置
DATA_DIR = "./data"
SRC_MD = f"{DATA_DIR}/blackwukong.md"
TS = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
OUT_BASE_JSONL = f"{DATA_DIR}/wukong_base_{TS}.jsonl"
OUT_JSONL = f"{DATA_DIR}/wukong_dataset_{TS}.jsonl"

# 注意：此处为演示方便，不推荐在生产环境硬编码敏感信息
BASE_URL = "https://api.siliconflow.cn/v1"
MODEL_ID = "Qwen/Qwen3-235B-A22B-Instruct-2507"
API_KEY = "sk-..." 

client = OpenAI(api_key=API_KEY, base_url=BASE_URL)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

with open(SRC_MD, "r", encoding="utf-8") as f:
    raw_markdown = f.read()

# 按标题切分
matches = list(re.finditer(r"(?m)^(#{2,3})\\s+(.+)$", raw_markdown))
sections = []
# ... (此处省略切分与去重逻辑) ...
for i, m in enumerate(matches):
    s = m.start()
    e = matches[i + 1].start() if i + 1 < len(matches) else len(raw_markdown)
    block = raw_markdown[s:e].strip()
    if len(block) >= 100:
        sections.append(block)
# ...
print(f"sections={len(sections)}")

1

sections=14

1
2
3
4
5
6
7
8
9

SYS_PROMPT = (
    "你是《黑神话：悟空》的资深资料整理者。"
    "将给定原文片段转写为一条训练样本，严格输出JSON："
    '{"instruction":"用户问题","output":"权威完整答案"}。'
    "要求："
    "1. instruction 是自然语言问题；"
    "2. output 仅依据原文，不要臆测；"
    "3. 禁止任何额外说明或代码块。"
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

os.makedirs(os.path.dirname(OUT_BASE_JSONL), exist_ok=True)
base_written = 0
with open(OUT_BASE_JSONL, "w", encoding="utf-8") as fbase:
    for seg in sections:
        # ... (此处省略 API 调用与重试逻辑) ...
        try:
            resp = client.chat.completions.create(
                model=MODEL_ID,
                messages=[
                    {"role": "system", "content": SYS_PROMPT},
                    {"role": "user", "content": seg},
                ],
                temperature=0.2,
                max_tokens=600,
                response_format={"type": "json_object"},
            )
            # ... (解析与写入逻辑) ...
            obj = json.loads(resp.choices[0].message.content)
            ins = (obj.get("instruction") or "").strip()
            out = (obj.get("output") or "").strip()
            if ins and out:
                fbase.write(json.dumps({"instruction": ins, "output": out}, ensure_ascii=False) + "\n")
                base_written += 1
        except Exception:
            continue

print(f"base saved: {base_written} -> {OUT_BASE_JSONL}")

1

base saved: 14 -> ./data/wukong_base_20251109_201117.jsonl

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

NUM_VARIANTS = 14
written = 0
# ...
with open(IN_BASE_JSONL, "r", encoding="utf-8") as fr, open(OUT_JSONL, "w", encoding="utf-8") as fw:
    for line in fr:
        # ...
        base_q = (json.loads(line).get("instruction") or "").strip()
        answer = (json.loads(line).get("output") or "").strip()

        # 设计用于数据增强的 Prompt
        paraphrase_prompt = (
            "严格输出 JSON 对象：{\\\"paraphrases\\\": [\\\"...\\\"]}；"
            "禁止任何额外文本/代码块/前后缀。"
            "每项必须是可直接回答的等价问法..."
        )
        
        # ... (API 调用与重试逻辑) ...
        r2 = client.chat.completions.create(...)
        
        # ... (解析与写入逻辑) ...
        arr = json.loads(r2.choices[0].message.content).get("paraphrases", [])
        for s in arr:
            # ... (去重与格式化) ...
            fw.write(json.dumps({"instruction": s, "output": answer}, ensure_ascii=False) + "\n")
            written += 1
            
print(f"saved: {written} -> {OUT_JSONL}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import os, glob, json, torch
from datasets import load_dataset
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
from peft import prepare_model_for_kbit_training

# ... (路径与配置) ...

# 加载数据集
train_set = load_dataset("json", data_files=train_jsonl, split="train")

# 加载分词器，并处理 pad_token
tokenizer = AutoTokenizer.from_pretrained(checkpoint_id, trust_remote_code=True)

tokenizer.pad_token, tokenizer.eos_token, tokenizer.pad_token_id, tokenizer.eos_token_id

1

('<|endoftext|>', '<|im_end|>', 151643, 151645)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# 4-bit 量化配置
bnb_cfg = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_use_double_quant=True,
    bnb_4bit_compute_dtype=torch.bfloat16,
)
base_model = AutoModelForCausalLM.from_pretrained(
    checkpoint_id,
    trust_remote_code=True,
    quantization_config=bnb_cfg,
    device_map="cuda:0",
)

# k-bit 训练预处理
base_model = prepare_model_for_kbit_training(base_model)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from peft import LoraConfig, get_peft_model
from peft.utils import TRANSFORMERS_MODELS_TO_LORA_TARGET_MODULES_MAPPING

# LoRA 配置
lora_cfg = LoraConfig(
    r=16,
    lora_alpha=32,
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
    target_modules=TRANSFORMERS_MODELS_TO_LORA_TARGET_MODULES_MAPPING["qwen2"],
)
peft_model = get_peft_model(base_model, lora_cfg)
peft_model.print_trainable_parameters()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

def format_sample_for_qwen(record):
    # ... 
    # 构造不包含答案的 prompt
    msgs_no_assist = [
        {"role": "system", "content": "..."},
        {"role": "user", "content": instr},
    ]
    prompt_ids = tokenizer.apply_chat_template(msgs_no_assist, ...)

    # 构造包含答案的完整对话
    msgs_full = msgs_no_assist + [{"role": "assistant", "content": ans}]
    full_ids = tokenizer.apply_chat_template(msgs_full, ...)

    # 生成 labels，仅保留答案部分的 token
    cut = len(prompt_ids)
    labels = [-100] * cut + full_ids[cut:]

    return {"input_ids": full_ids, "labels": labels}

# 对整个数据集应用格式化函数
proc_train = train_set.map(format_sample_for_qwen, ...)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from transformers import TrainingArguments, Trainer

args = TrainingArguments(
    output_dir=run_dir, # run_dir 为动态生成的检查点目录
    per_device_train_batch_size=1,
    gradient_accumulation_steps=8,
    learning_rate=1e-3,
    num_train_epochs=4,
    lr_scheduler_type="linear",
    warmup_ratio=0.03,
    logging_steps=1,
    optim="adamw_torch",
    bf16=True, # 如果支持，优先使用 bf16
)

trainer = Trainer(
    model=peft_model,
    args=args,
    train_dataset=proc_train,
    data_collator=collator,   # collator 为自定义的数据整理器
)

# 开始训练并保存 LoRA 适配器
train_output = trainer.train()
peft_model.save_pretrained(run_dir)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

peft_model.eval()
TEST_QUERIES = [ "我该怎么成为天命人？", "如何获得并合成出云棍？" ]

@torch.no_grad()
def infer_one(question: str) -> str:
    # ...（推理逻辑，构造prompt并调用 peft_model.generate）
    return tokenizer.decode(out_ids, skip_special_tokens=True).strip()

for q in TEST_QUERIES:
    ans = infer_one(q)
    print(f"Q: {q}\nA: {ans}\n" + "-" * 60)

1
2
3
4
5
6

Q: 我该怎么成为天命人？
A: 想要成为天命之人，首先你需要收集“部分灵根”，然后在特定地点（如石卵处）激活它们，之后才能解锁通往黄风岭、狼烟堡等地点的通行权，并进行后续剧情。
------------------------------------------------------------
Q: 如何获得并合成出云棍？
A: 在“黄风岭-挟魂崖”找到6个佛目珠后，前往“挟魂崖-枕石坪”BOSS石先锋所在区域，使用佛目珠召唤并击败BOSS石敢当，获得材料“铁石心”，之后即可解锁铸造出云棍。
------------------------------------------------------------