深度强化学习基础架构与核心算法实现

本文介绍了深度强化学习的工程化部署与核心算法实现。涵盖环境接口标准化（Gymnasium 向量化）、经验回放机制优化（优先经验回放 PER）、值函数方法演进（DQN、Rainbow 集成）以及策略梯度基础架构（REINFORCE、A2C/A3C）。通过 Python 代码示例展示了网络架构设计、目标网络软更新、分布贝尔曼更新及并行训练策略，为大规模分布式训练框架提供技术参考。

栈溢出发布于 2026/3/270 浏览

第一章基础架构与核心算法实现

现代深度强化学习系统的工程化部署高度依赖于底层架构的模块化设计与算法原语的高效实现。本章从环境接口标准化、经验回放机制优化、值函数方法演进以及策略梯度基础架构四个维度，构建具备千级 FPS 采样能力的工业级训练框架。所呈现的实现方案融合了 Gymnasium 向量环境接口、优先经验回放数据结构、分布式值函数近似以及异步并行训练等关键技术，确保算法原型研究与大规模部署之间的无缝迁移。

1.1 环境接口与数据流水线

环境接口的标准化封装与数据流水线的向量化处理构成了强化学习训练系统的 I/O 瓶颈突破点。高效的环境交互架构需同时满足单进程高吞吐与多进程并行化两种场景，并通过观测预处理流水线确保神经网络的输入稳定性。

1.1.1 Gymnasium API 深度封装与向量化

Gymnasium 作为强化学习环境的工业标准接口，其向量化抽象允许在单一 Python 进程中并行驱动多个环境实例，显著提升样本采集吞吐。同步向量化环境与异步向量化环境代表了两种截然不同的并行策略：前者在单进程中通过顺序执行实现批量观测堆叠，适用于计算轻量型环境；后者利用多进程架构实现真正的并行物理执行，在 CPU 密集型仿真场景下展现线性加速特性。观测预处理流水线通过自定义 Wrapper 链式组合实现，其中运行均值方差统计（RunningMeanStd）模块在线维护观测流的统计特性，为深度神经网络提供零均值、单位方差的稳定输入分布；帧堆叠（FrameStack）模块则通过时间维度上的观测拼接，赋予策略网络对运动轨迹的感知能力。

脚本 1.1.1：向量化环境封装与观测预处理流水线

#!/usr/bin/env python3
"""
脚本内容：Gymnasium 向量化环境封装与观测预处理
使用方式：python section_1_1_1_vectorized_env.py --env_id CartPole-v1 --num_envs 8 --async_mode
功能说明：
1. 实现 SyncVectorEnv 与 AsyncVectorEnv 的 OMP 线程感知型封装
2. 提供 RunningMeanStd 在线归一化 Wrapper，维护滑动均值与方差
3. 实现 FrameStack 时序帧堆叠，支持 LazyFrames 内存优化
4. 包含环境速度基准测试函数，验证千级 FPS 采样能力
"""
import argparse
import time
import warnings
from collections import deque
from typing import Any, Callable, Dict, List, Optional, Tuple, Union
import gymnasium as gym
import numpy as np
import torch
from gymnasium import spaces
from gymnasium.core import Env, Wrapper
from gymnasium.vector import AsyncVectorEnv, SyncVectorEnv

class RunningMeanStd:
    
     ():
        ._epsilon = epsilon
        ._shape = shape
        ._mean = np.zeros(shape, dtype=np.float64)
        ._var = np.ones(shape, dtype=np.float64)
        ._count = epsilon 

     () -> :
        
        batch_mean = np.mean(x, axis=)
        batch_var = np.var(x, axis=)
        batch_count = x.shape[]
        new_count = ._count + batch_count
        delta = batch_mean - ._mean
        new_mean = ._mean + delta * batch_count / new_count
        m_a = ._var * ._count
        m_b = batch_var * batch_count
        m2 = m_a + m_b + np.square(delta) * ._count * batch_count / new_count
        new_var = m2 / new_count
        ._mean = new_mean
        ._var = new_var
        ._count = new_count


     () -> np.ndarray:
         ._mean.astype(np.float32)


     () -> np.ndarray:
         np.sqrt(._var + ._epsilon).astype(np.float32)

     () -> :
        ._mean[:] = 
        ._var[:] = 
        ._count = ._epsilon

 ():
    
     ():
        ().__init__(env)
        ._rms = RunningMeanStd(shape=env.observation_space.shape)
        ._epsilon = epsilon
        ._clip = clip
        ._update_count = 

     () -> [np.ndarray, , , , ]:
        obs, reward, terminated, truncated, info = .env.step(action)
        ._rms.update(obs[, :]  obs.ndim ==   obs)
        ._update_count += 
         ._update_count > :
            obs = np.clip(
                (obs - ._rms.mean) / (._rms.std + ._epsilon),
                -._clip, ._clip
            )
         obs, reward, terminated, truncated, info

     () -> [np.ndarray, ]:
        obs, info = .env.reset(**kwargs)
         ._update_count > :
            obs = np.clip(
                (obs - ._rms.mean) / (._rms.std + ._epsilon),
                -._clip, ._clip
            )
         obs, info

 ():
    
     ():
        ().__init__(env)
        ._n_frames = n_frames
        ._frames: deque = deque(maxlen=n_frames)
        low = np.repeat(env.observation_space.low[np.newaxis, ...], n_frames, axis=)
        high = np.repeat(env.observation_space.high[np.newaxis, ...], n_frames, axis=)
        .observation_space = spaces.Box(
            low=low,
            high=high,
            dtype=env.observation_space.dtype
        )

     () -> [np.ndarray, ]:
        obs, info = .env.reset(**kwargs)
         _  (._n_frames):
            ._frames.append(obs)
         ._get_obs(), info

     () -> [np.ndarray, , , , ]:
        obs, reward, terminated, truncated, info = .env.step(action)
        ._frames.append(obs)
         ._get_obs(), reward, terminated, truncated, info

     () -> np.ndarray:
        
         np.stack(._frames, axis=)

 :
    
     ():
        ._env_id = env_id
        ._num_envs = num_envs
        ._use_async = use_async
        ._wrappers = wrappers  []
        ._seed = seed

         () -> :
             () -> Env:
                env = gym.make(env_id)
                env.reset(seed=seed + env_idx)
                 wrapper  ._wrappers:
                    env = wrapper(env)
                 env
             _init

        env_fns = [make_env(i)  i  (num_envs)]

         use_async:
            ._envs = AsyncVectorEnv(
                env_fns,
                context=  torch.cuda.is_available()  ,
                shared_memory=
            )
        :
            ._envs = SyncVectorEnv(env_fns)

        ._setup_threading()

     () -> :
         os
        os.environ[] = 
        os.environ[] = 
        os.environ[] = 

     () -> [np.ndarray, ]:
         ._envs.reset()

     () -> [np.ndarray, np.ndarray, np.ndarray, np.ndarray, ]:
         ._envs.step(actions)

     () -> :
        ._envs.close()


     () -> spaces.Space:
         ._envs.single_observation_space


     () -> spaces.Space:
         ._envs.single_action_space

 () -> :
    env = OptimizedVectorEnv(
        env_id=env_id,
        num_envs=num_envs,
        use_async=use_async,
        wrappers=[ e: NormalizeObservation(e),  e: FrameStack(e, n_frames=)]
    )
    obs, _ = env.reset()
    actions = np.array([env.single_action_space.sample()  _  (num_envs)])
    start_time = time.perf_counter()
    steps = 
     steps < total_steps:
        obs, reward, terminated, truncated, info = env.step(actions)
        steps += num_envs
        time.sleep()
         np.(terminated | truncated):
            
    elapsed = time.perf_counter() - start_time
    fps = total_steps / elapsed
    env.close()
     fps

 __name__ == :
    parser = argparse.ArgumentParser(description=)
    parser.add_argument(, =, default=)
    parser.add_argument(, =, default=)
    parser.add_argument(, action=, =)
    parser.add_argument(, action=, =)
    args = parser.parse_args()
     args.benchmark:
        fps = benchmark_sampling(args.env_id, args.num_envs, use_async=args.async_mode)
        ()
    :
        env = OptimizedVectorEnv(
            env_id=args.env_id,
            num_envs=args.num_envs,
            use_async=args.async_mode
        )
        obs, _ = env.reset()
        ()
        env.close()

相关免费在线工具

加密/解密文本

使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。在线工具，加密/解密文本在线工具，online

RSA密钥对生成器

生成新的随机RSA私钥和公钥pem证书。在线工具，RSA密钥对生成器在线工具，online

Mermaid 预览与可视化编辑

基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。在线工具，Mermaid 预览与可视化编辑在线工具，online

curl 转代码

解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。在线工具，curl 转代码在线工具，online

Base64 字符串编码/解码

将字符串编码和解码为其 Base64 格式表示形式即可。在线工具，Base64 字符串编码/解码在线工具，online

Base64 文件转换器

将字符串、文件或图像转换为其 Base64 表示形式。在线工具，Base64 文件转换器在线工具，online

#!/usr/bin/env python3 """ 脚本内容：优先经验回放（PER）与多步回报缓冲区实现使用方式：python section_1_1_2_prioritized_replay.py --capacity 100000 --alpha 0.6 功能说明： 1. SumTree 线段树实现 O(log n) 优先级采样 2. 支持重要性采样权重（IS weights）的批量计算与偏差校正 3. n-step 回报计算采用环形缓冲区，零内存泄漏设计 4. 提供与 PyTorch DataLoader 兼容的批次采样接口 """ import argparse import random from dataclasses import dataclass from typing import List, Tuple, Optional import numpy as np import torch from torch import Tensor @dataclass class Transition: obs: np.ndarray action: int reward: float next_obs: np.ndarray done: bool class SumTree: def __init__(self, capacity: int): self._capacity = capacity self._tree = np.zeros(2 * capacity - 1, dtype=np.float64) self._data_idx = 0 def _propagate(self, idx: int, change: float) -> None: parent = (idx - 1) // 2 self._tree[parent] += change if parent != 0: self._propagate(parent, change) def _retrieve(self, idx: int, s: float) -> int: left = 2 * idx + 1 right = left + 1 if left >= len(self._tree): return idx if s <= self._tree[left]: return self._retrieve(left, s) else: return self._retrieve(right, s - self._tree[left]) def total(self) -> float: return self._tree[0] def add(self, priority: float) -> int: idx = self._data_idx + self._capacity - 1 self._update(idx, priority) self._data_idx = (self._data_idx + 1) % self._capacity return self._data_idx def _update(self, idx: int, priority: float) -> None: change = priority - self._tree[idx] self._tree[idx] = priority self._propagate(idx, change) def get(self, s: float) -> Tuple[int, float]: idx = self._retrieve(0, s) data_idx = idx - (self._capacity - 1) return data_idx, self._tree[idx] def sample(self, batch_size: int) -> Tuple[np.ndarray, np.ndarray]: indices = np.zeros(batch_size, dtype=np.int64) priorities = np.zeros(batch_size, dtype=np.float64) total = self.total() segment = total / batch_size for i in range(batch_size): low = segment * i high = segment * (i + 1) s = random.uniform(low, high) idx, priority = self.get(s) indices[i] = idx priorities[i] = priority probs = priorities / total return indices, probs class NStepCircularBuffer: def __init__(self, n_step: int, gamma: float, num_envs: int = 1): self._n = n_step self._gamma = gamma self._num_envs = num_envs self._obs_buffer = [deque(maxlen=n_step) for _ in range(num_envs)] self._action_buffer = [deque(maxlen=n_step) for _ in range(num_envs)] self._reward_buffer = [deque(maxlen=n_step) for _ in range(num_envs)] def add(self, obs: np.ndarray, action: np.ndarray, reward: np.ndarray, done: np.ndarray) -> List[Optional[Tuple]]: completed_transitions = [] for env_idx in range(self._num_envs): self._obs_buffer[env_idx].append(obs[env_idx]) self._action_buffer[env_idx].append(action[env_idx]) self._reward_buffer[env_idx].append(reward[env_idx]) if len(self._reward_buffer[env_idx]) == self._n or done[env_idx]: n_step_return = 0.0 for i, r in enumerate(self._reward_buffer[env_idx]): n_step_return += (self._gamma ** i) * r first_obs = self._obs_buffer[env_idx][0] first_action = self._action_buffer[env_idx][0] next_obs = obs[env_idx] if not done[env_idx] else np.zeros_like(obs[env_idx]) transition = (first_obs, first_action, n_step_return, next_obs, done[env_idx]) completed_transitions.append(transition) if done[env_idx]: self._obs_buffer[env_idx].clear() self._action_buffer[env_idx].clear() self._reward_buffer[env_idx].clear() else: completed_transitions.append(None) return completed_transitions class PrioritizedReplayBuffer: def __init__( self, capacity: int, obs_shape: Tuple[int, ...], alpha: float = 0.6, beta_start: float = 0.4, beta_frames: int = 100000, eps: float = 1e-6 ): self._capacity = capacity self._alpha = alpha self._beta_start = beta_start self._beta_frames = beta_frames self._eps = eps self._tree = SumTree(capacity) self._data = np.zeros(capacity, dtype=object) self._frame = 0 self._obs_shape = obs_shape @property def beta(self) -> float: return min(1.0, self._beta_start + self._frame * (1.0 - self._beta_start) / self._beta_frames) def add(self, transition: Transition, td_error: Optional[float] = None) -> None: if td_error is None: priority = self._tree.max_priority() if self._tree.total() > 0 else 1.0 else: priority = (abs(td_error) + self._eps) ** self._alpha idx = self._tree.add(priority) self._data[idx] = transition self._frame += 1 def sample(self, batch_size: int, device: str = "cpu") -> Tuple[Tensor, ...]: indices, probs = self._tree.sample(batch_size) weights = (self._capacity * probs) ** (-self.beta) weights /= weights.max() batch = [self._data[idx] for idx in indices] obs = torch.stack([torch.from_numpy(t.obs) for t in batch]).to(device) actions = torch.tensor([t.action for t in batch], dtype=torch.long, device=device) rewards = torch.tensor([t.reward for t in batch], dtype=torch.float32, device=device) next_obs = torch.stack([torch.from_numpy(t.next_obs) for t in batch]).to(device) dones = torch.tensor([t.done for t in batch], dtype=torch.float32, device=device) weights = torch.from_numpy(weights).float().to(device) return obs, actions, rewards, next_obs, dones, indices, weights def update_priorities(self, indices: np.ndarray, td_errors: np.ndarray) -> None: for idx, td_error in zip(indices, td_errors): priority = (abs(td_error) + self._eps) ** self._alpha tree_idx = idx + self._capacity - 1 self._tree._update(tree_idx, priority) def __len__(self) -> int: return min(self._frame, self._capacity) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--capacity", type=int, default=100000) parser.add_argument("--alpha", type=float, default=0.6) parser.add_argument("--test_samples", type=int, default=1000) args = parser.parse_args() buffer = PrioritizedReplayBuffer( capacity=args.capacity, obs_shape=(4,), alpha=args.alpha ) print("填充缓冲区...") for i in range(args.test_samples): trans = Transition( obs=np.random.randn(4), action=np.random.randint(2), reward=np.random.randn(), next_obs=np.random.randn(4), done=random.random() < 0.1 ) td_error = random.random() buffer.add(trans, td_error) print("测试优先采样...") obs, actions, rewards, next_obs, dones, indices, weights = buffer.sample(32) print(f"批次观测形状：{obs.shape}, IS 权重范围：[{weights.min():.3f}, {weights.max():.3f}]") print(f"当前 beta 值：{buffer.beta:.3f}")

#!/usr/bin/env python3 """ 脚本内容：DQN 网络架构优化与目标网络软更新使用方式：python section_1_2_1_dqn_architecture.py --env_id CartPole-v1 --tau 0.005 功能说明： 1. 实现 Nature DQN 卷积架构（3 层 Conv + FC）与 LayerNorm 变体 2. 对比 LayerNorm 与 BatchNorm 在 RL 场景下的数值稳定性 3. Polyak 平均软更新实现，包含 PyTorch 原地操作优化 4. 目标网络梯度流隔离与参数冻结机制 """ import argparse import copy from typing import List, Tuple import torch import torch.nn as nn import torch.nn.functional as F from torch import Tensor class NatureDQN(nn.Module): def __init__(self, n_actions: int, n_input_channels: int = 4): super().__init__() self._n_actions = n_actions self.conv = nn.Sequential( nn.Conv2d(n_input_channels, 32, kernel_size=8, stride=4), nn.ReLU(), nn.Conv2d(32, 64, kernel_size=4, stride=2), nn.ReLU(), nn.Conv2d(64, 64, kernel_size=3, stride=1), nn.ReLU() ) self.fc = nn.Sequential( nn.Linear(64 * 7 * 7, 512), nn.ReLU(), nn.Linear(512, n_actions) ) self._init_weights() def _init_weights(self) -> None: for m in self.modules(): if isinstance(m, nn.Conv2d): nn.init.kaiming_normal_(m.weight, mode='fan_out', nonlinearity='relu') if m.bias is not None: nn.init.constant_(m.bias, 0) elif isinstance(m, nn.Linear): nn.init.kaiming_normal_(m.weight, mode='fan_out', nonlinearity='relu') nn.init.constant_(m.bias, 0) def forward(self, x: Tensor) -> Tensor: x = x.float() / 255.0 conv_out = self.conv(x) flat = conv_out.view(conv_out.size(0), -1) return self.fc(flat) class LayerNormDQN(nn.Module): def __init__(self, n_actions: int, n_input_channels: int = 4): super().__init__() self.conv = nn.Sequential( nn.Conv2d(n_input_channels, 32, kernel_size=8, stride=4), nn.LayerNorm([32, 20, 20]), nn.ReLU(), nn.Conv2d(32, 64, kernel_size=4, stride=2), nn.LayerNorm([64, 9, 9]), nn.ReLU(), nn.Conv2d(64, 64, kernel_size=3, stride=1), nn.LayerNorm([64, 7, 7]), nn.ReLU() ) self.fc = nn.Sequential( nn.Linear(64 * 7 * 7, 512), nn.LayerNorm(512), nn.ReLU(), nn.Linear(512, n_actions) ) def forward(self, x: Tensor) -> Tensor: x = x.float() / 255.0 conv_out = self.conv(x) flat = conv_out.view(conv_out.size(0), -1) return self.fc(flat) class SoftUpdateManager: def __init__(self, online_net: nn.Module, target_net: nn.Module, tau: float = 0.005): self._online = online_net self._target = target_net self._tau = tau self.hard_update() for param in self._target.parameters(): param.requires_grad = False def soft_update(self) -> None: with torch.no_grad(): for target_param, online_param in zip( self._target.parameters(), self._online.parameters() ): target_param.data.mul_(1.0 - self._tau) target_param.data.add_(online_param.data * self._tau) def hard_update(self) -> None: self._target.load_state_dict(self._online.state_dict()) def validate_sync(self) -> float: max_diff = 0.0 with torch.no_grad(): for t, o in zip(self._target.parameters(), self._online.parameters()): diff = (t - o).abs().max().item() max_diff = max(max_diff, diff) return max_diff class DQNAgent: def __init__( self, n_actions: int, obs_shape: Tuple[int, ...], architecture: str = "nature", lr: float = 1e-4, gamma: float = 0.99, tau: float = 0.005, device: str = "cuda" if torch.cuda.is_available() else "cpu" ): self._n_actions = n_actions self._gamma = gamma self._device = device if architecture == "nature": self._online_net = NatureDQN(n_actions).to(device) self._target_net = NatureDQN(n_actions).to(device) elif architecture == "layernorm": self._online_net = LayerNormDQN(n_actions).to(device) self._target_net = LayerNormDQN(n_actions).to(device) else: raise ValueError(f"未知架构：{architecture}") self._optimizer = torch.optim.Adam( self._online_net.parameters(), lr=lr, eps=1.5e-4 ) self._target_mgr = SoftUpdateManager(self._online_net, self._target_net, tau) self._train_steps = 0 def select_action(self, obs: np.ndarray, epsilon: float = 0.0) -> int: if np.random.random() < epsilon: return np.random.randint(self._n_actions) with torch.no_grad(): obs_t = torch.from_numpy(obs).unsqueeze(0).to(self._device) q_values = self._online_net(obs_t) return q_values.argmax(dim=1).item() def compute_loss( self, obs: Tensor, actions: Tensor, rewards: Tensor, next_obs: Tensor, dones: Tensor ) -> Tuple[Tensor, np.ndarray]: current_q = self._online_net(obs).gather(1, actions.unsqueeze(1)).squeeze(1) with torch.no_grad(): next_q = self._target_net(next_obs).max(dim=1)[0] target_q = rewards + self._gamma * next_q * (1 - dones) loss = F.smooth_l1_loss(current_q, target_q) td_errors = (target_q - current_q).abs().detach().cpu().numpy() return loss, td_errors def train_step(self, batch: Tuple[Tensor, ...]) -> float: obs, actions, rewards, next_obs, dones = batch loss, td_errors = self.compute_loss(obs, actions, rewards, next_obs, dones) self._optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(self._online_net.parameters(), 10.0) self._optimizer.step() self._target_mgr.soft_update() self._train_steps += 1 return loss.item() if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--architecture", type=str, default="nature", choices=["nature", "layernorm"]) parser.add_argument("--tau", type=float, default=0.005, help="软更新系数") parser.add_argument("--validate", action="store_true", help="验证目标网络同步") args = parser.parse_args() device = "cuda" if torch.cuda.is_available() else "cpu" agent = DQNAgent( n_actions=4, obs_shape=(4, 84, 84), architecture=args.architecture, tau=args.tau, device=device ) dummy_batch = ( torch.randn(32, 4, 84, 84, device=device), torch.randint(0, 4, (32,), device=device), torch.randn(32, device=device), torch.randn(32, 4, 84, 84, device=device), torch.zeros(32, device=device) ) loss = agent.train_step(dummy_batch) print(f"架构：{args.architecture}, 初始损失：{loss:.4f}") if args.validate: diff = agent._target_mgr.validate_sync() print(f"目标 - 在线网络最大参数差异：{diff:.6f} (应接近 {args.tau})")

#!/usr/bin/env python3 """ 脚本内容：C51 分布 DQN 与 Rainbow 六合一集成实现使用方式：python section_1_2_2_rainbow.py --n_atoms 51 --v_min -10 --v_max 10 功能说明： 1. C51 投影算子实现，支持分布贝尔曼更新在离散支撑点上的质量分配 2. Dueling 网络架构，独立价值流与优势流组合 3. NoisyNets 因子化高斯噪声层，替代 epsilon-贪心探索 4. Double DQN 与优先回放集成，完整 Rainbow 六合一实现 """ import argparse import math from typing import Tuple import torch import torch.nn as nn import torch.nn.functional as F from torch import Tensor class FactorizedNoisyLinear(nn.Module): def __init__(self, in_features: int, out_features: int, sigma_init: float = 0.5): super().__init__() self._in_features = in_features self._out_features = out_features self.weight_mu = nn.Parameter(torch.empty(out_features, in_features)) self.bias_mu = nn.Parameter(torch.empty(out_features)) self.weight_sigma = nn.Parameter(torch.empty(out_features, in_features)) self.bias_sigma = nn.Parameter(torch.empty(out_features)) self.register_buffer("weight_epsilon", torch.empty(out_features, in_features)) self.register_buffer("bias_epsilon", torch.empty(out_features)) self._sigma_init = sigma_init self.reset_parameters() self.reset_noise() def reset_parameters(self) -> None: mu_range = 1 / math.sqrt(self._in_features) self.weight_mu.data.uniform_(-mu_range, mu_range) self.bias_mu.data.uniform_(-mu_range, mu_range) self.weight_sigma.data.fill_(self._sigma_init / math.sqrt(self._in_features)) self.bias_sigma.data.fill_(self._sigma_init / math.sqrt(self._out_features)) def reset_noise(self) -> None: epsilon_in = torch.randn(self._in_features) epsilon_out = torch.randn(self._out_features) self.weight_epsilon.copy_(epsilon_out.outer(epsilon_in).sign() * (epsilon_out.outer(epsilon_in)).abs().sqrt_()) self.bias_epsilon.copy_(epsilon_out.sign() * epsilon_out.abs().sqrt_()) def forward(self, x: Tensor, use_noise: bool = True) -> Tensor: if use_noise: weight = self.weight_mu + self.weight_sigma * self.weight_epsilon bias = self.bias_mu + self.bias_sigma * self.bias_epsilon else: weight = self.weight_mu bias = self.bias_mu return F.linear(x, weight, bias) class DuelingC51Network(nn.Module): def __init__( self, n_actions: int, n_input_channels: int = 4, n_atoms: int = 51, v_min: float = -10.0, v_max: float = 10.0, noisy: bool = True ): super().__init__() self._n_actions = n_actions self._n_atoms = n_atoms self._v_min = v_min self._v_max = v_max self.register_buffer("support", torch.linspace(v_min, v_max, n_atoms)) self._delta_z = (v_max - v_min) / (n_atoms - 1) self.features = nn.Sequential( nn.Conv2d(n_input_channels, 32, kernel_size=8, stride=4), nn.ReLU(), nn.Conv2d(32, 64, kernel_size=4, stride=2), nn.ReLU(), nn.Conv2d(64, 64, kernel_size=3, stride=1), nn.ReLU() ) linear_cls = FactorizedNoisyLinear if noisy else nn.Linear self.value_stream = nn.Sequential( linear_cls(64 * 7 * 7, 512), nn.ReLU(), linear_cls(512, n_atoms) ) self.advantage_stream = nn.Sequential( linear_cls(64 * 7 * 7, 512), nn.ReLU(), linear_cls(512, n_actions * n_atoms) ) def forward(self, x: Tensor, use_noise: bool = True) -> Tensor: x = x.float() / 255.0 conv_out = self.features(x).view(x.size(0), -1) if isinstance(self.value_stream[0], FactorizedNoisyLinear): value_logits = self.value_stream[0](conv_out, use_noise) for layer in self.value_stream[1:]: if isinstance(layer, FactorizedNoisyLinear): value_logits = layer(value_logits, use_noise) else: value_logits = layer(value_logits) else: value_logits = self.value_stream(conv_out) value_logits = value_logits.view(-1, 1, self._n_atoms) if isinstance(self.advantage_stream[0], FactorizedNoisyLinear): adv_logits = self.advantage_stream[0](conv_out, use_noise) for layer in self.advantage_stream[1:]: if isinstance(layer, FactorizedNoisyLinear): adv_logits = layer(adv_logits, use_noise) else: adv_logits = layer(adv_logits) else: adv_logits = self.advantage_stream(conv_out) adv_logits = adv_logits.view(-1, self._n_actions, self._n_atoms) mean_adv = adv_logits.mean(dim=1, keepdim=True) logits = value_logits + adv_logits - mean_adv probs = F.softmax(logits, dim=2) return probs def reset_noise(self) -> None: for module in self.modules(): if isinstance(module, FactorizedNoisyLinear): module.reset_noise() @property def support(self) -> Tensor: return self._support def projection_operator( next_probs: Tensor, rewards: Tensor, dones: Tensor, support: Tensor, gamma: float, v_min: float, v_max: float, n_atoms: int ) -> Tensor: batch_size = next_probs.size(0) delta_z = (v_max - v_min) / (n_atoms - 1) rewards = rewards.unsqueeze(1) dones = dones.unsqueeze(1) support = support.unsqueeze(0) projected = rewards + gamma * (1 - dones) * support projected = projected.clamp(v_min, v_max) b = (projected - v_min) / delta_z l = b.floor().long() u = b.ceil().long() l[(l < 0)] = 0 u[(u > n_atoms - 1)] = n_atoms - 1 u_probs = (b - l.float()) l_probs = 1.0 - u_probs target_probs = torch.zeros(batch_size, n_atoms, device=next_probs.device) flat_probs = next_probs.view(-1, n_atoms) l_indices = l.view(batch_size, n_atoms, 1).expand(-1, -1, n_atoms) l_values = (flat_probs * l_probs).view(batch_size, n_atoms) u_indices = u.view(batch_size, n_atoms, 1).expand(-1, -1, n_atoms) u_values = (flat_probs * u_probs).view(batch_size, n_atoms) for i in range(batch_size): for j in range(n_atoms): target_probs[i, l[i, j]] += l_values[i, j] target_probs[i, u[i, j]] += u_values[i, j] return target_probs class RainbowAgent: def __init__( self, n_actions: int, n_atoms: int = 51, v_min: float = -10.0, v_max: float = 10.0, gamma: float = 0.99, lr: float = 1e-4, device: str = "cuda" if torch.cuda.is_available() else "cpu" ): self._n_actions = n_actions self._n_atoms = n_atoms self._gamma = gamma self._device = device self._online_net = DuelingC51Network( n_actions, n_atoms=n_atoms, v_min=v_min, v_max=v_max, noisy=True ).to(device) self._target_net = DuelingC51Network( n_actions, n_atoms=n_atoms, v_min=v_min, v_max=v_max, noisy=True ).to(device) self._target_net.load_state_dict(self._online_net.state_dict()) for param in self._target_net.parameters(): param.requires_grad = False self._optimizer = torch.optim.Adam(self._online_net.parameters(), lr=lr) def select_action(self, obs: Tensor) -> int: with torch.no_grad(): dist = self._online_net(obs.unsqueeze(0), use_noise=True) support = self._online_net.support.unsqueeze(0).unsqueeze(0) q_values = (dist * support).sum(dim=2) return q_values.argmax(dim=1).item() def compute_distribution_loss( self, obs: Tensor, actions: Tensor, rewards: Tensor, next_obs: Tensor, dones: Tensor ) -> Tuple[Tensor, np.ndarray]: batch_size = obs.size(0) current_dist = self._online_net(obs, use_noise=True) current_dist = current_dist[range(batch_size), actions] with torch.no_grad(): next_dist_online = self._online_net(next_obs, use_noise=False) support = self._online_net.support.unsqueeze(0).unsqueeze(0) next_q_online = (next_dist_online * support).sum(dim=2) next_actions = next_q_online.argmax(dim=1) next_dist_target = self._target_net(next_obs, use_noise=False) next_dist = next_dist_target[range(batch_size), next_actions] target_dist = projection_operator( next_dist, rewards, dones, self._online_net.support, self._gamma, self._online_net._v_min, self._online_net._v_max, self._n_atoms ) loss = -(target_dist * torch.log(current_dist + 1e-8)).sum(dim=1).mean() with torch.no_grad(): kl_div = (target_dist * (torch.log(target_dist + 1e-8) - torch.log(current_dist + 1e-8))).sum(dim=1) return loss, kl_div.cpu().numpy() def update_target(self) -> None: self._target_net.load_state_dict(self._online_net.state_dict()) def reset_noise(self) -> None: self._online_net.reset_noise() self._target_net.reset_noise() if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--n_atoms", type=int, default=51) parser.add_argument("--v_min", type=float, default=-10.0) parser.add_argument("--v_max", type=float, default=10.0) parser.add_argument("--test_projection", action="store_true") args = parser.parse_args() agent = RainbowAgent( n_actions=4, n_atoms=args.n_atoms, v_min=args.v_min, v_max=args.v_max ) dummy_obs = torch.randn(2, 4, 84, 84, device=agent._device) dist = agent._online_net(dummy_obs, use_noise=True) print(f"分布输出形状：{dist.shape}, 概率和验证：{dist.sum(dim=2)}") if args.test_projection: next_probs = torch.randn(2, args.n_atoms, device=agent._device).softmax(dim=1) rewards = torch.tensor([1.0, 0.0], device=agent._device) dones = torch.tensor([0.0, 1.0], device=agent._device) target = projection_operator( next_probs.unsqueeze(1).expand(-1, 4, -1), rewards, dones, agent._online_net.support, 0.99, args.v_min, args.v_max, args.n_atoms ) print(f"投影后分布形状：{target.shape}, 质量守恒验证：{target.sum(dim=1)}")

#!/usr/bin/env python3 """ 脚本内容：REINFORCE 策略梯度与可学习基线实现使用方式：python section_1_3_1_reinforce.py --env_id CartPole-v1 --n_episodes 1000 功能说明： 1. Monte-Carlo 梯度估计：利用 torch.distributions.Categorical 实现 log_prob 自动微分 2. 可学习状态价值网络作为基线，实现方差缩减 3. 策略网络与价值网络共享特征提取层或独立优化（可选配置） 4. 包含梯度累积与归一化技巧，确保训练稳定性 """ import argparse from typing import List, Tuple import gymnasium as gym import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from torch import Tensor from torch.distributions import Categorical class PolicyNetwork(nn.Module): def __init__(self, obs_dim: int, action_dim: int, hidden_dim: int = 128): super().__init__() self._shared = nn.Sequential( nn.Linear(obs_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU() ) self._policy_head = nn.Linear(hidden_dim, action_dim) nn.init.orthogonal_(self._policy_head.weight, gain=0.01) nn.init.constant_(self._policy_head.bias, 0) def forward(self, obs: Tensor) -> Tensor: features = self._shared(obs) return self._policy_head(features) def get_action_and_log_prob(self, obs: Tensor) -> Tuple[Tensor, Tensor]: logits = self.forward(obs) dist = Categorical(logits=logits) action = dist.sample() log_prob = dist.log_prob(action) return action, log_prob class ValueNetwork(nn.Module): def __init__(self, obs_dim: int, hidden_dim: int = 128): super().__init__() self._value_net = nn.Sequential( nn.Linear(obs_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, 1) ) nn.init.orthogonal_(self._value_net[-1].weight, gain=1.0) def forward(self, obs: Tensor) -> Tensor: return self._value_net(obs).squeeze(-1) class REINFORCEAgent: def __init__( self, obs_dim: int, action_dim: int, gamma: float = 0.99, lr_policy: float = 1e-3, lr_value: float = 1e-3, device: str = "cuda" if torch.cuda.is_available() else "cpu" ): self._gamma = gamma self._device = device self._policy_net = PolicyNetwork(obs_dim, action_dim).to(device) self._value_net = ValueNetwork(obs_dim).to(device) self._policy_opt = torch.optim.Adam(self._policy_net.parameters(), lr=lr_policy) self._value_opt = torch.optim.Adam(self._value_net.parameters(), lr=lr_value) self._returns = [] self._policy_losses = [] def collect_trajectory(self, env: gym.Env, max_steps: int = 1000) -> Tuple[List[Tensor], ...]: obs_list, action_list, log_prob_list, reward_list = [], [], [], [] obs, _ = env.reset() done = False steps = 0 while not done and steps < max_steps: obs_t = torch.from_numpy(obs).float().to(self._device) with torch.no_grad(): action, log_prob = self._policy_net.get_action_and_log_prob(obs_t) next_obs, reward, terminated, truncated, _ = env.step(action.item()) done = terminated or truncated obs_list.append(obs_t) action_list.append(action) log_prob_list.append(log_prob) reward_list.append(reward) obs = next_obs steps += 1 return obs_list, action_list, log_prob_list, reward_list def compute_returns(self, rewards: List[float]) -> Tensor: returns = [] G = 0.0 for r in reversed(rewards): G = r + self._gamma * G returns.insert(0, G) returns = torch.tensor(returns, dtype=torch.float32, device=self._device) returns = (returns - returns.mean()) / (returns.std() + 1e-8) return returns def update(self, trajectory: Tuple[List[Tensor], ...]) -> Tuple[float, float]: obs_list, _, log_prob_list, reward_list = trajectory returns = self.compute_returns(reward_list) obs_batch = torch.stack(obs_list) values = self._value_net(obs_batch) advantages = returns - values.detach() log_probs = torch.stack(log_prob_list) policy_loss = -(log_probs * advantages).mean() value_loss = F.mse_loss(values, returns) self._policy_opt.zero_grad() policy_loss.backward() torch.nn.utils.clip_grad_norm_(self._policy_net.parameters(), 0.5) self._policy_opt.step() self._value_opt.zero_grad() value_loss.backward() torch.nn.utils.clip_grad_norm_(self._value_net.parameters(), 0.5) self._value_opt.step() return policy_loss.item(), value_loss.item() def train(self, env: gym.Env, n_episodes: int = 1000, log_interval: int = 100) -> None: episode_returns = [] for episode in range(n_episodes): trajectory = self.collect_trajectory(env) policy_loss, value_loss = self.update(trajectory) total_return = sum(trajectory[3]) episode_returns.append(total_return) if episode % log_interval == 0: mean_return = np.mean(episode_returns[-log_interval:]) print(f"Episode {episode}, 平均回报：{mean_return:.2f}, 策略损失：{policy_loss:.4f}, 价值损失：{value_loss:.4f}") return episode_returns if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--env_id", type=str, default="CartPole-v1") parser.add_argument("--n_episodes", type=int, default=1000) parser.add_argument("--gamma", type=float, default=0.99) parser.add_argument("--lr_policy", type=float, default=1e-3) parser.add_argument("--lr_value", type=float, default=1e-3) args = parser.parse_args() env = gym.make(args.env_id) obs_dim = env.observation_space.shape[0] action_dim = env.action_space.n agent = REINFORCEAgent( obs_dim=obs_dim, action_dim=action_dim, gamma=args.gamma, lr_policy=args.lr_policy, lr_value=args.lr_value ) print(f"开始训练 REINFORCE on {args.env_id}...") returns = agent.train(env, n_episodes=args.n_episodes) print(f"训练完成，最终 100 回合平均回报：{np.mean(returns[-100:]):.2f}") env.close()

#!/usr/bin/env python3 """ 脚本内容：同步 A2C 与异步 A3C 并行架构实现使用方式：python section_1_3_2_a2c_a3c.py --mode async --num_workers 4 --env_id CartPole-v1 功能说明： 1. 同步 A2C：单 GPU 批量处理多环境，DataParallel 风格前向传播 2. 异步 A3C：多进程 Hogwild 训练，共享内存参数更新与锁自由优化 3. 通用 Actor-Critic 网络，支持策略 logits 与价值输出的并行计算 4. 包含进程间通信优化与梯度累积实现 """ import argparse import os import time from multiprocessing import Process, Queue, shared_memory from typing import Tuple import gymnasium as gym import numpy as np import torch import torch.multiprocessing as mp import torch.nn as nn import torch.nn.functional as F from torch import Tensor class ActorCriticNetwork(nn.Module): def __init__(self, obs_dim: int, action_dim: int, hidden_dim: int = 256): super().__init__() self._shared = nn.Sequential( nn.Linear(obs_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU() ) self._actor = nn.Linear(hidden_dim, action_dim) self._critic = nn.Linear(hidden_dim, 1) nn.init.orthogonal_(self._actor.weight, gain=0.01) nn.init.orthogonal_(self._critic.weight, gain=1.0) def forward(self, obs: Tensor) -> Tuple[Tensor, Tensor]: features = self._shared(obs) logits = self._actor(features) value = self._critic(features).squeeze(-1) return logits, value def get_action_and_value(self, obs: Tensor) -> Tuple[Tensor, Tensor, Tensor]: logits, value = self.forward(obs) dist = torch.distributions.Categorical(logits=logits) action = dist.sample() log_prob = dist.log_prob(action) return action, log_prob, value class A2CTrainer: def __init__( self, env_id: str, num_envs: int = 8, gamma: float = 0.99, gae_lambda: float = 0.95, lr: float = 1e-4, device: str = "cuda" if torch.cuda.is_available() else "cpu" ): self._env_id = env_id self._num_envs = num_envs self._gamma = gamma self._gae_lambda = gae_lambda self._device = device self._envs = gym.vector.SyncVectorEnv([ lambda: gym.make(env_id) for _ in range(num_envs) ]) obs_dim = self._envs.single_observation_space.shape[0] action_dim = self._envs.single_action_space.n self._net = ActorCriticNetwork(obs_dim, action_dim).to(device) self._optimizer = torch.optim.Adam(self._net.parameters(), lr=lr, eps=1e-5) def compute_gae( self, rewards: Tensor, values: Tensor, dones: Tensor, next_value: Tensor ) -> Tuple[Tensor, Tensor]: advantages = torch.zeros_like(rewards) last_gae = 0.0 for t in reversed(range(len(rewards))): if t == len(rewards) - 1: next_non_terminal = 1.0 - dones[t] next_val = next_value else: next_non_terminal = 1.0 - dones[t] next_val = values[t + 1] delta = rewards[t] + self._gamma * next_val * next_non_terminal - values[t] last_gae = delta + self._gamma * self._gae_lambda * next_non_terminal * last_gae advantages[t] = last_gae returns = advantages + values return advantages, returns def train_step(self, n_steps: int = 5) -> Tuple[float, float]: obs, _ = self._envs.reset() obs_batch, action_batch, log_prob_batch = [], [], [] reward_batch, done_batch, value_batch = [], [], [] for _ in range(n_steps): obs_t = torch.from_numpy(obs).float().to(self._device) with torch.no_grad(): action, log_prob, value = self._net.get_action_and_value(obs_t) next_obs, reward, terminated, truncated, _ = self._envs.step(action.cpu().numpy()) done = np.logical_or(terminated, truncated) obs_batch.append(obs_t) action_batch.append(action) log_prob_batch.append(log_prob) reward_batch.append(torch.from_numpy(reward).float().to(self._device)) done_batch.append(torch.from_numpy(done).float().to(self._device)) value_batch.append(value) obs = next_obs with torch.no_grad(): next_obs_t = torch.from_numpy(obs).float().to(self._device) _, next_value = self._net(next_obs_t) obs_batch = torch.stack(obs_batch) action_batch = torch.stack(action_batch) log_prob_batch = torch.stack(log_prob_batch) reward_batch = torch.stack(reward_batch) done_batch = torch.stack(done_batch) value_batch = torch.stack(value_batch) advantages, returns = self.compute_gae( reward_batch, value_batch, done_batch, next_value ) logits, values = self._net(obs_batch.view(-1, obs_batch.size(-1))) dist = torch.distributions.Categorical(logits=logits.view(n_steps, self._num_envs, -1)) new_log_probs = dist.log_prob(action_batch) values = values.view(n_steps, self._num_envs) entropy = dist.entropy().mean() advantage = advantages.detach() policy_loss = -(new_log_probs * advantage).mean() - 0.01 * entropy value_loss = F.mse_loss(values, returns.detach()) loss = policy_loss + 0.5 * value_loss self._optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(self._net.parameters(), 0.5) self._optimizer.step() return policy_loss.item(), value_loss.item() def train(self, total_updates: int = 10000): for update in range(total_updates): p_loss, v_loss = self.train_step() if update % 100 == 0: print(f"[A2C] Update {update}, Policy Loss: {p_loss:.4f}, Value Loss: {v_loss:.4f}") self._envs.close() class A3CWorker(Process): def __init__( self, worker_id: int, env_id: str, shared_net_state: dict, gradient_queue: Queue, stop_event, gamma: float = 0.99 ): super().__init__() self._worker_id = worker_id self._env_id = env_id self._shared_state = shared_net_state self._queue = gradient_queue self._stop_event = stop_event self._gamma = gamma def run(self): env = gym.make(self._env_id) obs_dim = env.observation_space.shape[0] action_dim = env.action_space.n local_net = ActorCriticNetwork(obs_dim, action_dim) local_net.load_state_dict(self._shared_state) optimizer = torch.optim.SGD(local_net.parameters(), lr=1e-3) while not self._stop_event.is_set(): local_net.load_state_dict(self._shared_state) obs, _ = env.reset() log_probs, values, rewards, dones = [], [], [], [] for _ in range(5): obs_t = torch.from_numpy(obs).float().unsqueeze(0) with torch.no_grad(): action, log_prob, value = local_net.get_action_and_value(obs_t) next_obs, reward, terminated, truncated, _ = env.step(action.item()) done = terminated or truncated log_probs.append(log_prob) values.append(value) rewards.append(reward) dones.append(float(done)) obs = next_obs if done: break returns = [] R = 0.0 if done else local_net(torch.from_numpy(obs).float().unsqueeze(0))[1].item() for r in reversed(rewards): R = r + self._gamma * R returns.insert(0, R) returns = torch.tensor(returns) log_probs = torch.stack(log_probs).squeeze() values = torch.stack(values).squeeze() advantages = returns - values.detach() policy_loss = -(log_probs * advantages).sum() value_loss = F.mse_loss(values, returns) loss = policy_loss + 0.5 * value_loss optimizer.zero_grad() loss.backward() grads = [p.grad.cpu().clone() for p in local_net.parameters()] self._queue.put((self._worker_id, grads)) env.close() class A3CTrainer: def __init__( self, env_id: str, num_workers: int = 4, gamma: float = 0.99, lr: float = 1e-3 ): self._env_id = env_id self._num_workers = num_workers self._gamma = gamma dummy_env = gym.make(env_id) obs_dim = dummy_env.observation_space.shape[0] action_dim = dummy_env.action_space.n dummy_env.close() self._shared_net = ActorCriticNetwork(obs_dim, action_dim) self._optimizer = torch.optim.SGD(self._shared_net.parameters(), lr=lr) self._queue = mp.Queue(maxsize=100) self._stop_event = mp.Event() def update_shared(self, gradients: list) -> None: for param, grad in zip(self._shared_net.parameters(), gradients): if param.grad is None: param.grad = grad.clone() else: param.grad.add_(grad) torch.nn.utils.clip_grad_norm_(self._shared_net.parameters(), 40.0) self._optimizer.step() self._optimizer.zero_grad() def train(self, total_updates: int = 10000): init_state = self._shared_net.state_dict() workers = [] for i in range(self._num_workers): worker = A3CWorker( i, self._env_id, init_state, self._queue, self._stop_event, self._gamma ) workers.append(worker) worker.start() updates = 0 try: while updates < total_updates: worker_id, grads = self._queue.get(timeout=10.0) self.update_shared(grads) updates += 1 if updates % 100 == 0: print(f"[A3C] 已处理 {updates} 个异步梯度更新") except Exception as e: print(f"训练中断：{e}") finally: self._stop_event.set() for w in workers: w.join(timeout=5.0) print("[A3C] 训练完成") if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--mode", type=str, default="sync", choices=["sync", "async"]) parser.add_argument("--env_id", type=str, default="CartPole-v1") parser.add_argument("--num_envs", type=int, default=8, help="A2C 环境数") parser.add_argument("--num_workers", type=int, default=4, help="A3C 进程数") parser.add_argument("--total_updates", type=int, default=1000) args = parser.parse_args() if args.mode == "sync": print(f"启动同步 A2C，{args.num_envs} 环境并行...") trainer = A2CTrainer(args.env_id, num_envs=args.num_envs) trainer.train(args.total_updates) else: print(f"启动异步 A3C，{args.num_workers} 工作进程...") mp.set_start_method("spawn", force=True) trainer = A3CTrainer(args.env_id, num_workers=args.num_workers) trainer.train(args.total_updates)

深度强化学习基础架构与核心算法实现

第一章基础架构与核心算法实现

1.1 环境接口与数据流水线

1.1.1 Gymnasium API 深度封装与向量化

更多推荐文章

相关免费在线工具

1.1.2 经验回放缓冲区工程实现

1.2 值函数方法代码实战

1.2.1 DQN 及其网络架构优化

1.2.2 分布式值函数与 Rainbow 集成

1.3 策略梯度与基础 Actor-Critic

1.3.1 REINFORCE 与方差缩减技术

1.3.2 A2C/A3C 的并行架构

深度强化学习基础架构与核心算法实现

第一章 基础架构与核心算法实现

1.1 环境接口与数据流水线

1.1.1 Gymnasium API 深度封装与向量化

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具

1.1.2 经验回放缓冲区工程实现

1.2 值函数方法代码实战

1.2.1 DQN 及其网络架构优化

1.2.2 分布式值函数与 Rainbow 集成

1.3 策略梯度与基础 Actor-Critic

1.3.1 REINFORCE 与方差缩减技术

1.3.2 A2C/A3C 的并行架构

第一章基础架构与核心算法实现