清远建设局网站,安阳吧百度贴吧,公众号文章模板素材,云南省工程建设信息网站GPT系列模型的演进
chatgpt系列模型演进的重要节点包含下面几个模型#xff08;当然#xff0c;这两年模型发展太快了#xff0c;4o这些推理模型我就先不写了#xff09; (Transformer) → GPT-1 → GPT-2 → GPT-3 → InstructGPT/ChatGPT(GPT-3.5) → GPT-4 下面介绍一…GPT系列模型的演进
chatgpt系列模型演进的重要节点包含下面几个模型当然这两年模型发展太快了4o这些推理模型我就先不写了 (Transformer) → GPT-1 → GPT-2 → GPT-3 → InstructGPT/ChatGPT(GPT-3.5) → GPT-4 下面介绍一下各个模型之前的重点差异
(1)Transformer2017
定位NLP基础架构革命奠定GPT系列技术底座核心创新 多头自注意力机制替代RNN/CNN解决长距离依赖问题位置编码通过正弦函数或可学习向量表征序列位置并行计算架构突破序列处理的效率瓶颈 局限未形成完整生成模型需配合任务微调
(2)GPT-12018
技术定位首个基于Transformer的生成式预训练模型核心改进 单向掩码机制仅允许左向注意力实现自回归文本生成两阶段训练无监督预训练BooksCorpus 下游任务微调参数规模1.17亿 应用场景文本续写、简单问答
(3)GPT-22019
技术跃迁验证规模扩展零样本学习可行性核心改进 参数爆炸15亿参数较GPT-1增长12.8倍零样本迁移无需微调即可完成翻译、摘要等任务WebText数据集800万网页数据提升多样性 局限生成文本存在重复和不连贯现象
(4)GPT-32020
技术突破定义大模型即服务范式核心创新 超大规模参数1750亿参数开启千亿级模型时代上下文学习通过Prompt工程实现少样本/单样本学习混合训练数据融合Common Crawl、书籍、维基百科等 局限存在事实性错误和伦理风险
(5)InstructGPT/ChatGPTGPT-3.5, 2022
技术定位首个实现人类对齐的对话模型核心改进 三阶段对齐流程SFT → RM → PPORLHF技术通过人类反馈强化学习优化输出安全性指令微调使用人工标注指令-答案对提升任务理解参数规模保持1750亿参数但训练数据量扩展
(6)GPT-42023
技术革命多模态超智能体架构核心创新 多模态处理支持图像输入与文本生成联动混合专家模型推测采用MoE架构参数达1.8万亿动态推理优化思维链CoT增强复杂问题解决能力安全增强毒性输出较GPT-3.5降低50%以上工程突破32K上下文窗口支持长文档处理
可以见的从 GPT1 到 GPT3最主要的技术进步就是参数量和预训练数据的扩大这也验证可 Scale Law规模法则即模型性能随模型规模的增长而提高 而从 GPT3 到可以直接对话的 chatgpt对齐优化则是最重要的突破RLHF作为对齐阶段的里程碑技术通过三阶段流程SFT→RM→PPO将模型输出与人类偏好深度绑定标志着语言模型从“通用生成”到“可控服务”的范式转变。 下面 图片就演示了 RLHF而本篇文章重点讲解一下 RLHFReinforcement Learning from Human Feedback基于人类反馈的强化学习
RLHF讲解与实战
整体流程介绍 ChatGPT是怎么变聪明的 想象一下ChatGPT一开始就像个刚学说话的小孩虽然懂一些知识但回答得不太好。科学家们为了让它的回答更符合人类喜好用了三步训练法让它像打游戏升级一样越练越强
第1步先教它“标准答案”监督学习 方法从网上找一大堆问题和答案比如“怎么煮咖啡”让ChatGPT学习正确的回答方式。 结果它学会了基本的对话能力但还不够聪明回答可能很死板或者不讨喜。这时候的版本叫“弱弱的ChatGPT”。第2步教它“哪种回答更讨喜”奖励模型 方法让“弱弱的ChatGPT”对同一个问题生成多个答案比如回答“煮咖啡”时有的详细有的简短然后请人类给这些答案打分哪个更好哪个更差。 训练奖励模型用这些打分数据训练一个“评分AI”让它学会人类的喜好以后能自动给ChatGPT的回答打分。第3步让它“自己和自己比赛”强化学习 方法让“弱弱的ChatGPT”继续回答问题但这次用“评分AI”给它打分然后告诉它“这个回答得分高下次多这样答那个回答得分低下次别这样了。” 升级版ChatGPT经过反复调整它的回答越来越符合人类喜好变得更自然、更聪明。循环升级越练越强 升级后的ChatGPT可以重新训练“评分AI”因为它的回答更好了。 更好的“评分AI”又能帮ChatGPT进一步优化回答…… 这样循环训练就像武侠小说里的高手左右手互搏越练越厉害最终成为惊艳世界的ChatGPT
强化学习基础知识
强化学习Reinforcement Learning, RL 是一种通过试错学习实现目标的人工智能范式。其核心是智能体Agent在与环境Environment的交互中通过最大化累积奖励Reward来学习最优策略Policy。
关键要素 智能体Agent决策主体如机器人、游戏角色。环境Environment智能体交互的物理或虚拟世界。状态State环境的当前描述如迷宫中的位置。动作Action智能体可执行的操作如移动方向。奖励Reward环境对动作的即时反馈如到达终点100分撞墙-10分。策略Policy状态到动作的映射规则如“遇到障碍物时左转”。 与其他机器学习的区别 监督学习依赖标注数据输入-答案对优化目标是预测误差最小化。强化学习无需标注数据通过试错优化长期累积奖励注重延迟反馈如围棋中某一步可能在几十步后才决定胜负
在训练ChatGPT时OpenAI让一组人类评估者来评价模型的回答。这些评估者拿到了一组指导方针告诉他们什么样的回答应该被高度评价什么样的回答应该被低度评价。在评估者评价的过程中通过不断的试错和学习机器人试图找到一种策略使得在与用户交谈过程中获取的总奖励最大。这就是通过基于人类反馈的强化学习调优ChatGPT的基本思想。
PPO讲解
PPO 是一种强化学习算法专门用于让AI模型比如ChatGPT通过试错和反馈优化自己的策略即参数。它的核心思想是 “小步调整参数避免一次更新太大导致模型崩溃”就像健身时循序渐进而不是突然举100kg受伤。
PPO的输入
输入 当前模型策略比如“弱弱的ChatGPT”参数为θ。 奖励模型RM能给ChatGPT的回答打分比如1~10分。 - 输出 优化后的新模型参数θ’生成的回答更符合人类偏好。
PPO分步骤讲解 步骤1生成回答并打分 从Prompt库抽样一个问题比如“怎么煮咖啡”。 让当前ChatGPT生成多个答案比如答案A、B、C。 奖励模型RM给这些答案打分比如A7分B3分C5分 步骤2计算“优势”Advantage 关键问题当前回答比“平均水平”好多少 公式优势A 当前回答得分 - 平均预期得分 如果A7平均预期5 → 优势2鼓励这类回答。 如果B3平均预期5 → 优势-2抑制这类回答。 步骤3计算“策略比率”Policy Ratio 关键问题新参数θ’生成的回答和旧参数θ生成的回答概率相差多少 公式比率 Pθ’(回答) / Pθ(回答) 如果比率≈1新旧策略对回答的选择概率相似。 如果比率1新策略更倾向于生成该回答。 如果比率1新策略更抑制该回答。 步骤4PPO的核心目标函数 PPO通过以下公式调整参数θ’同时限制更新幅度避免突变
**目标函数 min(比率×优势, clip(比率, 1-ε, 1ε)×优势)**clip函数强制比率在[1-ε, 1ε]范围内比如ε0.2 → 比率限制在0.8~1.2。 如果比率1.5更新太大→ 被clip到1.2防止参数剧烈变化。 如果比率0.9安全范围→ 保持不变。
步骤5梯度下降更新参数 最大化目标函数 → 让高优势回答的概率增加低优势回答的概率降低。 但通过clip限制步长保证训练稳定。
PPO 的整体公式就是
L(θ) E[ min(ratio * Advantage, clip(ratio, 1-ε, 1ε) * Advantage) ] - β * Entropyratio Pθ_new(a|s) / Pθ_old(a|s)Entropy 是熵奖励鼓励生成多样性回答。
这个本身是一个相对复杂的算法建议看看网上更专业的讲解奥 推荐一下李宏毅老师的课程http://speech.ee.ntu.edu.tw/~tlkagk/courses_MLDS18.html
ppo代码示例
import torch
import torch.nn as nn
import torch.optim as optim
from torch.distributions import Categorical
import numpy as np
# 设置随机种子
torch.manual_seed(42)
np.random.seed(42)class LanguageModel(nn.Module):def __init__(self, vocab_size, embedding_dim128, hidden_dim256):super().__init__()self.embedding nn.Embedding(vocab_size, embedding_dim)self.lstm nn.LSTM(embedding_dim, hidden_dim, batch_firstTrue)self.fc nn.Linear(hidden_dim, vocab_size) # 输出每个词的概率def forward(self, x):# x: [batch_size, seq_len]x self.embedding(x) # [batch_size, seq_len, embedding_dim]lstm_out, _ self.lstm(x) # [batch_size, seq_len, hidden_dim]logits self.fc(lstm_out) # [batch_size, seq_len, vocab_size]return logitsdef generate(self, prompt, max_len20):生成文本类似ChatGPT的推理过程with torch.no_grad():tokens promptfor _ in range(max_len):logits self.forward(tokens) # [batch_size, seq_len, vocab_size]next_token_logits logits[:, -1, :] # 取最后一个词的logitsprobs torch.softmax(next_token_logits, dim-1)next_token torch.multinomial(probs, 1) # 按概率采样tokens torch.cat([tokens, next_token], dim1)return tokensclass RewardModel(nn.Module):def __init__(self, vocab_size, embedding_dim128, hidden_dim256):super().__init__()self.embedding nn.Embedding(vocab_size, embedding_dim)self.lstm nn.LSTM(embedding_dim, hidden_dim, batch_firstTrue)self.fc nn.Linear(hidden_dim, 1) # 输出单个奖励值def forward(self, x):# x: [batch_size, seq_len]x self.embedding(x) # [batch_size, seq_len, embedding_dim]lstm_out, _ self.lstm(x) # [batch_size, seq_len, hidden_dim]# 使用最后一个时间步的隐藏状态来预测奖励rewards self.fc(lstm_out[:, -1, :]) # [batch_size, 1]return rewards.squeeze(-1) # [batch_size]def ppo_train(actor_model, # 语言模型策略网络reward_model, # 奖励模型optimizer, # 优化器如Adamprompts, # 输入的prompt问题num_epochs10, # PPO训练轮数clip_epsilon0.2, # PPO的clip参数通常0.1~0.3gamma0.99, # 折扣因子batch_size32 # 每批数据量
):actor_model.train() # 切换到训练模式reward_model.eval() # 奖励模型设置为评估模式for epoch in range(num_epochs):# 1. 生成回答采样with torch.no_grad():responses actor_model.generate(prompts) # [batch_size, seq_len]# 计算旧策略的概率old_logits actor_model(responses) # [batch_size, seq_len, vocab_size]old_log_probs torch.log_softmax(old_logits, dim-1)old_log_probs old_log_probs.gather(-1, responses.unsqueeze(-1)).squeeze(-1) # [batch_size, seq_len]old_log_probs old_log_probs.mean(dim1) # 平均每个token的log_prob# 2. 计算奖励用奖励模型with torch.no_grad():rewards reward_model(responses) # [batch_size]# 3. 计算优势Advantage# 这里简化计算Advantage ≈ 归一化的Rewardadvantages (rewards - rewards.mean()) / (rewards.std() 1e-8)# 4. 计算新策略的概率new_logits actor_model(responses) # [batch_size, seq_len, vocab_size]new_log_probs torch.log_softmax(new_logits, dim-1)new_log_probs new_log_probs.gather(-1, responses.unsqueeze(-1)).squeeze(-1) # [batch_size, seq_len]new_log_probs new_log_probs.mean(dim1) # 平均每个token的log_prob# 5. 计算策略比率Policy Ratioratios torch.exp(new_log_probs - old_log_probs) # e^{log(π_new/π_old)}# 6. PPO 目标函数Clipped Surrogate Objectivesurr1 ratios * advantagessurr2 torch.clamp(ratios, 1 - clip_epsilon, 1 clip_epsilon) * advantagespolicy_loss -torch.min(surr1, surr2).mean() # 取min防止更新过大# 7. 计算熵正则化鼓励探索entropy Categorical(logitsnew_logits).entropy().mean()entropy_bonus 0.01 * entropy # 调节系数# 8. 总损失 Policy Loss - 熵奖励loss policy_loss - entropy_bonus# 9. 反向传播 优化optimizer.zero_grad()loss.backward()optimizer.step()print(fEpoch {epoch1}, Loss: {loss.item():.4f}, Avg Reward: {rewards.mean().item():.4f})# 设置参数
vocab_size 10000 # 词汇表大小
embedding_dim 128
hidden_dim 256# 创建模型
actor_model LanguageModel(vocab_size, embedding_dim, hidden_dim)
reward_model RewardModel(vocab_size, embedding_dim, hidden_dim)# 优化器
optimizer optim.Adam(actor_model.parameters(), lr1e-5)# 创建示例输入数据
batch_size 32
seq_len 5
prompts torch.randint(0, vocab_size, (batch_size, seq_len))# 开始PPO训练
ppo_train(actor_model, reward_model, optimizer, prompts, num_epochs10)简单RLHF实战
下面实现了一个完整的RLHF系统包含了完整的RLHF三阶段流程
预训练阶段训练基础语言模型奖励模型训练基于人类偏好数据训练奖励模型PPO强化学习使用PPO算法优化策略
学习思路即可 目前效果确实一般
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
import numpy as np
import random
from collections import deque
import json
import os
from typing import List, Dict, Tuple, Optional
import logging# 设置日志
logging.basicConfig(levellogging.INFO)
logger logging.getLogger(__name__)# 设置设备
device torch.device(cuda if torch.cuda.is_available() else cpu)
logger.info(f使用设备: {device})class SimpleTokenizer:简单的分词器def __init__(self, vocab_size5000):self.vocab_size vocab_sizeself.vocab {pad: 0, unk: 1, start: 2, end: 3}self.reverse_vocab {0: pad, 1: unk, 2: start, 3: end}def build_vocab(self, texts):构建词汇表word_freq {}for text in texts:words text.lower().split()for word in words:word_freq[word] word_freq.get(word, 0) 1# 按频率排序取前vocab_size-4个词sorted_words sorted(word_freq.items(), keylambda x: x[1], reverseTrue)for i, (word, _) in enumerate(sorted_words[:self.vocab_size-4]):idx i 4self.vocab[word] idxself.reverse_vocab[idx] word# 确保测试词在词汇表中test_words [今天, 天气, 人工, 智能, 学习, 健康, 工作]for word in test_words:if word not in self.vocab:idx len(self.vocab)self.vocab[word] idxself.reverse_vocab[idx] wordlogger.info(f构建词汇表完成词汇量: {len(self.vocab)})logger.info(f测试词是否在词汇表中: {[word in self.vocab for word in test_words]})def encode(self, text, max_length128):编码文本words text.lower().split()tokens [self.vocab.get(word, 1) for word in words] # 1是unk# 添加开始和结束标记tokens [2] tokens [3] # 2是start, 3是end# 填充或截断if len(tokens) max_length:tokens [0] * (max_length - len(tokens)) # 0是padelse:tokens tokens[:max_length]tokens[-1] 3 # 确保以end结尾return tokensdef decode(self, tokens):解码令牌words []for token in tokens:if token in [0, 2]: # 跳过pad和startcontinueif token 3: # endbreakwords.append(self.reverse_vocab.get(token, unk))return .join(words)class SimpleTransformer(nn.Module):简化的Transformer模型def __init__(self, vocab_size, d_model256, nhead8, num_layers4, max_length128):super().__init__()self.d_model d_modelself.vocab_size vocab_sizeself.max_length max_length# 嵌入层self.embedding nn.Embedding(vocab_size, d_model)self.pos_encoding nn.Parameter(torch.randn(max_length, d_model))# Transformer层encoder_layer nn.TransformerEncoderLayer(d_modeld_model, nheadnhead, dim_feedforwardd_model*4,dropout0.1, batch_firstTrue)self.transformer nn.TransformerEncoder(encoder_layer, num_layersnum_layers)# 输出头self.lm_head nn.Linear(d_model, vocab_size)# 初始化权重self._init_weights()def _init_weights(self):初始化权重for module in self.modules():if isinstance(module, nn.Linear):nn.init.normal_(module.weight, mean0.0, std0.02)if module.bias is not None:nn.init.zeros_(module.bias)elif isinstance(module, nn.Embedding):nn.init.normal_(module.weight, mean0.0, std0.02)def forward(self, input_ids, attention_maskNone):seq_length input_ids.size(1)# 嵌入和位置编码embeddings self.embedding(input_ids)# 确保位置编码维度匹配pos_enc self.pos_encoding[:seq_length].unsqueeze(0).expand(embeddings.size(0), -1, -1)embeddings embeddings pos_enc# 注意力掩码if attention_mask is None:attention_mask (input_ids ! 0).float()# Transformermask (attention_mask 0)hidden_states self.transformer(embeddings, src_key_padding_maskmask)# 语言模型头logits self.lm_head(hidden_states)return logitsclass RewardModel(nn.Module):奖励模型def __init__(self, vocab_size, d_model256, nhead8, num_layers4, max_length128):super().__init__()self.d_model d_model# 嵌入层self.embedding nn.Embedding(vocab_size, d_model)self.pos_encoding nn.Parameter(torch.randn(max_length, d_model))# Transformer层encoder_layer nn.TransformerEncoderLayer(d_modeld_model, nheadnhead, dim_feedforwardd_model*4,dropout0.1, batch_firstTrue)self.transformer nn.TransformerEncoder(encoder_layer, num_layersnum_layers)# 奖励头self.reward_head nn.Sequential(nn.Linear(d_model, d_model // 2),nn.ReLU(),nn.Linear(d_model // 2, 1))def forward(self, input_ids, attention_maskNone):seq_length input_ids.size(1)# 嵌入和位置编码embeddings self.embedding(input_ids)# 确保位置编码维度匹配pos_enc self.pos_encoding[:seq_length].unsqueeze(0).expand(embeddings.size(0), -1, -1)embeddings embeddings pos_enc# 注意力掩码if attention_mask is None:attention_mask (input_ids ! 0).float()# Transformermask (attention_mask 0)hidden_states self.transformer(embeddings, src_key_padding_maskmask)# 使用最后一个非填充位置的隐藏状态batch_size input_ids.size(0)# 计算每个序列的实际长度非padding token的数量seq_lengths attention_mask.sum(dim1).long()# 避免索引为0的情况至少为1last_positions torch.clamp(seq_lengths - 1, min0)# 使用torch.arange确保索引类型正确batch_indices torch.arange(batch_size, deviceinput_ids.device)last_hidden hidden_states[batch_indices, last_positions]# 奖励分数reward self.reward_head(last_hidden).squeeze(-1)return rewardclass PreferenceDataset(Dataset):偏好数据集def __init__(self, preferences, tokenizer, max_length128):self.preferences preferencesself.tokenizer tokenizerself.max_length max_lengthdef __len__(self):return len(self.preferences)def __getitem__(self, idx):pref self.preferences[idx]prompt pref[prompt]chosen pref[chosen]rejected pref[rejected]# 编码prompt_tokens self.tokenizer.encode(prompt, self.max_length)chosen_tokens self.tokenizer.encode(prompt chosen, self.max_length)rejected_tokens self.tokenizer.encode(prompt rejected, self.max_length)return {prompt: torch.tensor(prompt_tokens, dtypetorch.long),chosen: torch.tensor(chosen_tokens, dtypetorch.long),rejected: torch.tensor(rejected_tokens, dtypetorch.long)}class PPOTrainer:PPO训练器def __init__(self, policy_model, reward_model, tokenizer, lr1e-4):self.policy_model policy_modelself.reward_model reward_modelself.tokenizer tokenizer# 创建参考模型冻结的策略模型副本self.ref_model SimpleTransformer(vocab_sizepolicy_model.vocab_size,d_modelpolicy_model.d_model,max_lengthpolicy_model.max_length).to(device)self.ref_model.load_state_dict(policy_model.state_dict())self.ref_model.eval()# 优化器self.optimizer optim.Adam(policy_model.parameters(), lrlr)# PPO参数self.clip_ratio 0.2self.kl_coeff 0.02self.value_coeff 0.1self.entropy_coeff 0.01def generate_response(self, prompt_tokens, max_new_tokens50, temperature0.8):生成响应self.policy_model.eval()with torch.no_grad():input_ids prompt_tokens.clone()for _ in range(max_new_tokens):logits self.policy_model(input_ids)next_token_logits logits[:, -1, :] / temperatureprobs F.softmax(next_token_logits, dim-1)next_token torch.multinomial(probs, 1)input_ids torch.cat([input_ids, next_token], dim-1)# 检查是否所有序列都生成了结束标记if (next_token 3).all(): # endbreak# 防止序列过长if input_ids.size(1) self.policy_model.max_length:breakreturn input_idsdef compute_rewards(self, sequences):计算奖励self.reward_model.eval()with torch.no_grad():rewards self.reward_model(sequences)return rewardsdef compute_log_probs(self, model, sequences, actions):计算动作的对数概率# 确保序列长度足够if sequences.size(1) 1:return torch.zeros(actions.size(), deviceactions.device)logits model(sequences[:, :-1]) # 不包括最后一个tokenlog_probs F.log_softmax(logits, dim-1)# 获取动作的对数概率action_log_probs log_probs.gather(2, actions.unsqueeze(-1)).squeeze(-1)return action_log_probsdef ppo_step(self, prompts, responses, rewards, old_log_probs):PPO更新步骤self.policy_model.train()# 检查响应是否为空if responses.size(1) 0:return {policy_loss: 0.0,kl_penalty: 0.0,entropy: 0.0,total_loss: 0.0}# 计算新的对数概率full_sequences torch.cat([prompts, responses], dim1)new_log_probs self.compute_log_probs(self.policy_model, full_sequences, responses)# 计算参考模型的对数概率用于KL散度with torch.no_grad():ref_log_probs self.compute_log_probs(self.ref_model, full_sequences, responses)# 计算比率ratio torch.exp(new_log_probs - old_log_probs)# PPO裁剪损失advantages rewards.unsqueeze(-1) - rewards.mean()clipped_ratio torch.clamp(ratio, 1 - self.clip_ratio, 1 self.clip_ratio)policy_loss -torch.min(ratio * advantages, clipped_ratio * advantages).mean()# KL散度惩罚kl_penalty self.kl_coeff * (new_log_probs - ref_log_probs).mean()# 熵奖励entropy -(torch.exp(new_log_probs) * new_log_probs).sum(-1).mean()entropy_bonus self.entropy_coeff * entropy# 总损失total_loss policy_loss kl_penalty - entropy_bonus# 反向传播self.optimizer.zero_grad()total_loss.backward()torch.nn.utils.clip_grad_norm_(self.policy_model.parameters(), 1.0)self.optimizer.step()return {policy_loss: policy_loss.item(),kl_penalty: kl_penalty.item(),entropy: entropy.item(),total_loss: total_loss.item()}class RLHFTrainer:完整的RLHF训练器def __init__(self, vocab_size5000, d_model256, max_length128):self.vocab_size vocab_sizeself.d_model d_modelself.max_length max_length# 初始化组件self.tokenizer SimpleTokenizer(vocab_size)self.policy_model SimpleTransformer(vocab_size, d_model, max_lengthmax_length).to(device)self.reward_model RewardModel(vocab_size, d_model, max_lengthmax_length).to(device)logger.info(RLHF训练器初始化完成)def prepare_data(self):准备示例数据 - 增加更多样化的数据# 预训练数据 - 增加数据量和多样性pretrain_texts [今天天气很好阳光明媚适合出门散步,我喜欢学习新知识这让我感到充实和快乐,人工智能很有趣它正在改变我们的生活,编程是一门艺术需要逻辑思维和创造力,音乐让人放松能够舒缓心情和压力,读书使人进步开拓视野增长见识,运动有益健康保持身体活力和精神状态,美食令人愉悦带来味觉和心灵的享受,旅行开阔视野体验不同文化和风景,友谊珍贵无比真挚的友情值得珍惜,工作需要专注认真负责才能做好,家庭很重要亲情是最温暖的港湾,创新推动发展新思路带来新机遇,学习永无止境持续进步才能成长,沟通很关键理解彼此才能合作,时间很宝贵珍惜当下把握机会,健康最重要身体是革命的本钱,梦想值得追求坚持努力终会实现,思考很重要深度思考带来智慧,合作共赢好团结协作力量大] * 20 # 增加重复次数# 偏好数据 - 增加数据量preferences [{prompt: 今天天气,chosen: 今天天气很好阳光明媚适合出门游玩和散步,rejected: 今天天气不好下雨了},{prompt: 学习的意义,chosen: 学习能够丰富知识提升能力让人变得更加智慧,rejected: 学习很累很困难},{prompt: 人工智能,chosen: 人工智能是未来科技发展的重要方向将改变生活,rejected: 人工智能很难懂很复杂},{prompt: 工作态度,chosen: 认真负责的工作态度是成功的关键因素,rejected: 工作很辛苦很累},{prompt: 健康生活,chosen: 健康的生活方式包括运动、营养和良好作息,rejected: 健康生活很难坚持},{prompt: 友谊价值,chosen: 真挚的友谊珍贵无比朋友间相互支持很重要,rejected: 朋友关系很复杂}] * 10 # 增加重复次数return pretrain_texts, preferencesdef pretrain(self, texts, epochs20, batch_size8, lr1e-3):预训练阶段 - 增加训练轮数logger.info(开始预训练...)# 构建词汇表self.tokenizer.build_vocab(texts)# 准备数据 - 使用与模型一致的最大长度encoded_texts [self.tokenizer.encode(text, self.max_length) for text in texts]dataset torch.utils.data.TensorDataset(torch.tensor(encoded_texts, dtypetorch.long))dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue)# 优化器 - 调整学习率optimizer optim.Adam(self.policy_model.parameters(), lrlr, weight_decay1e-5)criterion nn.CrossEntropyLoss(ignore_index0) # 忽略padding# 学习率调度器scheduler optim.lr_scheduler.CosineAnnealingLR(optimizer, T_maxepochs)self.policy_model.train()for epoch in range(epochs):total_loss 0num_batches 0for batch in dataloader:input_ids batch[0].to(device)# 前向传播logits self.policy_model(input_ids)# 计算损失预测下一个tokenshift_logits logits[:, :-1, :].contiguous()shift_labels input_ids[:, 1:].contiguous()loss criterion(shift_logits.view(-1, self.vocab_size), shift_labels.view(-1))# 反向传播optimizer.zero_grad()loss.backward()torch.nn.utils.clip_grad_norm_(self.policy_model.parameters(), 1.0)optimizer.step()total_loss loss.item()num_batches 1scheduler.step()avg_loss total_loss / num_batcheslogger.info(f预训练 Epoch {epoch1}/{epochs}, Loss: {avg_loss:.4f}, LR: {scheduler.get_last_lr()[0]:.6f})logger.info(预训练完成)def train_reward_model(self, preferences, epochs30, batch_size4, lr1e-4):训练奖励模型 - 增加训练轮数logger.info(开始训练奖励模型...)dataset PreferenceDataset(preferences, self.tokenizer, self.max_length)dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue)optimizer optim.Adam(self.reward_model.parameters(), lrlr, weight_decay1e-5)self.reward_model.train()for epoch in range(epochs):total_loss 0correct 0total 0for batch in dataloader:chosen_ids batch[chosen].to(device)rejected_ids batch[rejected].to(device)# 计算奖励分数chosen_rewards self.reward_model(chosen_ids)rejected_rewards self.reward_model(rejected_ids)# 排序损失chosen应该比rejected得分更高loss -torch.log(torch.sigmoid(chosen_rewards - rejected_rewards)).mean()# 反向传播optimizer.zero_grad()loss.backward()torch.nn.utils.clip_grad_norm_(self.reward_model.parameters(), 1.0)optimizer.step()total_loss loss.item()# 计算准确率correct (chosen_rewards rejected_rewards).sum().item()total chosen_rewards.size(0)avg_loss total_loss / len(dataloader)accuracy correct / totallogger.info(f奖励模型 Epoch {epoch1}/{epochs}, Loss: {avg_loss:.4f}, Acc: {accuracy:.4f})logger.info(奖励模型训练完成)def rl_training(self, prompts, epochs15, batch_size2):强化学习训练阶段 - 增加训练轮数logger.info(开始PPO强化学习训练...)ppo_trainer PPOTrainer(self.policy_model, self.reward_model, self.tokenizer, lr5e-5)# 编码提示encoded_prompts [self.tokenizer.encode(prompt, self.max_length//2) for prompt in prompts]prompt_dataset torch.utils.data.TensorDataset(torch.tensor(encoded_prompts, dtypetorch.long))prompt_dataloader DataLoader(prompt_dataset, batch_sizebatch_size, shuffleTrue)for epoch in range(epochs):epoch_stats {policy_loss: 0, kl_penalty: 0, entropy: 0, total_loss: 0}num_batches 0for batch in prompt_dataloader:prompt_tokens batch[0].to(device)# 生成响应with torch.no_grad():full_sequences ppo_trainer.generate_response(prompt_tokens, max_new_tokens20)# 确保响应序列不为空if full_sequences.size(1) prompt_tokens.size(1):continueresponses full_sequences[:, prompt_tokens.size(1):]# 计算奖励rewards ppo_trainer.compute_rewards(full_sequences)# 计算旧的对数概率old_log_probs ppo_trainer.compute_log_probs(ppo_trainer.policy_model, full_sequences, responses)# PPO更新stats ppo_trainer.ppo_step(prompt_tokens, responses, rewards, old_log_probs)for key in epoch_stats:epoch_stats[key] stats[key]num_batches 1# 记录平均统计if num_batches 0:for key in epoch_stats:epoch_stats[key] / num_batcheslogger.info(fPPO Epoch {epoch1}/{epochs}: {epoch_stats})else:logger.info(fPPO Epoch {epoch1}/{epochs}: 没有有效的批次)logger.info(PPO训练完成)def generate_text(self, prompt, max_length30, temperature0.7, top_k50):生成文本 - 改进解码策略self.policy_model.eval()# 确保提示词在词汇表中prompt_tokens torch.tensor([self.tokenizer.encode(prompt, self.max_length//2)], dtypetorch.long).to(device)with torch.no_grad():input_ids prompt_tokens.clone()generated_tokens []for _ in range(max_length):# 获取模型预测logits self.policy_model(input_ids)next_token_logits logits[:, -1, :] / temperature# 应用top-k过滤if top_k 0:indices_to_remove next_token_logits torch.topk(next_token_logits, top_k)[0][..., -1, None]next_token_logits[indices_to_remove] -float(Inf)# 计算概率分布probs F.softmax(next_token_logits, dim-1)# 采样下一个tokennext_token torch.multinomial(probs, 1)# 如果生成了结束标记停止生成if next_token.item() 3: # endbreak# 如果生成了未知标记尝试重新采样if next_token.item() 1: # unkcontinue# 添加新生成的tokeninput_ids torch.cat([input_ids, next_token], dim-1)generated_tokens.append(next_token.item())# 检查序列长度if input_ids.size(1) self.max_length:break# 如果没有生成任何token返回原始提示if not generated_tokens:return prompt# 解码生成的文本generated_text self.tokenizer.decode(generated_tokens)# 如果生成了空文本返回原始提示if not generated_text.strip():return promptreturn prompt generated_textdef save_models(self, save_dir./rlhf_models):保存模型os.makedirs(save_dir, exist_okTrue)torch.save(self.policy_model.state_dict(), os.path.join(save_dir, policy_model.pt))torch.save(self.reward_model.state_dict(), os.path.join(save_dir, reward_model.pt))# 保存分词器with open(os.path.join(save_dir, tokenizer.json), w, encodingutf-8) as f:json.dump({vocab: self.tokenizer.vocab,reverse_vocab: self.tokenizer.reverse_vocab}, f, ensure_asciiFalse, indent2)logger.info(f模型已保存到 {save_dir})def load_models(self, save_dir./rlhf_models):加载模型self.policy_model.load_state_dict(torch.load(os.path.join(save_dir, policy_model.pt), map_locationdevice))self.reward_model.load_state_dict(torch.load(os.path.join(save_dir, reward_model.pt), map_locationdevice))# 加载分词器with open(os.path.join(save_dir, tokenizer.json), r, encodingutf-8) as f:tokenizer_data json.load(f)self.tokenizer.vocab tokenizer_data[vocab]self.tokenizer.reverse_vocab {int(k): v for k, v in tokenizer_data[reverse_vocab].items()}logger.info(f模型已从 {save_dir} 加载)def main():主函数 - 完整的RLHF训练流程logger.info(开始完整的RLHF训练流程)# 初始化训练器 - 调整参数trainer RLHFTrainer(vocab_size2000, d_model128, max_length64)# 准备数据pretrain_texts, preferences trainer.prepare_data()# 阶段1预训练trainer.pretrain(pretrain_texts, epochs15, batch_size4)# 阶段2训练奖励模型trainer.train_reward_model(preferences, epochs20, batch_size2)# 阶段3PPO强化学习prompts [今天天气, 学习的意义, 人工智能, 编程语言, 音乐的魅力, 健康生活, 工作态度]trainer.rl_training(prompts, epochs10, batch_size2)# 测试生成logger.info(\n 生成测试 )test_prompts [今天天气, 人工智能, 学习, 健康, 工作]for prompt in test_prompts:generated trainer.generate_text(prompt, max_length20)logger.info(f提示: {prompt})logger.info(f生成: {generated})logger.info(- * 50)# 保存模型trainer.save_models()logger.info(RLHF训练流程完成)if __name__ __main__:main()
