深度强化学习DQN训练避障

目录

一.前言

二.代码

2.1完整代码

2.2运行环境

2.3动作空间

2.4奖励函数

2.5状态输入

2.6实验结果

---

一.前言

深度Q网络（DQN）是深度强化学习领域的一项革命性技术，它成功地将深度学习的强大感知能力与强化学习的决策能力相结合。在过去的几年里，DQN已经在许多复杂的问题上展示了其卓越的性能，从经典的Atari游戏到更复杂的机器人控制任务。特别值得一提的是，DQN在处理高维状态空间和动作空间的问题时表现出了显著的优势，使得它能够处理传统强化学习方法难以解决的难题。

二维环境避障问题是强化学习领域的一个经典问题，它要求智能体在二维空间中移动，同时避免与障碍物碰撞。这个问题看似简单，但实际上涉及到了许多复杂的因素，如环境的不确定性、部分可观察性、以及智能体的感知和决策能力限制等。因此，开发一种能够有效解决二维环境避障问题的方法具有重要的实际意义和应用价值。

近年来，随着深度学习和强化学习技术的不断发展，基于DQN的二维环境避障方法已经取得了显著的进展。通过利用深度学习的强大特征提取能力，DQN能够从原始的环境状态中学习到有用的特征表示，并基于这些特征进行高效的决策。此外，DQN还引入了经验回放和目标网络等关键技术，进一步提高了算法的稳定性和收敛速度。

本文研究如何让智能体在一个二维障碍物环境中进行避障。

二.代码

2.1完整代码

import gym  
import torch,copy
import torch.nn as nn
import torch.nn.functional as F
import numpy as np  
from gym import spaces  
from gym.utils import seeding  
import matplotlib.pyplot as plt  
#超参数
BATCH_SIZE = 32
LR = 0.01
EPSILON = 0.9 #随机选取的概率，如果概率小于这个随机数，就采取greedy的行为
GAMMA = 0.9
TARGET_REPLACE_ITER = 100
MEMORY_CAPACITY = 2000
N_ACTIONS=len(np.arange(0, 2*np.pi, 0.1))
N_STATES=4
ENV_A_SHAPE =0
class Net(nn.Module):
    def __init__(self, ):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(N_STATES, 500)
        self.fc1.weight.data.normal_(0, 0.1)   # initialization
        self.out = nn.Linear(500, N_ACTIONS)
        self.out.weight.data.normal_(0, 0.1)   # initialization
    def forward(self, x):
        x = self.fc1(x)
        x = F.relu(x)
        actions_value = self.out(x)
        return actions_value
class DQN(object):
    def __init__(self):
        #DQN是Q-Leaarning的一种方法，但是有两个神经网络，一个是eval_net一个是target_net
        #两个神经网络相同，参数不同，是不是把eval_net的参数转化成target_net的参数，产生延迟的效果
        self.eval_net=Net()
        self.target_net =copy.deepcopy(self.eval_net)
        self.learn_step_counter = 0 #学习步数计数器
        self.memory_counter = 0 #记忆库中位值的计数器
        self.memory = np.zeros((MEMORY_CAPACITY,N_STATES * 2 + 2)) #初始化记忆库
        #记忆库初始化为全0，存储两个state的数值加上一个a(action)和一个r(reward)的数值
        self.optimizer = torch.optim.Adam(self.eval_net.parameters(),lr = LR)
        self.loss_func = nn.MSELoss() #优化器和损失函数    
    #接收环境中的观测值，并采取动作
    def choose_action(self,x):
        #x为观测值
        x = torch.unsqueeze(torch.FloatTensor(x),0)
        if np.random.uniform() < EPSILON:
            #随机值得到的数有百分之九十的可能性<0.9,所以该if成立的几率是90%
            #90%的情况下采取actions_value高的作为最终动作
            actions_value = self.eval_net.forward(x)
            action = torch.max(actions_value,1)[1].data.numpy()
            action = action[0] if ENV_A_SHAPE == 0 else action.reshape(ENV_A_SHAPE) # return the argmax index
        else:
            #其他10%采取随机选取动作
            action = np.random.randint(0,N_ACTIONS) #从动作中选一个动作
            action = action if ENV_A_SHAPE == 0 else action.reshape(ENV_A_SHAPE)
        return action    
    #记忆库，存储之前的记忆，学习之前的记忆库里的东西
    def store_transition(self,s,a,r,s_):

        transition = np.hstack((s, [a, r], s_))
        # 如果记忆库满了, 就覆盖老数据
        index = self.memory_counter % MEMORY_CAPACITY
        self.memory[index, :] = transition
        self.memory_counter += 1
    def learn(self):
         # target net 参数更新,每隔TARGET_REPLACE_ITE更新一下
        if self.learn_step_counter % TARGET_REPLACE_ITER == 0:
            self.target_net.load_state_dict(self.eval_net.state_dict())
        self.learn_step_counter += 1
        #targetnet是时不时更新一下，evalnet是每一步都更新
 
        # 抽取记忆库中的批数据
        sample_index = np.random.choice(MEMORY_CAPACITY, BATCH_SIZE)
        b_memory = self.memory[sample_index, :] 
        #打包记忆，分开保存进b_s，b_a，b_r，b_s
        b_s = torch.FloatTensor(b_memory[:, :N_STATES])
        b_a = torch.LongTensor(b_memory[:, N_STATES:N_STATES+1].astype(int))
        b_r = torch.FloatTensor(b_memory[:, N_STATES+1:N_STATES+2])
        b_s_ = torch.FloatTensor(b_memory[:, -N_STATES:])
 
        # 针对做过的动作b_a, 来选 q_eval 的值, (q_eval 原本有所有动作的值)
        q_eval = self.eval_net(b_s).gather(1, b_a)  # shape (batch, 1)
        q_next = self.target_net(b_s_).detach()     # q_next 不进行反向传递误差, 所以 detach
        q_target = b_r + GAMMA * q_next.max(1)[0]   # shape (batch, 1)
        loss = self.loss_func(q_eval, q_target)
 
        # 计算, 更新 eval net
        self.optimizer.zero_grad()
        loss.backward() #误差反向传播
        self.optimizer.step()

class RobotEnv(gym.Env):  
    metadata = {'render.modes': ['human']}  
  
    def __init__(self):  
        super(RobotEnv, self).__init__()  
        
        self.grid_size = 1000  
        self.viewobs = np.zeros((self.grid_size, self.grid_size))  
        self.robot_radius = 10  
        self.obstacle_radius = 50  
        self.robot_pos = np.array([self.grid_size // 8, self.grid_size // 2])  
        self.obstacle_pos = np.array([self.grid_size *3// 4, self.grid_size // 2])  
        self.target_pos = np.array([self.grid_size *7 // 8, self.grid_size // 2])  
        self.action_space =np.arange(0, 2*np.pi, 0.1)  
        self.observation_space = spaces.Box(low=0, high=1, shape=(self.grid_size, self.grid_size))  
        self._seed()  
    def _reset(self):
        self.robot_pos = np.array([self.grid_size  // 8, self.grid_size // 2])  
        self.obstacle_pos = np.array([self.grid_size *3// 4, self.grid_size // 2])  
        self.observation_space = spaces.Box(low=0, high=1, shape=(self.grid_size, self.grid_size))  
        return self._get_obs()
    def _seed(self, seed=None):  
        self.np_random, seed = seeding.np_random(seed)  
        return [seed]  
  
    def _step(self, action):  
        angle = action*0.1  

        dx = self.robot_radius * np.cos(angle)  
        dy = self.robot_radius * np.sin(angle)  
        new_pos = self.robot_pos + np.array([dx, dy])  

        if np.linalg.norm(new_pos - self.obstacle_pos) <= self.obstacle_radius:  
            reward = -1
            done = True  
        elif self.robot_pos[0]<=self.robot_radius+self.robot_radius or self.robot_pos[0]>=self.grid_size-self.robot_radius-self.robot_radius:
            reward=-1
            done = True  
        elif self.robot_pos[1]<=self.robot_radius+self.robot_radius or self.robot_pos[1]>=self.grid_size-self.robot_radius-self.robot_radius:
            reward=-1
            done = True  
        else:  
            reward = 1 / (1 + np.linalg.norm(new_pos - self.target_pos))  
            done = False  
        self.robot_pos = new_pos  
        return self._get_obs(), reward, done, {}  
  
    def _get_obs(self):  
        self.viewobs = np.zeros((self.grid_size, self.grid_size))  


        self.viewobs[int(self.robot_pos[0])][int(self.robot_pos[1])] = 1.0
        self.viewobs[int(self.obstacle_pos[0] - self.obstacle_radius):int(self.obstacle_pos[0] + self.obstacle_radius),   
           int(self.obstacle_pos[1] - self.obstacle_radius):int(self.obstacle_pos[1] + self.obstacle_radius)] = 0.5  
        obs=[self.robot_pos[0],self.robot_pos[1],self.obstacle_pos[0],self.obstacle_pos[1]]
        return obs

  
    def _render(self, mode='human', close=False):  
        if close:  
            return  
        plt.imshow(self.viewobs, cmap='gray')  
        plt.scatter(self.robot_pos[1], self.robot_pos[0], c='red')  
        plt.scatter(self.obstacle_pos[1], self.obstacle_pos[0], c='blue')  
        plt.xlim([0, self.grid_size])  
        plt.ylim([0, self.grid_size])  
        #plt.show()  
        plt.pause(0.01)  
        plt.clf()  
  
if __name__ == "__main__":  
    env = RobotEnv()  
    dqn = DQN()
    for i in range(400000):
        s=env.reset()
        for j in range(300):  # Run for 100 steps as an example  
            action  = dqn.choose_action(s) 
            s_, reward, done, info = env.step(action)  
            print(reward)
            env.render()  
            dqn.store_transition(s,action,reward,s_)
            if dqn.memory_counter > MEMORY_CAPACITY:
                dqn.learn()
            if done:  
                print("Episode finished after {} timesteps".format(i+1))  
                break
            s = s_ # 现在的状态赋值到下一个状态上去

2.2运行环境

gym== 0.7.0

torch==2.1.1

2.3动作空间

??? 在深度强化学习领域中，动作空间的设计对于智能体的决策能力和学习效果具有至关重要的影响。针对连续动作空间的问题，一种常见的处理方式是将其离散化，以便应用离散动作空间的强化学习算法。本实验提出了一种将0-2π的连续动作空间等分为63个离散动作的方法。

??? 通过采用等分为63个离散动作的策略。使得实验在保留了足够的动作分辨率的同时，避免了过度离散化可能带来的维度灾难问题。通过将动作空间划分为足够细密的离散区间，智能体能够更精确地控制自身的行为，从而在面对复杂环境时实现更好的性能。

2.4奖励函数

在本研究中，我们设计的奖励函数主要基于两个核心原则：

1. 靠近障碍物的奖励：当智能体靠近障碍物后方时，将获得一个与距离大小成反比的奖励值。具体而言，随着智能体逐渐接近障碍物，奖励值将逐渐增大，以鼓励智能体保持与障碍物的安全距离。这种设计策略旨在通过提供即时的正反馈来促使智能体学习避免与障碍物发生碰撞的行为。
2. 违反仿真边界或进入障碍物区域的惩罚：一旦智能体离开仿真边界或进入障碍物区域，将获得一个负奖励（惩罚），并且仿真将立即停止。这种严厉的惩罚机制确保了智能体能够清晰地认识到这些行为的不可取性，从而在学习过程中避免重复犯下相同的错误。

2.5状态输入

??? 在本研究中，我们将智能体的当前位置以及障碍物的位置信息作为状态输入。具体来说，状态空间包括智能体的二维坐标和障碍物的二维坐标。这样的设计使得智能体能够直接感知到周围环境的信息，并根据这些信息做出相应的决策。

2.6实验结果

??? 无论是在静态还是动态环境中，智能体都能够准确地感知到障碍物的存在，并有效地规划出避开障碍物的路径。