当前位置：首页 > news >正文

简单的强化学习举例

news 2025/5/12 14:49:51

1，定义奖励函数

首先，需要根据具体的任务需求来定义奖励函数。例如，对于机器人导航任务，可以根据机器人与目标点的距离来定义奖励函数：

import numpy as npdef navigation_reward(robot_position, target_position):# 计算机器人当前位置与目标位置之间的欧几里得距离distance = np.linalg.norm(np.array(robot_position) - np.array(target_position))# 距离越近，奖励越高reward = -distancereturn reward

2. 集成奖励函数到环境中

在强化学习中，环境负责与智能体进行交互，并根据智能体的动作给出相应的奖励。通常，环境类会有一个 step 方法，该方法接受智能体的动作作为输入，并返回下一个状态、奖励、是否终止等信息

import numpy as npclass NavigationEnv: # 定义导航环境类def __init__(self):# 初始化机器人的起始位置和目标位置self.robot_position = np.array([0, 0]) # 机器人初始位置 (x=0, y=0)self.target_position = np.array([10, 10]) # 目标位置 (x=10, y=10)def step(self, action):# 核心逻辑：执行动作 → 更新状态 → 计算奖励 → 判断终止# 根据动作更新机器人的位置self.robot_position += action # 通过向量加法更新位置# 通过向量加法更新机器人位置。例如，若动作是 [2, 3]，新位置为 (0+2, 0+3) = (2, 3)# 计算奖励reward = navigation_reward(self.robot_position, self.target_position)# 判断是否到达目标位置done = np.linalg.norm(self.robot_position - self.target_position) < 0.1# 返回下一个状态、奖励、是否终止等信息，返回四元组return self.robot_position, reward, done, {}

3. 在训练循环中使用奖励函数

在训练智能体时，需要在每个时间步调用环境的 step 方法，并根据返回的奖励来更新智能体的策略。以下是一个简单的训练循环示例：

代码分为三大部分：

环境 (NavigationEnv)
- 状态：机器人的二维坐标
- 动作：离散动作空间 (0 = 向右移动，1 = 向上移动)
- 奖励：负距离 (-distance)，鼓励机器人靠近目标
策略网络 (PolicyNetwork)
- 输入：机器人当前位置 (二维向量)
- 输出：两个动作的概率分布
- 结构：两层全连接网络，使用 ReLU 激活和 softmax 输出
训练循环
- 收集数据：每个 episode 收集状态、动作、奖励序列
- 计算回报：使用折扣因子计算每个时间步的累积回报
- 更新策略：通过最大化累积回报的对数概率来更新网络参数

# 修正后的完整代码
import numpy as np
import torch
import torch.optim as optimclass NavigationEnv:def __init__(self):self.reset() ## 初始化时直接调用reset方法def reset(self):self.robot_position = np.array([0.0, 0.0])self.target_position = np.array([10.0, 10.0])return self.robot_positiondef step(self, action):self.robot_position += action # # 执行动作（向量加法）distance = np.linalg.norm(self.robot_position - self.target_position)reward = -distance  # 简单奖励函数，奖励设计：距离越近奖励越高（负值越小）done = distance < 0.1 # 终止条件：距离目标小于0.1单位return self.robot_position, reward, done, {}class PolicyNetwork(torch.nn.Module):def __init__(self, input_size, output_size):super().__init__()self.fc1 = torch.nn.Linear(input_size, 128) # 输入层→隐藏层（128神经元）self.fc2 = torch.nn.Linear(128, output_size) # 隐藏层→输出层（动作维度）def forward(self, x):x = torch.relu(self.fc1(x)) # ReLU激活函数引入非线性 return torch.softmax(self.fc2(x), dim=-1) # 输出动作概率分布env = NavigationEnv()
policy_net = PolicyNetwork(2, 2) # 输入状态维度2，输出动作数2
optimizer = optim.Adam(policy_net.parameters(), lr=0.001) # 自适应学习率优化器
action_map = {0: np.array([1,0]), 1: np.array([0,1])} # 动作映射表 0向右移动1个单位，1向上移动1个单位
gamma = 0.99 #未来奖励折扣因子
for episode in range(1000): #训练1000轮state = env.reset()  # 重置环境done = Falserewards, states, actions = [], [], [] # 数据收集容器while not done:state_tensor = torch.FloatTensor(state) # 转为PyTorch张量probs = policy_net(state_tensor)        # 获取动作概率action_idx = torch.multinomial(probs, 1).item() # 按概率抽样动作action = action_map[action_idx]         # 转换为实际动作向量next_state, reward, done, _ = env.step(action)  # 环境反馈# 存储轨迹数据rewards.append(reward)                  states.append(state_tensor)actions.append(action_idx)state = next_state  # 状态更新# 计算累积回报returns = []cumulative = 0for r in reversed(rewards):             # 从最后一步反向计算cumulative = r + gamma * cumulative # 计算累积回报returns.insert(0, cumulative)       # 保持时序# 策略梯度更新optimizer.zero_grad()                   # 清空梯度for s, a, R in zip(states, actions, returns):prob = policy_net(s)                # 重新计算动作概率（需梯度）log_prob = torch.log(prob[a])       # 选择动作的对数概率loss = -log_prob * R                # 策略梯度损失loss.backward()                     # 反向传播累积梯度optimizer.step()print(f"Episode {episode}, Total Reward: {sum(rewards):.1f}")