强化学习是人工智能领域最激动人心的方向之一,它让智能体通过与环境的交互来学习最优策略。从AlphaGo到ChatGPT的RLHF(基于人类反馈的强化学习),强化学习正在改变AI的发展轨迹。本文将带你从零开始掌握强化学习的核心概念和实用算法。
强化学习的五个核心要素:
MDP是强化学习的数学框架:
MDP = (S, A, P, R, γ) - S:状态空间(State Space) - A:动作空间(Action Space) - P:状态转移概率(Transition Probability) - R:奖励函数(Reward Function) - γ:折扣因子(Discount Factor)
智能体的目标是找到最优策略π*,使得累积奖励最大化:
Gt = Rt+1 + γRt+2 + γ²Rt+3 + ... = Σ γ^k * Rt+k+1
最基础的value-based算法。
**核心思想:**学习Q函数,表示在状态s下采取动作a的期望回报。
更新规则:
import numpy as np class QLearning: def __init__(self, state_size, action_size, learning_rate=0.1, discount_factor=0.95, epsilon=0.1): self.state_size = state_size self.action_size = action_size self.learning_rate = learning_rate self.discount_factor = discount_factor self.epsilon = epsilon # 探索率 self.q_table = np.zeros((state_size, action_size)) def choose_action(self, state): # ε-贪心策略 if np.random.random() < self.epsilon: return np.random.randint(self.action_size) # 探索 else: return np.argmax(self.q_table[state]) # 利用 def learn(self, state, action, reward, next_state): # Q-Learning更新公式 predict = self.q_table[state, action] target = reward + self.discount_factor * np.max(self.q_table[next_state]) self.q_table[state, action] += self.learning_rate * (target - predict) # 使用示例 agent = QLearning(state_size=16, action_size=4) # 训练循环 for episode in range(1000): state = env.reset() done = False while not done: # 选择动作 action = agent.choose_action(state) # 执行动作,获得反馈 next_state, reward, done = env.step(action) # 学习 agent.learn(state, action, reward, next_state) state = next_state
Q-Learning + 深度学习,解决高维状态空间问题。
核心创新:
代码实现:
import torch import torch.nn as nn import torch.optim as optim import random from collections import deque class DQN(nn.Module): def __init__(self, state_size, action_size): super(DQN, self).__init__() self.fc1 = nn.Linear(state_size, 64) self.fc2 = nn.Linear(64, 64) self.fc3 = nn.Linear(64, action_size) def forward(self, x): x = torch.relu(self.fc1(x)) x = torch.relu(self.fc2(x)) return self.fc3(x) class DQNAgent: def __init__(self, state_size, action_size): self.state_size = state_size self.action_size = action_size self.memory = deque(maxlen=10000) self.gamma = 0.95 # 折扣因子 self.epsilon = 1.0 # 探索率 self.epsilon_min = 0.01 self.epsilon_decay = 0.995 self.learning_rate = 0.001 # 主网络和目标网络 self.model = DQN(state_size, action_size) self.target_model = DQN(state_size, action_size) self.update_target_model() self.optimizer = optim.Adam(self.model.parameters(), lr=self.learning_rate) def update_target_model(self): """将主网络的权重复制到目标网络""" self.target_model.load_state_dict(self.model.state_dict()) def remember(self, state, action, reward, next_state, done): """存储经验""" self.memory.append((state, action, reward, next_state, done)) def act(self, state): """选择动作(ε-贪心)""" if np.random.random() <= self.epsilon: return random.randrange(self.action_size) act_values = self.model(torch.FloatTensor(state)) return np.argmax(act_values.data.numpy()) def replay(self, batch_size): """经验回放训练""" if len(self.memory) < batch_size: return minibatch = random.sample(self.memory, batch_size) for state, action, reward, next_state, done in minibatch: target = self.model(torch.FloatTensor(state)) if done: target[0][action] = reward else: t = self.target_model(torch.FloatTensor(next_state)) target[0][action] = reward + self.gamma * torch.max(t).item() # 训练 self.optimizer.zero_grad() output = self.model(torch.FloatTensor(state)) loss = nn.MSELoss()(output, target) loss.backward() self.optimizer.step() # 衰减探索率 if self.epsilon > self.epsilon_min: self.epsilon *= self.epsilon_decay # 使用示例 env = gym.make('CartPole-v1') state_size = env.observation_space.shape[0] action_size = env.action_space.n agent = DQNAgent(state_size, action_size) # 训练 for episode in range(500): state = env.reset() for time in range(500): action = agent.act(state) next_state, reward, done, _ = env.step(action) agent.remember(state, action, reward, next_state, done) state = next_state if done: agent.update_target_model() break if len(agent.memory) > 32: agent.replay(32)
直接学习策略函数,而非值函数。
REINFORCE算法:
class PolicyGradientAgent: def __init__(self, state_size, action_size): self.state_size = state_size self.action_size = action_size self.gamma = 0.99 self.learning_rate = 0.001 # 策略网络 self.model = nn.Sequential( nn.Linear(state_size, 64), nn.ReLU(), nn.Linear(64, action_size), nn.Softmax(dim=-1) ) self.optimizer = optim.Adam(self.model.parameters(), lr=self.learning_rate) def choose_action(self, state): """根据策略采样动作""" state = torch.FloatTensor(state) probs = self.model(state) action_dist = torch.distributions.Categorical(probs) action = action_dist.sample() return action.item() def compute_returns(self, rewards): """计算折扣回报""" returns = [] R = 0 for r in reversed(rewards): R = r + self.gamma * R returns.insert(0, R) return returns def train(self, episode): """训练一个episode""" states = torch.FloatTensor(np.array([s for s, _, _, _ in episode])) actions = torch.LongTensor(np.array([a for _, a, _, _ in episode])) rewards = [r for _, _, r, _ in episode] # 计算回报 returns = self.compute_returns(rewards) returns = torch.FloatTensor(returns) # 标准化回报 returns = (returns - returns.mean()) / (returns.std() + 1e-8) # 计算策略梯度 probs = self.model(states) action_dist = torch.distributions.Categorical(probs) log_probs = action_dist.log_prob(actions) # 损失函数:-log_prob * return loss = -(log_probs * returns).mean() # 优化 self.optimizer.zero_grad() loss.backward() self.optimizer.step()
结合价值函数和策略函数的优势。
class Actor(nn.Module): def __init__(self, state_size, action_size): super(Actor, self).__init__() self.fc1 = nn.Linear(state_size, 64) self.fc2 = nn.Linear(64, action_size) def forward(self, x): x = torch.relu(self.fc1(x)) return torch.softmax(self.fc2(x), dim=-1) class Critic(nn.Module): def __init__(self, state_size): super(Critic, self).__init__() self.fc1 = nn.Linear(state_size, 64) self.fc2 = nn.Linear(64, 1) def forward(self, x): x = torch.relu(self.fc1(x)) return self.fc2(x) class A2CAgent: def __init__(self, state_size, action_size): self.actor = Actor(state_size, action_size) self.critic = Critic(state_size) self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=0.001) self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=0.001) self.gamma = 0.99 def train(self, state, action, reward, next_state, done): # 计算TD目标 with torch.no_grad(): target = reward + (1 - done) * self.gamma * self.critic(next_state) # 更新Critic value = self.critic(state) critic_loss = nn.MSELoss()(value, target) self.critic_optimizer.zero_grad() critic_loss.backward() self.critic_optimizer.step() # 更新Actor advantage = target - value.detach() probs = self.actor(state) action_dist = torch.distributions.Categorical(probs) actor_loss = -action_dist.log_prob(action) * advantage self.actor_optimizer.zero_grad() actor_loss.backward() self.actor_optimizer.step()
class PPOAgent: def __init__(self, state_size, action_size): self.actor = Actor(state_size, action_size) self.critic = Critic(state_size) self.clip_epsilon = 0.2 def compute_ppo_loss(self, states, actions, old_probs, advantages, returns): # 计算新的动作概率 probs = self.actor(states) action_dist = torch.distributions.Categorical(probs) new_probs = action_dist.log_prob(actions) # 概率比 ratio = torch.exp(new_probs - old_probs) # PPO裁剪目标 surr1 = ratio * advantages surr2 = torch.clamp(ratio, 1 - self.clip_epsilon, 1 + self.clip_epsilon) * advantages actor_loss = -torch.min(surr1, surr2).mean() # Critic损失 value = self.critic(states) critic_loss = nn.MSELoss()(value, returns) return actor_loss + critic_loss
ChatGPT等大模型训练的关键技术。
# RLHF训练流程 def train_with_rlhf(base_model, reward_model, prompts): # 1. 从基础模型采样 responses = [base_model.generate(p) for p in prompts] # 2. 使用reward模型打分 rewards = [reward_model.score(p, r) for p, r in zip(prompts, responses)] # 3. 使用PPO训练策略模型 for state, action, reward in zip(prompts, responses, rewards): advantage = compute_advantage(reward, baseline) ppo_agent.train(state, action, reward, advantage)
# Atari游戏训练 env = gym.make('Breakout-v0') agent = DQNAgent(state_size, action_size) for episode in range(10000): # 预处理游戏画面 state = preprocess_frame(env.reset()) while True: # 选择动作 action = agent.act(state) # 执行动作 next_state, reward, done, info = env.step(action) next_state = preprocess_frame(next_state) # 存储经验 agent.remember(state, action, reward, next_state, done) # 训练 if len(agent.memory) > 1000: agent.replay(32) state = next_state if done: break
# 强化学习推荐 class RLRecommender: def __init__(self, num_items, user_features_size): self.agent = DQNAgent(user_features_size, num_items) def train(self, user_interactions): for episode in user_interactions: user_state = episode['user_features'] item = self.agent.act(user_state) feedback = episode['feedback'][item] # 正反馈=+1,负反馈=-1 reward = 1 if feedback == 'click' else -1 # 更新模型 self.agent.remember(user_state, item, reward, user_state, True) if len(self.agent.memory) > 100: self.agent.replay(32)
# 简化的自动驾驶RL class DrivingEnv: def step(self, action): # action: 加速, 刹车, 左转, 右转 next_state = self.simulate(action) reward = self.compute_reward() done = self.check_collision() return next_state, reward, done def compute_reward(self): reward = 0 # 保持速度 if self.speed > 0: reward += 0.1 # 安全距离 if self.distance_to_car_ahead < 2: reward -= 1 # 遵守交通规则 if self.run_red_light: reward -= 10 return reward
**问题:**环境反馈很少,难以学习
解决方案:
**问题:**需要百万级交互才能学习
解决方案:
**问题:**探索不够 vs 利用不足
解决方案:
强化学习是从数据到决策的关键技术: