最近在学习大模型微调，大模型构建，遇到一些看不懂的python语法就想记录一下

1.带条件的列表推导式

[表达式 for 变量 in 可迭代对象 if 条件式]

代码例子

# ---------------- 1. 读取并解析 JSONL 数据文件 ----------------
# 采用列表推导式（List Comprehension）高效逐行处理：
# - open("/root/qixiu_data.jsonl", encoding="utf-8"): 打开本地汽修问答/指令数据集
# - if l.strip(): 过滤掉文件中的空行或纯空白字符，防止 json.loads 解析空字符串报错
# - json.loads(l): 将每一行 JSON 字符串反序列化为 Python 字典对象（dict）
rows = [json.loads(l) for l in open("/root/qixiu_data.jsonl", encoding="utf-8") if l.strip()]

2.字典键索引

例如

r["instruction"]

以下是r的描述
r = {
    "instruction": "冷车启动困难怎么办？",
    "output": "建议检查蓄电池电压和火花塞积碳情况。"
}

执行 r["instruction"] 时：

• 返回结果为字符串："冷车启动困难怎么办？"

注意： 如果字典 r 中不存在 "instruction" 这个键，直接使用 r["instruction"] 会抛出 KeyError 异常。若要安全取值，也可以使用 r.get("instruction")。

3.tokenizer.apply_chat_template

tokenizer.apply_chat_template 是 Hugging Face transformers 库提供的一个核心方法，专门用于把结构化的多轮对话列表（List of Dicts）自动转换并拼接成当前模型专属的 Prompt 纯文本或 Token ID 格式。

参数逐一解析：：

参数	作用
msgs	对话数据源。标准的 OpenAI 格式列表，例如：   [{"role": "system", "content": "..."}, {"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]
tokenize=False	返回格式控制：   • False：只返回拼接好的纯字符串文本（str）。   • True（默认）：直接返回编码后的数字 Token ID 列表 / 张量。
chat_template_kwargs	传递给 Jinja 模板的自定义控制参数。   {"enable_thinking": False} 是针对支持思维链/思考模式的模型（如 Qwen-Thinking、DeepSeek-R1 等），指示模板不要插入 <think> 引导词，直接输出正文。
add_generation_prompt	加不加上引出回答的标签

4.Python 切片语法

4.1 省略参数（从头开始 / 切到末尾）

data[:4]       # [0, 1, 2, 3]（省略 start，默认从头切到索引 3）
data[6:]       # [6, 7, 8, 9]（省略 stop，从索引 6 切到末尾）
data[:]        # [0, 1, 2, ..., 9]（全量浅拷贝）

4.2 使用负索引（微调代码中的关键用法）

data[:-1]      # [0, 1, 2, 3, 4, 5, 6, 7, 8]（切掉倒数第 1 个元素，保留前面所有内容）
data[-3:]      # [7, 8, 9]（只取倒数后 3 个元素）

4.3 步长 step

data[::2]      # [0, 2, 4, 6, 8]（步长为 2，隔一个取一个）
data[::-1]     # [9, 8, 7, 6, 5, 4, 3, 2, 1, 0]（步长为 -1，快速翻转序列）

5.列表乘法/重复构造

[-100] * 5

生成一个由 5 个 -100 组成的列表：

[-100, -100, -100, -100, -100]

6.迭代器组合与嵌套元组解包

6.1 代码

for i, (tid, lab) in enumerate(zip(ex["input_ids"], ex["labels"])):

原理

循环头部的 for i, (tid, lab) in ... 运用了 结构模式匹配 / 嵌套解包 语法。

Python 会严格按照等号两边的数据形状（Shape）进行一一对应：

ex["input_ids"] ──┐
                  ├──► zip(...) ──► (tid, lab) ──┐
ex["labels"]    ──┘                              ├──► enumerate(...) ──► (i, (tid, lab))
                                                 │
                                 自动生成计数器 ──┘

7.huggingface 数据集制作

代码示例

from datasets import Dataset
from transformers import DataCollatorForSeq2Seq

# ---------------- 1. 将原始列表转换为 Hugging Face Dataset 并应用模板 ----------------
# Dataset.from_list(rows): 将 Python 字典列表转换为高效的内存 Dataset 对象
# .map(fmt): 遍历数据集中的每条样本，调用先前定义的 fmt() 函数，追加 "full" 和 "prompt" 两个字段
ds = Dataset.from_list(rows).map(fmt)

# ---------------- 2. 批量执行分词与 Loss 掩码 ----------------
# .map(tokenize, ...): 对数据集执行分词、截断以及 Prompt 部分的 label=-100 掩码处理
# remove_columns=ds.column_names: 移除所有原始文本字段（如 "instruction", "output", "full", "prompt"），
# 仅保留 tokenize 返回的纯数值特征 ("input_ids", "attention_mask", "labels")，减少内存占用并适配 Trainer 输入
tok_ds = ds.map(tokenize, remove_columns=ds.column_names)

# ---------------- 3. 构建动态批填充器 (Data Collator) ----------------
# DataCollatorForSeq2Seq: 专用于生成式模型的批整理器
# 1. 动态 Padding: 自动将同一个 Batch 内的样本填充（Pad）到当前批次中的「最大长度」，而非全局最大长度（1536），大幅节省显存并加快训练
# 2. 标签对齐: 对 labels 中的 Pad 区域同样自动填充 -100，确保填充位置不参与 Loss 计算
collator = DataCollatorForSeq2Seq(tokenizer, padding=True)

# ---------------- 4. 打印训练步数统计 ----------------
# 计算在 batch_size=4（每步处理 4 条样本）的配置下，完成一个 Epoch（轮次）所需的优化步数（Steps）
print("训练样本:", len(tok_ds), "| 每步 4 条 →", len(tok_ds) // 4, "步/轮")

7.1 Dataset.from_list(rows)（结构转换）

• 输入：原生的 Python 字典列表 rows（例如 [{"instruction": "...", "output": "..."}, ...]）。

• 动作：将其转化为 Hugging Face datasets.Dataset 对象。

• 目的：利用 Arrow 内存格式优化读写速度，并解锁 .map()、多进程处理、批量流式加载等高级功能。

7.2. .map(fmt)（批量应用模板）

• 动作：类似遍历循环，自动把数据集里的每一条样本传给 fmt 函数执行一遍。

• 效果：将 fmt() 函数返回的新字段（"full" 和 "prompt"）自动追加写回数据集中。

执行前后的数据形态对比

• 处理前（单条样本）：

  {
      "instruction": "冷车打不着火怎么修？",
      "output": "先检查电瓶电压与火花塞积碳。"
  }
  

• 执行 ds = Dataset.from_list(rows).map(fmt) 后（单条样本新增两个字段）：

tok_ds = ds.map(tokenize, remove_columns=ds.column_names)

完整执行流程

1. 获取字段名列表:
   ds.column_names  ──►  ["instruction", "output", "full", "prompt"]

2. 逐行执行 tokenize:
   单行样本 ex  ──►  tokenize(ex)  ──►  返回 {"input_ids": ..., "attention_mask": ..., "labels": ...}

3. map 自己执行删除清理:
   .map 拿到结果后，对照 remove_columns 里的列表，把旧字段 ("instruction", "output" 等) 从数据集里

7.4 map(tokenize)

map()函数介绍

# Dataset.map 的标准签名示意：
ds.map(
    function,             # ① 唯一必填：要执行的函数
    with_indices=False,   # ② 自身的控制开关
    batched=False,        # ② 自身的控制开关
    batch_size=1000,      # ② 自身的控制开关
    remove_columns=None,  # ② 自身的控制开关
    num_proc=None,        # ② 自身的控制开关
    fn_kwargs=None,       # ③ 专门用来打包「传递给 function 的参数」的通道
    ...
)

将之前写好的 tokenize 函数应用到数据集中的每一条样本上。自动完成分词、最大长度截断、以及把 Prompt 部分的标签标记为 -100。

7.5 remove_columns=ds.column_names

ds.column_names 会获取当前数据集现有的所有列名（如 ["instruction", "output", "full", "prompt"]）。

这里的指令是：在执行完 tokenize 后，把这些旧的纯文本列全部删掉，只保留 tokenize 函数返回的新字段。

处理前后的数据形态对比

处理前（ds 中的单条数据）：全是文本字符串

{
    "instruction": "冷车打不着火怎么修？",
    "output": "先检查电瓶电压与火花塞积碳。",
    "prompt": "<|im_start|>user\n冷车打不着火怎么修？...",
    "full": "<|im_start|>user\n冷车打不着火怎么修？...先检查电瓶..."
}

处理后（tok_ds 中的单条数据）：全是 PyTorch / Trainer 可直接读取的纯数字列表

{
    "input_ids":      [101, 872, 534, ..., 998, 203, 102],
    "attention_mask": [  1,   1,   1, ...,   1,   1,   1],
    "labels":         [-100, -100, -100, ..., 998, 203, 102]
}

collator = DataCollatorForSeq2Seq(tokenizer, padding=True)

DataCollatorForSeq2Seq 是一个动态批处理器（Data Collator），专门用于在模型训练取数据时，把长度不一的样本整齐地打包成 Batch 张量。

处理前后的 Batch 数据对比示例

假设一个 Batch 抽中了 2 条长短不一的样本：

• 样本 1 长度为 3：input_ids = [10, 20, 30]，labels = [-100, 20, 30]

• 样本 2 长度为 5：input_ids = [11, 22, 33, 44, 55]，labels = [-100, -100, 33, 44, 55]

经过 collator 动态批处理后输出的 Tensor：

# input_ids (短的样本 1 自动补 0 到长度 5):
tensor([[ 10,  20,  30,   0,   0],
        [ 11,  22,  33,  44,  55]])

# attention_mask (补 0 的位置标记为 0):
tensor([[  1,   1,   1,   0,   0],
        [  1,   1,   1,   1,   1]])

# labels 经过tokenizer+DataCollatorForSeq2Seq 处理(短的样本 1 自动补 -100 到长度 5，不计算损失):
tensor([[-100,   20,   30, -100, -100],
        [-100, -100,   33,   44,   55]])

8.后续学习代码

注入lora

lora_cfg = LoraConfig(
    r=16,                # 低秩维度（越大学得越强，但过拟合风险越高）
    lora_alpha=32,       # 缩放系数，通常 = 2*r
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj",   # 注意力
                    "gate_proj", "up_proj", "down_proj"],      # 前馈
)

model = get_peft_model(model, lora_cfg)
model.print_trainable_parameters()   # 看可训练参数占比

配置训练超参数

# 训练权重与日志检查点的保存输出目录
OUT = "/root/qixiu-lora-out-nb"

# ---------------- 1. 配置训练超参数 (TrainingArguments) ----------------
args = TrainingArguments(
    output_dir=OUT,                         # 模型检查点 (Checkpoints) 和输出文件的保存路径
    num_train_epochs=10,                    # 完整遍历训练数据集的总轮数 (Epochs)
    per_device_train_batch_size=2,          # 单张 GPU 上的单步批处理大小 (Batch Size)
    gradient_accumulation_steps=1,          # 梯度累积步数；1 表示每处理 1 个 batch 就立即更新一次梯度
    
    # 显存优化：激活梯度检查点技术 (Gradient Checkpointing)
    # 用少量的计算时间换取大量显存节省（前向传播不保存所有中间激活值，反向传播时重新计算）
    gradient_checkpointing=True,
    
    learning_rate=2e-4,                     # 最大学习率（LoRA 微调通常采用 1e-4 ~ 3e-4，高于全量微调）
    lr_scheduler_type="cosine",             # 学习率调度策略：余弦退火衰减 (Cosine)，平滑降至 0 避免震荡
    bf16=True,                              # 开启 bfloat16 混合精度训练，提升训练速度并节省显存
    logging_steps=20,                       # 每训练 20 个 Step 打印并记录一次日志（如 Loss、LR）
    save_strategy="epoch",                  # 保存检查点的策略："epoch" 表示每个 Epoch 结束时保存一次
    save_total_limit=1,                     # 最多保留 1 个最新的 Checkpoint，自动删除旧检查点以节省磁盘空间
    optim="adamw_torch",                    # 优化器选择：PyTorch 原生实现的 AdamW
    report_to="none",                       # 不使用第三方监控面板（如 wandb/tensorboard），仅本地控制台输出
    seed=42,                                # 固定全局随机种子，确保实验具备可复现性
)

# ---------------- 2. 自定义损失记录回调函数 (TrainerCallback) ----------------
class LossHistory(TrainerCallback):
    """
    自定义回调类：在 Trainer 触发日志记录时自动捕获并存储 (step, loss)，
    便于后续使用 matplotlib 绘制直观的 Loss 下降收敛曲线。
    """
    def __init__(self):
        self.losses = []                    # 用于存放历史记录的列表，元素格式为 (global_step, loss_value)

    def on_log(self, args, state, control, logs=None, **kwargs):
        # 当本次日志中包含训练损失 "loss" 字段时，记录全局步数和对应的损失值
        if logs and "loss" in logs:
            self.losses.append((state.global_step, logs["loss"]))

# 实例化回调对象
hist = LossHistory()

# ---------------- 3. 计算并预估训练总耗时 ----------------
# 总步数 = (样本总量 * 总轮数) // 单步批大小
total_steps = len(tok_ds) * 10 // 2

# 假设在当前 GPU 与显存配置下，处理 1 个 step 约耗时 0.5 秒
print(f"总步数 ≈ {total_steps}（约 {total_steps * 0.5:.0f} 秒）")
print("准备好了，下一步开跑！")

进行训练

# ---------------- 2. 实例化 Hugging Face Trainer ----------------
trainer = Trainer(
    model=model,                 # 已注入 LoRA 适配器的待训练模型 (PeftModel)
    args=args,                   # 训练超参数配置 (TrainingArguments)
    train_dataset=tok_ds,        # 经过分词与 Loss 掩码预处理后的训练集
    data_collator=collator,      # 动态 Batch Padding 整理器 (DataCollatorForSeq2Seq)
    callbacks=[hist]             # 挂载自定义回调对象，用于实时收集 (step, loss) 日志
)

# ---------------- 3. 启动模型微调训练 (Start Training) ----------------
# 自动执行 Forward、Backward、Optimizer Step、Scheduler Step 以及 Checkpoint 周期保存
result = trainer.train()

# ---------------- 4. 输出训练结果摘要 ----------------
# result 包含训练运行期间的关键统计指标（如总耗时、总优化步数、全周期加权平均 Loss）
print("\n✅ 训练结束  |  共", result.global_step, "步  |  平均 loss:", round(result.training_loss, 4))

输出lora参数

# ---------------- 1. 指定最终 LoRA 适配器（Adapter）的保存路径 ----------------
# OUT 是之前定义的根目录（如 "/root/qixiu-lora-out-nb"）
# 拼接子路径为 "/root/qixiu-lora-out-nb/lora_adapter"
ADAPTER = OUT + "/lora_adapter"

# ---------------- 2. 保存微调后的 LoRA 适配器权重 ----------------
# 注意：对于被 PEFT 包装的模型，save_pretrained 仅保存可训练的 LoRA 低秩矩阵参数
# （生成 adapter_model.safetensors / adapter_config.json 等），通常只有几十 MB，而不会重复保存整个基座模型
model.save_pretrained(ADAPTER)

# ---------------- 3. 保存配套的分词器配置文件 ----------------
# 将当前 tokenizer 的词表、特殊 token 映射以及 chat_template 配置同步保存至同目录
# 便于后续使用 AutoTokenizer.from_pretrained(ADAPTER) 或部署工具时直接开箱即用
tokenizer.save_pretrained(ADAPTER)

print("✅ 适配器已保存:", ADAPTER)

# ---------------- 4. 遍历并打印保存目录下的所有生成文件 ----------------
# 检查保存的完整性（通常包含 adapter_config.json、adapter_model.safetensors、tokenizer.json 等）
for f in sorted(os.listdir(ADAPTER)):
    print("  ", f)

以生产方式加载lora进行对话测试

import gc
import torch
from peft import PeftModel

# ---------------- 1. 深度清理训练态显存 ----------------
# 训练过程中显存中常驻有：模型梯度、优化器动量状态 (Optimizer States) 及中间激活值
# del 删除变量引用，断开 PyTorch 对象的强引用计数
del model, trainer

# 强制触发 Python 垃圾回收机制 (Garbage Collection)，回收无引用对象
gc.collect()

# 清理 PyTorch 显存分配器中的未分配缓存碎片，将可用显存归还给操作系统/GPU 驱动
torch.cuda.empty_cache()

# ---------------- 2. 重新加载基座并挂载微调后的 LoRA 适配器 ----------------
print("重新加载基座 + LoRA 适配器...")

# 加载干净的原始基座模型（Base Model）
base2 = AutoModelForCausalLM.from_pretrained(
    MODEL_ID,
    torch_dtype=torch.bfloat16,  # 推荐使用规范命名 torch_dtype
    device_map="auto"
)

# 使用 PEFT 将刚刚训练保存的 LoRA 权重挂载到基座模型上
model_ft = PeftModel.from_pretrained(base2, ADAPTER)

# 切换为推理/评估模式：关闭 Dropout、冻结 Batch Normalization 行为
model_ft.eval()

print("✅ 微调模型已就绪")

# ---------------- 3. 定义简易问答推理函数 ----------------
def ask(q, max_new=600, temperature=0.7):
    """
    接收用户自然语言问题，使用微调后的模型生成专业解答。
    
    参数:
        q (str): 用户的提问文本
        max_new (int): 最大新生成的 Token 数量
        temperature (float): 采样温度，控制生成的随机性（0.0 更确定、保守；1.0 更发散）
    返回:
        str: 模型生成的回答文本（已剥离 Prompt 与特殊标记）
    """
    # 1. 组装单轮对话格式
    msgs = [{"role": "user", "content": q}]
    
    # 2. 套用 Chat 模板，并关闭思考链以直接获取正文回复
    prompt = tokenizer.apply_chat_template(
        msgs,
        tokenize=False,
        add_generation_prompt=True,
        chat_template_kwargs={"enable_thinking": False}
    )
    
    # 3. 将 Prompt 分词并转为 PyTorch 张量，同时自动放置到模型所在的 GPU 设备上
    inputs = tokenizer(prompt, return_tensors="pt").to(model_ft.device)
    
    # 4. 生成预测结果
    # torch.no_grad(): 禁用自动求导引擎，避免在推理阶段保存计算图与梯度，大幅节省显存并提速
    with torch.no_grad():
        out = model_ft.generate(
            **inputs,
            max_new_tokens=max_new,    # 生成的最大 token 限制
            do_sample=True,            # 开启采样策略（结合 temperature 与 top_p）
            temperature=temperature,    # 控制文本多样性
            top_p=0.9                  # Nucleus Sampling：仅在累积概率达到 90% 的词表中采样，避免生成离群低质词
        )
    
    # 5. 解码并提取回复部分
    # inputs.input_ids.shape[1] 是输入 Prompt 的长度；切片截取掉前面的输入，仅解码模型新生成的 Token
    # skip_special_tokens=True: 自动过滤掉 <|im_end|> 等特殊结束符号
    return tokenizer.decode(out[0][inputs.input_ids.shape[1]:], skip_special_tokens=True).strip()

将lora参数合入模型

# ---------------- 1. 指定合并后全量模型的保存路径 ----------------
OUT_MERGED = "/root/Qwen3-4B-qixiu-merged-nb"

# ---------------- 2. 检查并执行权重合并 (Merge & Save) ----------------
# hasattr(model_ft, "merge_and_unload"): 判断当前模型是否为带适配器的 PeftModel 实例
# 只有未合并的 PEFT 模型才具有 merge_and_unload 方法
if hasattr(model_ft, "merge_and_unload"):
    print("合并中...")
    
    # 1. 物理合并权重：
    # 将 LoRA 的低秩矩阵增量 (W = W_base + (B * A) * scaling) 永久加合到基座模型的原始权重矩阵中
    # 并卸载 LoRA 适配器结构，返回一个纯净的标准 AutoModelForCausalLM 实例
    # 好处：消除 LoRA 旁路带来的额外推理计算延迟，提升推理吞吐量，且无需依赖 PEFT 库即可加载
    merged = model_ft.merge_and_unload()
    
    # 2. 保存合并后的完整模型权重（包含 config.json、model.safetensors 等完整模型文件）
    merged.save_pretrained(OUT_MERGED)
    
    # 3. 同步保存分词器文件，使该目录构成一个开箱即用的完整独立大模型
    tokenizer.save_pretrained(OUT_MERGED)
    
    print("✅ 合并完成:", OUT_MERGED)

else:
    # 防呆提示：如果已经执行过合并，model_ft 已经是普通模型，无需也不能重复合并
    print("模型已是合并后的普通模型，跳过。想重新合并请先重跑第 14 步重新加载。")