DDPG(Deep Deterministic Policy Gradient)는 Deep Q-Network에서 영감을 받은 모델이 없는 비정책 심층 강화 알고리즘입니다. 이 기사에서는 정책 그라데이션을 사용하는 Actor-Critic을 기반으로 합니다.
DDPG의 구현 및 설명은
다음은 하나입니다. 구현할 하나 단계별:
DDPG는 Replay Buffer를 사용하여 환경(Sₜ, aₜ, Rₜ, Sₜ+₁)을 탐색하여 샘플링된 프로세스와 보상을 저장합니다. Replay Buffer는 에이전트가 학습을 가속화하고 DDPG의 안정성을 높이는 데 중요한 역할을 합니다.
class Replay_buffer(): ''' Code based on: https://github.com/openai/baselines/blob/master/baselines/deepq/replay_buffer.py Expects tuples of (state, next_state, action, reward, done) ''' def __init__(self, max_size=capacity): """Create Replay buffer. Parameters ---------- size: int Max number of transitions to store in the buffer. When the buffer overflows the old memories are dropped. """ self.storage = [] self.max_size = max_size self.ptr = 0 def push(self, data): if len(self.storage) == self.max_size: self.storage[int(self.ptr)] = data self.ptr = (self.ptr + 1) % self.max_size else: self.storage.append(data) def sample(self, batch_size): """Sample a batch of experiences. Parameters ---------- batch_size: int How many transitions to sample. Returns ------- state: np.array batch of state or observations action: np.array batch of actions executed given a state reward: np.array rewards received as results of executing action next_state: np.array next state next state or observations seen after executing action done: np.array done[i] = 1 if executing ation[i] resulted in the end of an episode and 0 otherwise. """ ind = np.random.randint(0, len(self.storage), size=batch_size) state, next_state, action, reward, done = [], [], [], [], [] for i in ind: st, n_st, act, rew, dn = self.storage[i] state.append(np.array(st, copy=False)) next_state.append(np.array(n_st, copy=False)) action.append(np.array(act, copy=False)) reward.append(np.array(rew, copy=False)) done.append(np.array(dn, copy=False)) return np.array(state), np.array(next_state), np.array(action), np.array(reward).reshape(-1, 1), np.array(done).reshape(-1, 1)
이것은 Actor-Critic 강화 학습 알고리즘을 PyTorch로 구현한 것입니다. 이 코드는 두 개의 신경망 모델인 Actor와 Critic을 정의합니다.
액터 모델의 입력: 환경 상태, 액터 모델의 출력: 연속 값을 갖는 작업.
비평가 모델의 입력: 환경 상태 및 행동, 비판 모델의 출력: Q 값은 현재 상태-행동 쌍의 예상 총 보상입니다.
class Actor(nn.Module): """ The Actor model takes in a state observation as input and outputs an action, which is a continuous value. It consists of four fully connected linear layers with ReLU activation functions and a final output layer selects one single optimized action for the state """ def __init__(self, n_states, action_dim, hidden1): super(Actor, self).__init__() self.net = nn.Sequential( nn.Linear(n_states, hidden1), nn.ReLU(), nn.Linear(hidden1, hidden1), nn.ReLU(), nn.Linear(hidden1, hidden1), nn.ReLU(), nn.Linear(hidden1, 1) ) def forward(self, state): return self.net(state) class Critic(nn.Module): """ The Critic model takes in both a state observation and an action as input and outputs a Q-value, which estimates the expected total reward for the current state-action pair. It consists of four linear layers with ReLU activation functions, State and action inputs are concatenated before being fed into the first linear layer. The output layer has a single output, representing the Q-value """ def __init__(self, n_states, action_dim, hidden2): super(Critic, self).__init__() self.net = nn.Sequential( nn.Linear(n_states + action_dim, hidden2), nn.ReLU(), nn.Linear(hidden2, hidden2), nn.ReLU(), nn.Linear(hidden2, hidden2), nn.ReLU(), nn.Linear(hidden2, action_dim) ) def forward(self, state, action): return self.net(torch.cat((state, action), 1))
배우가 선택한 동작에 노이즈를 추가하는 것은 탐색을 장려하고 학습 과정을 개선하기 위해 DDPG에서 사용하는 기술입니다.
Gaussian 잡음이나 Ornstein-Uhlenbeck 잡음을 사용할 수 있습니다. 가우스 잡음은 간단하고 구현하기 쉬우며 Ornstein-Uhlenbeck 잡음은 에이전트가 작업 공간을 보다 효율적으로 탐색하는 데 도움이 될 수 있는 시간 상관 잡음을 생성합니다. 그러나 Ornstein-Uhlenbeck 잡음 변동은 가우스 잡음 방법보다 더 부드럽고 덜 무작위적입니다.
import numpy as np import random import copy class OU_Noise(object): """Ornstein-Uhlenbeck process. code from : https://math.stackexchange.com/questions/1287634/implementing-ornstein-uhlenbeck-in-matlab The OU_Noise class has four attributes size: the size of the noise vector to be generated mu: the mean of the noise, set to 0 by default theta: the rate of mean reversion, controlling how quickly the noise returns to the mean sigma: the volatility of the noise, controlling the magnitude of fluctuations """ def __init__(self, size, seed, mu=0., theta=0.15, sigma=0.2): self.mu = mu * np.ones(size) self.theta = theta self.sigma = sigma self.seed = random.seed(seed) self.reset() def reset(self): """Reset the internal state (= noise) to mean (mu).""" self.state = copy.copy(self.mu) def sample(self): """Update internal state and return it as a noise sample. This method uses the current state of the noise and generates the next sample """ dx = self.theta * (self.mu - self.state) + self.sigma * np.array([np.random.normal() for _ in range(len(self.state))]) self.state += dx return self.state
DDPG에서 가우스 노이즈를 사용하려면 에이전트의 작업 선택 프로세스에 가우스 노이즈를 직접 추가하면 됩니다.
DDPG(Deep Deterministic Policy Gradient)는 기능 근사를 위해 두 세트의 행위자-비평가 신경망을 사용합니다. DDPG에서 대상 네트워크는 Actor-Critic이며, 이는 Actor-Critic 네트워크와 동일한 구조 및 매개변수화를 갖습니다.
훈련 기간 동안 에이전트는 Actor-Critic 네트워크를 사용하여 환경과 상호 작용하고 경험 튜플(Sₜ, Aₜ, Rₜ, Sₜ+₁)을 재생 버퍼에 저장합니다. 그런 다음 에이전트는 재생 버퍼에서 샘플링하고 Actor-Critic 네트워크를 데이터로 업데이트합니다. Actor-Critic 네트워크에서 직접 복사하여 대상 네트워크 가중치를 업데이트하는 대신 DDPG 알고리즘은 소프트 대상 업데이트라는 프로세스를 통해 대상 네트워크 가중치를 천천히 업데이트합니다.
소프트 타겟의 업데이트는 Actor-Critic 네트워크에서 타겟 업데이트 속도(τ)라고 하는 타겟 네트워크로 전송되는 가중치의 일부입니다.
소프트 타겟의 업데이트 공식은 다음과 같습니다.
소프트 타겟 기술을 사용하면 학습 안정성을 크게 향상시킬 수 있습니다.
#Set Hyperparameters # Hyperparameters adapted for performance from capacity=1000000 batch_size=64 update_iteration=200 tau=0.001 # tau for soft updating gamma=0.99 # discount factor directory = './' hidden1=20 # hidden layer for actor hidden2=64. #hiiden laye for critic class DDPG(object): def __init__(self, state_dim, action_dim): """ Initializes the DDPG agent. Takes three arguments: state_dim which is the dimensionality of the state space, action_dim which is the dimensionality of the action space, and max_action which is the maximum value an action can take. Creates a replay buffer, an actor-critic networks and their corresponding target networks. It also initializes the optimizer for both actor and critic networks alog with counters to track the number of training iterations. """ self.replay_buffer = Replay_buffer() self.actor = Actor(state_dim, action_dim, hidden1).to(device) self.actor_target = Actor(state_dim, action_dim,hidden1).to(device) self.actor_target.load_state_dict(self.actor.state_dict()) self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=3e-3) self.critic = Critic(state_dim, action_dim,hidden2).to(device) self.critic_target = Critic(state_dim, action_dim,hidden2).to(device) self.critic_target.load_state_dict(self.critic.state_dict()) self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=2e-2) # learning rate self.num_critic_update_iteration = 0 self.num_actor_update_iteration = 0 self.num_training = 0 def select_action(self, state): """ takes the current state as input and returns an action to take in that state. It uses the actor network to map the state to an action. """ state = torch.FloatTensor(state.reshape(1, -1)).to(device) return self.actor(state).cpu().data.numpy().flatten() def update(self): """ updates the actor and critic networks using a batch of samples from the replay buffer. For each sample in the batch, it computes the target Q value using the target critic network and the target actor network. It then computes the current Q value using the critic network and the action taken by the actor network. It computes the critic loss as the mean squared error between the target Q value and the current Q value, and updates the critic network using gradient descent. It then computes the actor loss as the negative mean Q value using the critic network and the actor network, and updates the actor network using gradient ascent. Finally, it updates the target networks using soft updates, where a small fraction of the actor and critic network weights are transferred to their target counterparts. This process is repeated for a fixed number of iterations. """ for it in range(update_iteration): # For each Sample in replay buffer batch state, next_state, action, reward, done = self.replay_buffer.sample(batch_size) state = torch.FloatTensor(state).to(device) action = torch.FloatTensor(action).to(device) next_state = torch.FloatTensor(next_state).to(device) done = torch.FloatTensor(1-done).to(device) reward = torch.FloatTensor(reward).to(device) # Compute the target Q value target_Q = self.critic_target(next_state, self.actor_target(next_state)) target_Q = reward + (done * gamma * target_Q).detach() # Get current Q estimate current_Q = self.critic(state, action) # Compute critic loss critic_loss = F.mse_loss(current_Q, target_Q) # Optimize the critic self.critic_optimizer.zero_grad() critic_loss.backward() self.critic_optimizer.step() # Compute actor loss as the negative mean Q value using the critic network and the actor network actor_loss = -self.critic(state, self.actor(state)).mean() # Optimize the actor self.actor_optimizer.zero_grad() actor_loss.backward() self.actor_optimizer.step() """ Update the frozen target models using soft updates, where tau,a small fraction of the actor and critic network weights are transferred to their target counterparts. """ for param, target_param in zip(self.critic.parameters(), self.critic_target.parameters()): target_param.data.copy_(tau * param.data + (1 - tau) * target_param.data) for param, target_param in zip(self.actor.parameters(), self.actor_target.parameters()): target_param.data.copy_(tau * param.data + (1 - tau) * target_param.data) self.num_actor_update_iteration += 1 self.num_critic_update_iteration += 1 def save(self): """ Saves the state dictionaries of the actor and critic networks to files """ torch.save(self.actor.state_dict(), directory + 'actor.pth') torch.save(self.critic.state_dict(), directory + 'critic.pth') def load(self): """ Loads the state dictionaries of the actor and critic networks to files """ self.actor.load_state_dict(torch.load(directory + 'actor.pth')) self.critic.load_state_dict(torch.load(directory + 'critic.pth'))
여기에서는 OpenAI Gym의 "MountainCarContinuous-v0"을 사용하여 DDPG RL 모델을 훈련합니다. 이곳의 환경은 지속적인 행동과 관찰 공간을 제공하며 목표는 차량을 최대한 빨리 산 정상에 올리는 것입니다. 가능한 한.
최대 훈련 횟수, 탐색 노이즈, 기록 간격 등과 같은 알고리즘의 다양한 매개 변수는 아래에 정의되어 있습니다. 고정된 무작위 시드를 사용하면 프로세스를 역추적할 수 있습니다.
import gym # create the environment env_name='MountainCarContinuous-v0' env = gym.make(env_name) device = 'cuda' if torch.cuda.is_available() else 'cpu' # Define different parameters for training the agent max_episode=100 max_time_steps=5000 ep_r = 0 total_step = 0 score_hist=[] # for rensering the environmnet render=True render_interval=10 # for reproducibility env.seed(0) torch.manual_seed(0) np.random.seed(0) #Environment action ans states state_dim = env.observation_space.shape[0] action_dim = env.action_space.shape[0] max_action = float(env.action_space.high[0]) min_Val = torch.tensor(1e-7).float().to(device) # Exploration Noise exploration_noise=0.1 exploration_noise=0.1 * max_action
DDPG 에이전트 클래스의 인스턴스를 생성하여 지정된 횟수만큼 에이전트를 교육합니다. 에이전트의 update() 메서드는 각 라운드가 끝날 때 호출되어 매개변수를 업데이트하고, save() 메서드는 매 10라운드마다 에이전트의 매개변수를 파일에 저장하는 데 사용됩니다.
# Create a DDPG instance
agent = DDPG(state_dim, action_dim)
# Train the agent for max_episodes
for i in range(max_episode):
total_reward = 0
step =0
state = env.reset()
fort in range(max_time_steps):
action = agent.select_action(state)
# Add Gaussian noise to actions for exploration
action = (action + np.random.normal(0, 1, size=action_dim)).clip(-max_action, max_action)
#action += ou_noise.sample()
next_state, reward, done, info = env.step(action)
total_reward += reward
if render and i >= render_interval : env.render()
agent.replay_buffer.push((state, next_state, action, reward, np.float(done)))
state = next_state
if done:
break
step += 1
score_hist.append(total_reward)
total_step += step+1
print("Episode: t{} Total Reward: t{:0.2f}".format( i, total_reward))
agent.update()
if i % 10 == 0:
agent.save()
env.close()test_iteration=100
for i in range(test_iteration):
state = env.reset()
for t in count():
action = agent.select_action(state)
next_state, reward, done, info = env.step(np.float32(action))
ep_r += reward
print(reward)
env.render()
if done:
print("reward{}".format(reward))
print("Episode t{}, the episode reward is t{:0.2f}".format(i, ep_r))
ep_r = 0
env.render()
break
state = next_state다음 매개변수를 사용하여 모델을 수렴시킵니다.
75회 학습 후 효과는 다음과 같습니다.
DDPG 알고리즘은 메서드입니다. 네트워크(DQN) 알고리즘에서 영감을 받은 Deep Q -Model-free off-policy Actor-Critic 알고리즘의 영향을 받았습니다. 정책 경사 방법과 Q-학습의 장점을 결합하여 연속 행동 공간에서 결정론적 정책을 학습합니다.
DQN과 유사하게 재생 버퍼를 사용하여 네트워크 훈련을 위한 과거 경험과 대상 네트워크를 저장함으로써 훈련 프로세스의 안정성을 향상시킵니다.
DDPG 알고리즘은 최적의 성능을 위해 신중한 하이퍼파라미터 조정이 필요합니다. 하이퍼파라미터에는 학습률, 배치 크기, 대상 네트워크 업데이트 속도, 감지 노이즈 매개변수가 포함됩니다. 하이퍼파라미터의 작은 변화는 알고리즘 성능에 큰 영향을 미칠 수 있습니다.
위 내용은 PyTorch 코드 구현 및 DDPG 강화 학습의 단계별 설명의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
