Deep Reinforcement Learning bằng Python
Timothée Carayol
Principal Machine Learning Engineer, Komment

for step = 1 to T do: # Thực hiện hành động tối ưu theo hàm giá trị # Quan sát trạng thái kế tiếp và phần thưởng # Thêm chuyển tiếp vào bộ đệm replay# Gán mức ưu tiên cao nhất (1)# Lấy mẫu một lô chuyển tiếp trước đây# Dựa trên mức ưu tiên (2)# Tính sai số TD cho lô# Tính loss và cập nhật Q Network# Dùng trọng số lấy mẫu theo tầm quan trọng (4)# Cập nhật mức ưu tiên của các chuyển tiếp đã lấy mẫu (3)# Tăng dần lấy mẫu theo tầm quan trọng theo thời gian. (5)
(1) Chuyển tiếp mới được thêm với mức ưu tiên cao nhất $p_i = \max_k(p_k)$
(2) Lấy mẫu chuyển tiếp $i$ với xác suất $$P(i) = p_i^{\alpha} / \sum_k p_k^{\alpha}\ \ \ \ \ \ \ \ (0<\alpha<1)$$
(3) Chuyển tiếp đã lấy mẫu được cập nhật mức ưu tiên bằng sai số TD của chúng: $p_i = |\delta_i| + \varepsilon$
(4) Dùng trọng số lấy mẫu theo tầm quan trọng $$w_i = \left( \frac{1}{N} \cdot \frac{1}{P(i)} \right)^\beta\ \ \ \ \ \ \ \ (0<\beta<1)$$
(5) Tăng dần $\beta$ tiến tới 1
def __init__(self, capacity, alpha=0.6, beta=0.4, beta_increment=0.001, epsilon=0.001): # Khởi tạo bộ đệm bộ nhớ self.memory = deque(maxlen=capacity)# Lưu tham số và khởi tạo mức ưu tiên self.alpha, self.beta, self.beta_increment, self.epsilon = (alpha, beta, beta_increment, epsilon) self.priorities = deque(maxlen=capacity)...
... def push(self, state, action, reward, next_state, done): # Thêm trải nghiệm vào bộ đệm bộ nhớ experience_tuple = (state, action, reward, next_state, done) self.memory.append(experience_tuple)# Đặt ưu tiên của chuyển tiếp mới thành mức tối đa max_priority = max(self.priorities) if self.memory else 1.0 self.priorities.append(max_priority)...
def sample(self, batch_size): priorities = np.array(self.priorities) # Tính xác suất lấy mẫu probabilities = priorities**self.alpha / np.sum(priorities**self.alpha)# Chọn ngẫu nhiên các chỉ số mẫu indices = np.random.choice(len(self.memory), batch_size, p=probabilities)# Tính trọng số weights = (1 / (len(self.memory) * probabilities)) ** self.beta weights /= np.max(weights) states, actions, rewards, next_states, dones = zip(*[self.memory[idx] for idx in indices]) weights = [weights[idx] for idx in indices] states, actions, rewards, next_states, dones = (zip(*[self.memory[idx] for idx in indices]))# Trả về tensor states = torch.tensor(states, dtype=torch.float32) ... # Lặp lại cho rewards, next_states, dones, weights actions = torch.tensor(actions, dtype=torch.long).unsqueeze(1) return (states, actions, rewards, next_states, dones, indices, weights)
... def update_priorities(self, indices, td_errors: torch.Tensor): # Cập nhật mức ưu tiên cho các chuyển tiếp đã lấy mẫu for idx, td_error in zip(indices, td_errors.abs()): self.priorities[idx] = abs(td_error.item()) + self.epsilondef increase_beta(self): # Tăng dần beta về 1 self.beta = min(1.0, self.beta + self.beta_increment)
Trong mã trước vòng lặp:
buffer = PrioritizedReplayBuffer(capacity)
Đầu mỗi tập:
buffer.increase_beta()
# Sau khi chọn hành động buffer.push(state, action, reward, next_state, done) ...# Trước khi tính sai số TD: replay_buffer.sample(batch_size) ...# Sau khi tính sai số TD buffer.update_priorities(indices, td_errors)loss = torch.sum(weights * (td_errors ** 2))
100 lượt huấn luyện trong môi trường Cartpole:

Sau 100 epoch:

Sau 400 epoch:


Deep Reinforcement Learning bằng Python