GPT系列模型的演進

chatgpt系列模型演進的重要節點包含下面幾個模型（當然，這兩年模型發展太快了，4o這些推理模型我就先不寫了）
(Transformer) → GPT-1 → GPT-2 → GPT-3 → InstructGPT/ChatGPT(GPT-3.5) → GPT-4

下面介紹一下各個模型之前的重點差異

(1)Transformer（2017）

• 定位：NLP基礎架構革命，奠定GPT系列技術底座
• 核心創新：
  • 多頭自注意力機制：替代RNN/CNN，解決長距離依賴問題
  • 位置編碼：通過正弦函數或可學習向量表征序列位置
  • 并行計算架構：突破序列處理的效率瓶頸
• 局限：未形成完整生成模型，需配合任務微調

(2)GPT-1（2018）

• 技術定位：首個基于Transformer的生成式預訓練模型
• 核心改進：
  • 單向掩碼機制：僅允許左向注意力，實現自回歸文本生成
  • 兩階段訓練：無監督預訓練（BooksCorpus）+ 下游任務微調
  • 參數規模：1.17億
• 應用場景：文本續寫、簡單問答

(3)GPT-2（2019）

• 技術躍遷：驗證"規模擴展+零樣本學習"可行性
• 核心改進：
  • 參數爆炸：15億參數，較GPT-1增長12.8倍
  • 零樣本遷移：無需微調即可完成翻譯、摘要等任務
  • WebText數據集：800萬網頁數據提升多樣性
• 局限：生成文本存在重復和不連貫現象

(4)GPT-3（2020）

• 技術突破：定義"大模型即服務"范式
• 核心創新：
  • 超大規模參數：1750億參數，開啟千億級模型時代
  • 上下文學習：通過Prompt工程實現少樣本/單樣本學習
  • 混合訓練數據：融合Common Crawl、書籍、維基百科等
• 局限：存在事實性錯誤和倫理風險

(5)InstructGPT/ChatGPT（GPT-3.5, 2022）

• 技術定位：首個實現人類對齊的對話模型
• 核心改進：
  • 三階段對齊流程：SFT → RM → PPO
  • RLHF技術：通過人類反饋強化學習優化輸出安全性
  • 指令微調：使用人工標注指令-答案對提升任務理解
  • 參數規模：保持1750億參數，但訓練數據量擴展

(6)GPT-4（2023）

• 技術革命：多模態+超智能體架構
• 核心創新：
  • 多模態處理：支持圖像輸入與文本生成聯動
  • 混合專家模型：推測采用MoE架構，參數達1.8萬億
  • 動態推理優化：思維鏈（CoT）增強復雜問題解決能力
  • 安全增強：毒性輸出較GPT-3.5降低50%以上
  • 工程突破：32K上下文窗口支持長文檔處理

可以見的，從 GPT1 到 GPT3，最主要的技術進步就是參數量和預訓練數據的擴大，這也驗證可 Scale Law（規模法則）即模型性能隨模型規模的增長而提高
而從 GPT3 到可以直接對話的 chatgpt，對齊優化則是最重要的突破，RLHF作為對齊階段的里程碑技術，通過三階段流程（SFT→RM→PPO）將模型輸出與人類偏好深度綁定，標志著語言模型從“通用生成”到“可控服務”的范式轉變。
下面 圖片就演示了 RLHF，而本篇文章重點講解一下 RLHF（Reinforcement 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給這些答案打分（比如A=7分，B=3分，C=5分）

• 步驟2：計算“優勢”（Advantage）??
  ??關鍵問題??：當前回答比“平均水平”好多少？
  ??公式??：優勢A = 當前回答得分 - 平均預期得分
  如果A=7，平均預期=5 → 優勢=+2（鼓勵這類回答）。
  如果B=3，平均預期=5 → 優勢=-2（抑制這類回答）。

• 步驟3：計算“策略比率”（Policy Ratio）??
  ??關鍵問題??：新參數θ’生成的回答，和舊參數θ生成的回答概率相差多少？
  ??公式??：比率 = Pθ’(回答) / Pθ(回答)
  如果比率≈1：新舊策略對回答的選擇概率相似。
  如果比率>1：新策略更傾向于生成該回答。
  如果比率<1：新策略更抑制該回答。

• 步驟4：PPO的核心目標函數??
  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) ] - β * Entropy

• ratio = 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_dim=128, hidden_dim=256):super().__init__()self.embedding = nn.Embedding(vocab_size, embedding_dim)self.lstm = nn.LSTM(embedding_dim, hidden_dim, batch_first=True)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_len=20):"""生成文本（類似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], dim=1)return tokensclass RewardModel(nn.Module):def __init__(self, vocab_size, embedding_dim=128, hidden_dim=256):super().__init__()self.embedding = nn.Embedding(vocab_size, embedding_dim)self.lstm = nn.LSTM(embedding_dim, hidden_dim, batch_first=True)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,            # 優化器（如Adam）prompts,              # 輸入的prompt（問題）num_epochs=10,        # PPO訓練輪數clip_epsilon=0.2,     # PPO的clip參數（通常0.1~0.3）gamma=0.99,           # 折扣因子batch_size=32         # 每批數據量
):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(dim=1)  # 平均每個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(dim=1)  # 平均每個token的log_prob# 5. 計算策略比率（Policy Ratio）ratios = torch.exp(new_log_probs - old_log_probs)  # e^{log(π_new/π_old)}# 6. PPO 目標函數（Clipped Surrogate Objective）surr1 = ratios * advantagessurr2 = torch.clamp(ratios, 1 - clip_epsilon, 1 + clip_epsilon) * advantagespolicy_loss = -torch.min(surr1, surr2).mean()  # 取min防止更新過大# 7. 計算熵正則化（鼓勵探索）entropy = Categorical(logits=new_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(f"Epoch {epoch+1}, 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(), lr=1e-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_epochs=10)

簡單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(level=logging.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_size=5000):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(), key=lambda x: x[1], reverse=True)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_length=128):"""編碼文本"""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是<pad>else:tokens = tokens[:max_length]tokens[-1] = 3  # 確保以<end>結尾return tokensdef decode(self, tokens):"""解碼令牌"""words = []for token in tokens:if token in [0, 2]:  # 跳過<pad>和<start>continueif token == 3:  # <end>breakwords.append(self.reverse_vocab.get(token, "<unk>"))return " ".join(words)class SimpleTransformer(nn.Module):"""簡化的Transformer模型"""def __init__(self, vocab_size, d_model=256, nhead=8, num_layers=4, max_length=128):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_model=d_model, nhead=nhead, dim_feedforward=d_model*4,dropout=0.1, batch_first=True)self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_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, mean=0.0, std=0.02)if module.bias is not None:nn.init.zeros_(module.bias)elif isinstance(module, nn.Embedding):nn.init.normal_(module.weight, mean=0.0, std=0.02)def forward(self, input_ids, attention_mask=None):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_mask=mask)# 語言模型頭logits = self.lm_head(hidden_states)return logitsclass RewardModel(nn.Module):"""獎勵模型"""def __init__(self, vocab_size, d_model=256, nhead=8, num_layers=4, max_length=128):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_model=d_model, nhead=nhead, dim_feedforward=d_model*4,dropout=0.1, batch_first=True)self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_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_mask=None):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_mask=mask)# 使用最后一個非填充位置的隱藏狀態batch_size = input_ids.size(0)# 計算每個序列的實際長度（非padding token的數量）seq_lengths = attention_mask.sum(dim=1).long()# 避免索引為0的情況，至少為1last_positions = torch.clamp(seq_lengths - 1, min=0)# 使用torch.arange確保索引類型正確batch_indices = torch.arange(batch_size, device=input_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_length=128):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, dtype=torch.long),"chosen": torch.tensor(chosen_tokens, dtype=torch.long),"rejected": torch.tensor(rejected_tokens, dtype=torch.long)}class PPOTrainer:"""PPO訓練器"""def __init__(self, policy_model, reward_model, tokenizer, lr=1e-4):self.policy_model = policy_modelself.reward_model = reward_modelself.tokenizer = tokenizer# 創建參考模型（凍結的策略模型副本）self.ref_model = SimpleTransformer(vocab_size=policy_model.vocab_size,d_model=policy_model.d_model,max_length=policy_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(), lr=lr)# 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_tokens=50, temperature=0.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():  # <end>break# 防止序列過長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(), device=actions.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], dim=1)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_size=5000, d_model=256, max_length=128):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_length=max_length).to(device)self.reward_model = RewardModel(vocab_size, d_model, max_length=max_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, epochs=20, batch_size=8, lr=1e-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, dtype=torch.long))dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)# 優化器 - 調整學習率optimizer = optim.Adam(self.policy_model.parameters(), lr=lr, weight_decay=1e-5)criterion = nn.CrossEntropyLoss(ignore_index=0)  # 忽略padding# 學習率調度器scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=epochs)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)# 計算損失（預測下一個token）shift_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 {epoch+1}/{epochs}, Loss: {avg_loss:.4f}, LR: {scheduler.get_last_lr()[0]:.6f}")logger.info("預訓練完成")def train_reward_model(self, preferences, epochs=30, batch_size=4, lr=1e-4):"""訓練獎勵模型 - 增加訓練輪數"""logger.info("開始訓練獎勵模型...")dataset = PreferenceDataset(preferences, self.tokenizer, self.max_length)dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)optimizer = optim.Adam(self.reward_model.parameters(), lr=lr, weight_decay=1e-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 {epoch+1}/{epochs}, Loss: {avg_loss:.4f}, Acc: {accuracy:.4f}")logger.info("獎勵模型訓練完成")def rl_training(self, prompts, epochs=15, batch_size=2):"""強化學習訓練階段 - 增加訓練輪數"""logger.info("開始PPO強化學習訓練...")ppo_trainer = PPOTrainer(self.policy_model, self.reward_model, self.tokenizer, lr=5e-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, dtype=torch.long))prompt_dataloader = DataLoader(prompt_dataset, batch_size=batch_size, shuffle=True)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_tokens=20)# 確保響應序列不為空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(f"PPO Epoch {epoch+1}/{epochs}: {epoch_stats}")else:logger.info(f"PPO Epoch {epoch+1}/{epochs}: 沒有有效的批次")logger.info("PPO訓練完成")def generate_text(self, prompt, max_length=30, temperature=0.7, top_k=50):"""生成文本 - 改進解碼策略"""self.policy_model.eval()# 確保提示詞在詞匯表中prompt_tokens = torch.tensor([self.tokenizer.encode(prompt, self.max_length//2)], dtype=torch.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:  # <end>break# 如果生成了未知標記，嘗試重新采樣if next_token.item() == 1:  # <unk>continue# 添加新生成的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_ok=True)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", encoding="utf-8") as f:json.dump({"vocab": self.tokenizer.vocab,"reverse_vocab": self.tokenizer.reverse_vocab}, f, ensure_ascii=False, indent=2)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_location=device))self.reward_model.load_state_dict(torch.load(os.path.join(save_dir, "reward_model.pt"), map_location=device))# 加載分詞器with open(os.path.join(save_dir, "tokenizer.json"), "r", encoding="utf-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_size=2000, d_model=128, max_length=64)# 準備數據pretrain_texts, preferences = trainer.prepare_data()# 階段1：預訓練trainer.pretrain(pretrain_texts, epochs=15, batch_size=4)# 階段2：訓練獎勵模型trainer.train_reward_model(preferences, epochs=20, batch_size=2)# 階段3：PPO強化學習prompts = ["今天天氣", "學習的意義", "人工智能", "編程語言", "音樂的魅力", "健康生活", "工作態度"]trainer.rl_training(prompts, epochs=10, batch_size=2)# 測試生成logger.info("\n=== 生成測試 ===")test_prompts = ["今天天氣", "人工智能", "學習", "健康", "工作"]for prompt in test_prompts:generated = trainer.generate_text(prompt, max_length=20)logger.info(f"提示: {prompt}")logger.info(f"生成: {generated}")logger.info("-" * 50)# 保存模型trainer.save_models()logger.info("RLHF訓練流程完成！")if __name__ == "__main__":main()