分为上下两集吧。

上集。

这家没有自我介绍。

我面试的大模型类型基本上都是深挖过去的项目经历，以及大模型相关知识掌握程度。希望对你有所帮助。

1.从你最近的项目里边，聊一下你在里面负责的东西，你在里边所承担的一些角色和分工相关的一些

2.当时你们选择XB(16B)基模的时候是出于什么考虑？

3.终端大模型部署约束，你们是怎么做的？在为微调方法和量化策略上你这边是怎么做技术方面的选型和决策的？

4.在数据层面主动优化提升某个方面的一些指标，这个你有一些具体的案例吗？

5.微调之后上线是你这边负责吗？

6.你们这边除了一些公开基本测试方法，具体的业务数据，在你们训练这个模型的时候，还有那些定制化的评估策略呢？

7.我刚听到你这边只是做了一些意图识别相关精调吗？还有其他的吗？

8.全参微调、LoRA微调等等，你觉得他们各自解决了哪些方面的问题？

9.在你们整个迭代中，整条链路中，模型从数据集再到微调再到迭代，那你整个链路，撕下来，时长是多少？

10.这个迭代的时长，会有哪些相关因素影响它？

11.针对训练时长的话，你有自己的一个时间相对换算的一个方式吗？

12.了解换算公式吗。

13.DPO和RHLH有做什么区分吗？

14.你用的是全量微调，那有没有遇到全量微调好了之后把之前获得的知识东西搞乱的情况

15.多模态的大模型你这边有涉及到吗？

未完待续...

送大家一个DPO和RHLH的demo，Cluade分享

"""

用"道歉邮件"偏好数据演示 RLHF 与 DPO 训练流程

依赖: pip install torch transformers trl datasets

"""

import torch

import torch.nn.functional as F

from torch.optim import AdamW

from transformers import AutoTokenizer, AutoModelForCausalLM, AutoModelForSequenceClassification

# ─────────────────────────────────────────────

# 共用: 偏好数据（chosen / rejected 对）

# ─────────────────────────────────────────────

PREFERENCE_DATA = [

    {

        "prompt":   "",

        "chosen":   "您好，对帮我写一封道歉邮件，语气要真诚。于上次的误解我深感抱歉，这完全是我的疏忽。我已仔细反思了自己的行为，并会认真改正。希望您能接受我诚挚的道歉。",

        "rejected": "你好，我道歉。希望你不要生气，以后不会再犯了。",

    }

]

 

# ══════════════════════════════════════════════════════════════════

#  Part 1: RLHF 训练流程

#  步骤: SFT → 训练奖励模型(RM) → PPO 强化学习

# ══════════════════════════════════════════════════════════════════

class RewardModel(torch.nn.Module):

    """

    奖励模型: 在预训练模型顶部加一个线性层输出标量分数。

    用 chosen/rejected 对做 pairwise ranking loss 训练。

    """

    def __init__(self, base_model_name: str):

        super().__init__()

        self.backbone = AutoModelForSequenceClassification.from_pretrained(

            base_model_name, num_labels=1

        )

    def forward(self, input_ids, attention_mask):

        outputs = self.backbone(input_ids=input_ids, attention_mask=attention_mask)

        return outputs.logits.squeeze(-1)          # shape: (batch,)

 

def train_reward_model(

    reward_model: RewardModel,

    tokenizer: AutoTokenizer,

    preference_data: list,

    epochs: int = 3,

    lr: float = 1e-5,

):

    """

    用 Bradley-Terry pairwise ranking loss 训练奖励模型:

        L = -log σ(r_chosen - r_rejected)

    """

    optimizer = AdamW(reward_model.parameters(), lr=lr)

    reward_model.train()

    for epoch in range(epochs):

        total_loss = 0.0

        for item in preference_data:

            # ── 编码 chosen ──────────────────────────────────────

            chosen_text   = item["prompt"] + item["chosen"]

            rejected_text = item["prompt"] + item["rejected"]

            chosen_enc = tokenizer(

                chosen_text, return_tensors="pt",

                truncation=True, max_length=256, padding="max_length"

            )

            rejected_enc = tokenizer(

                rejected_text, return_tensors="pt",

                truncation=True, max_length=256, padding="max_length"

            )

            r_chosen   = reward_model(**chosen_enc)    # 标量分数

            r_rejected = reward_model(**rejected_enc)

            # Pairwise ranking loss

            loss = -F.logsigmoid(r_chosen - r_rejected).mean()

            optimizer.zero_grad()

            loss.backward()

            optimizer.step()

            total_loss += loss.item()

        print(f"[RM] Epoch {epoch+1}/{epochs}  loss={total_loss:.4f}"

              f"  r_chosen={r_chosen.item():.3f}  r_rejected={r_rejected.item():.3f}")

    return reward_model

 

def ppo_step(

    policy_model,

    ref_model,

    reward_model: RewardModel,

    tokenizer: AutoTokenizer,

    prompt: str,

    kl_coef: float = 0.1,

    lr: float = 1e-6,

):

    """

    简化版单步 PPO 更新:

    1. 策略模型对 prompt 采样一条回复

    2. 奖励模型给分

    3. 减去 KL 惩罚 (防止偏离参考模型太远)

    4. 用 policy gradient 更新策略

    完整实现见 trl.PPOTrainer

    """

    optimizer = AdamW(policy_model.parameters(), lr=lr)

    policy_model.train()

    ref_model.eval()

    """`AdamW` 是一个**优化器**，它的工作是根据梯度去更新模型参数。

    拆开来看：

    **`policy_model.parameters()`**

    告诉优化器要优化哪些参数——这里就是策略模型的全部权重。优化器会持有这些参数的引用，每次 `optimizer.step()` 时去修改它们。

    **`lr=lr`**（学习率）

    控制每次更新的步子迈多大。比如 `lr=1e-6` 就是每次只微调一点点，防止把原来训好的模型改崩。

    **为什么用 AdamW 而不是普通 Adam？**

    AdamW 在 Adam 的基础上加了**权重衰减（weight decay）**，直接对参数本身施加 L2 惩罚，而不是混在梯度里。这在 LLM 微调中更稳定，是目前的标准做法。

    用一个简单类比来说：

    ```

    loss.backward()      # 量出山坡的坡度（梯度）

    optimizer.step()     # 沿坡度往下走一步（lr 决定步长，AdamW 决定怎么走）

    optimizer.zero_grad() # 擦掉上一步留下的脚印，准备下一次测量

    ```

    这三行几乎出现在所有 PyTorch 训练循环里，顺序不能乱。"""

    # ── 1. 策略模型采样回复 ──────────────────────────────────────

    prompt_enc = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=128)

    with torch.no_grad():

        generated = policy_model.generate(

            **prompt_enc,

            max_new_tokens=80,

            do_sample=True,

            temperature=0.9,

        )

    response_ids = generated[:, prompt_enc["input_ids"].shape[1]:]

    response_text = tokenizer.decode(response_ids[0], skip_special_tokens=True)

    # ── 2. 奖励模型打分 ──────────────────────────────────────────

    full_text = prompt + response_text

    full_enc  = tokenizer(full_text, return_tensors="pt",

                          truncation=True, max_length=256, padding="max_length")

    reward = reward_model(**full_enc)                   # 标量

    # ── 3. KL 惩罚: log π_policy - log π_ref ────────────────────

    with torch.no_grad():

        ref_logits    = ref_model(**full_enc).logits

    policy_logits = policy_model(**full_enc).logits

    ref_log_prob    = F.log_softmax(ref_logits,    dim=-1)

    policy_log_prob = F.log_softmax(policy_logits, dim=-1)

    kl_div = (policy_log_prob.exp() * (policy_log_prob - ref_log_prob)).sum(-1).mean()

    # ── 4. PPO 目标: 最大化奖励 - β * KL ───────────────────────

    loss = -(reward - kl_coef * kl_div)

    optimizer.zero_grad()

    loss.backward()

    optimizer.step()

    print(f"[PPO] reward={reward.item():.3f}  kl={kl_div.item():.4f}"

          f"  loss={loss.item():.4f}")

    print(f"      生成回复: {response_text[:60]}...")

    return response_text

 

def run_rlhf_pipeline(model_name: str = "gpt2"):

    print("\n" + "═"*60)

    print("  RLHF 训练流程")

    print("═"*60)

    tokenizer = AutoTokenizer.from_pretrained(model_name)

    tokenizer.pad_token = tokenizer.eos_token

    # ── Step 1: SFT（略，假设已完成）────────────────────────────

    print("\n[Step 1] SFT: 在示范数据上微调（此处跳过，加载预训练权重）")

    sft_model = AutoModelForCausalLM.from_pretrained(model_name)

    # ── Step 2: 训练奖励模型 ─────────────────────────────────────

    print("\n[Step 2] 训练奖励模型（RM）…")

    reward_model = RewardModel(model_name)

    reward_model = train_reward_model(reward_model, tokenizer, PREFERENCE_DATA, epochs=3)

    # ── Step 3: PPO 强化学习 ─────────────────────────────────────

    print("\n[Step 3] PPO 强化学习优化策略模型…")

    ref_model = AutoModelForCausalLM.from_pretrained(model_name)

    ref_model.eval()

    policy_model = AutoModelForCausalLM.from_pretrained(model_name)

    for _ in range(2):          # 演示跑 2 步

        ppo_step(

            policy_model, ref_model, reward_model,

            tokenizer, PREFERENCE_DATA[0]["prompt"]

        )

    print("\n✓ RLHF 完成\n")

    return policy_model

 

# ══════════════════════════════════════════════════════════════════

#  Part 2: DPO 训练流程

#  直接在 (prompt, chosen, rejected) 上优化，无需奖励模型

# ══════════════════════════════════════════════════════════════════

def compute_log_probs(

    model,

    tokenizer: AutoTokenizer,

    prompt: str,

    response: str,

) -> torch.Tensor:

    """

    计算模型在给定 prompt 条件下对 response 的平均 log 概率。

    只计算 response 部分的 token loss（prompt 部分 mask 掉）。

    """

    full_text = prompt + response

    full_enc  = tokenizer(full_text, return_tensors="pt",

                          truncation=True, max_length=256)

    prompt_enc = tokenizer(prompt, return_tensors="pt",

                           truncation=True, max_length=256)

    prompt_len = prompt_enc["input_ids"].shape[1]

    with torch.no_grad() if model.training is False else torch.enable_grad():

        logits = model(**full_enc).logits                   # (1, seq, vocab)

    # shift: 预测下一个 token

    shift_logits = logits[:, :-1, :]                        # (1, seq-1, vocab)

    shift_labels = full_enc["input_ids"][:, 1:]             # (1, seq-1)

    log_probs = F.log_softmax(shift_logits, dim=-1)         # (1, seq-1, vocab)

    token_log_probs = log_probs.gather(

        2, shift_labels.unsqueeze(-1)

    ).squeeze(-1)                                           # (1, seq-1)

    # 只取 response 部分

    response_log_probs = token_log_probs[:, prompt_len - 1:]

    return response_log_probs.mean()                        # 标量

 

def dpo_loss(

    policy_model,

    ref_model,

    tokenizer: AutoTokenizer,

    prompt: str,

    chosen: str,

    rejected: str,

    beta: float = 0.1,

) -> torch.Tensor:

    """

    DPO 损失函数:

        L_DPO = -log σ [ β * (log π(chosen|x) - log π_ref(chosen|x))

                           - β * (log π(rejected|x) - log π_ref(rejected|x)) ]

    等价于: 让策略相对参考模型更偏向 chosen，远离 rejected。

    """

    # ── 策略模型 log prob ────────────────────────────────────────

    lp_chosen   = compute_log_probs(policy_model, tokenizer, prompt, chosen)

    lp_rejected = compute_log_probs(policy_model, tokenizer, prompt, rejected)

    # ── 参考模型 log prob（冻结）────────────────────────────────

    ref_model.eval()

    with torch.no_grad():

        ref_lp_chosen   = compute_log_probs(ref_model, tokenizer, prompt, chosen)

        ref_lp_rejected = compute_log_probs(ref_model, tokenizer, prompt, rejected)

    # ── 计算隐式奖励差值 ─────────────────────────────────────────

    chosen_reward   = beta * (lp_chosen   - ref_lp_chosen)

    rejected_reward = beta * (lp_rejected - ref_lp_rejected)

    loss = -F.logsigmoid(chosen_reward - rejected_reward)

    # 打印调试信息

    print(f"  log π(chosen)  ={lp_chosen.item():.4f}  "

          f"ref={ref_lp_chosen.item():.4f}  "

          f"implicit_reward={chosen_reward.item():.4f}")

    print(f"  log π(rejected)={lp_rejected.item():.4f}  "

          f"ref={ref_lp_rejected.item():.4f}  "

          f"implicit_reward={rejected_reward.item():.4f}")

    print(f"  DPO loss={loss.item():.4f}")

    return loss

 

def run_dpo_pipeline(model_name: str = "gpt2"):

    print("\n" + "═"*60)

    print("  DPO 训练流程")

    print("═"*60)

    tokenizer = AutoTokenizer.from_pretrained(model_name)

    tokenizer.pad_token = tokenizer.eos_token

    # ── Step 1: SFT（同 RLHF，跳过）────────────────────────────

    print("\n[Step 1] SFT: 在示范数据上微调（此处跳过）")

    ref_model    = AutoModelForCausalLM.from_pretrained(model_name)

    policy_model = AutoModelForCausalLM.from_pretrained(model_name)

    ref_model.eval()

    optimizer = AdamW(policy_model.parameters(), lr=1e-6)

    # ── Step 2: DPO 直接优化 ─────────────────────────────────────

    print("\n[Step 2] DPO 直接优化（无需奖励模型）…")

    item = PREFERENCE_DATA[0]

    for epoch in range(3):

        print(f"\n  Epoch {epoch+1}/3")

        policy_model.train()

        loss = dpo_loss(

            policy_model, ref_model, tokenizer,

            prompt=item["prompt"],

            chosen=item["chosen"],

            rejected=item["rejected"],

            beta=0.1,

        )

        optimizer.zero_grad()

        loss.backward()

        optimizer.step()

    print("\n✓ DPO 完成\n")

    return policy_model

 

# ══════════════════════════════════════════════════════════════════

#  主入口

# ══════════════════════════════════════════════════════════════════

if __name__ == "__main__":

    # 使用 gpt2 作为玩具模型演示（替换为你的中文模型即可）

    MODEL_NAME = "gpt2"

    rlhf_model = run_rlhf_pipeline(MODEL_NAME)

    dpo_model  = run_dpo_pipeline(MODEL_NAME)

    print("两种方法均已完成训练。")

    print("替换 MODEL_NAME 为中文 LLM（如 Qwen/Qwen2-1.5B）可直接运行。")