深度确定性策略梯度算法 (DDPG) 详解与实现

3. 目标网络更新

目标网络采用软更新方式，缓慢地向当前网络靠近：

![\theta^{Q'} \leftarrow \tau \theta^Q + (1 - \tau) \theta^{Q'}] ![\theta^{\mu'} \leftarrow \tau \theta^\mu + (1 - \tau) \theta^{\mu'}]

其中 $\tau \in (0, 1)$ 是软更新系数。

算法流程

1. 初始化: 初始化 Actor、Critic 网络和它们对应的目标网络，初始化经验回放池。
2. 交互环境: 在状态 $s_t$ 下，通过 Actor 网络生成动作 $a_t$，执行动作获取奖励 $r_t$ 和下一状态 $s_{t+1}$。
3. 存储经验: 将 $(s_t, a_t, r_t, s_{t+1})$ 存储到经验回放池。
4. 采样训练: 从经验池中随机采样小批量数据 $(s_i, a_i, r_i, s'_i)$。
5. 更新 Critic 网络: 计算目标 Q 值，最小化 Critic 的损失函数。
6. 更新 Actor 网络: 使用 Critic 网络的梯度来调整 Actor 网络的参数。
7. 目标网络更新: 按照软更新公式更新目标网络的参数。
8. 重复以上步骤, 直到达到学习目标。

Python 实现

下面给出 DDPG 算法的完整 Python 实现。该实现包括 Actor-Critic 架构、缓冲区和目标网络等。代码基于 PyTorch 编写，适合理解算法原理及复现。

1. 导入必要库

import gym
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from collections import deque
import random

这里主要用到 gym 创建环境，numpy 处理数组，torch 构建神经网络，deque 实现经验回放池。

2. 定义 Actor 网络

Actor 网络的作用是生成给定状态下的最优动作。

class Actor(nn.Module):
    def __init__(self, state_dim, action_dim, max_action):
        super(Actor, self).__init__()
        self.layer1 = nn.Linear(state_dim, 256)
        self.layer2 = nn.Linear(256, 256)
        self.layer3 = nn.Linear(256, action_dim)
        self.max_action = max_action

    def forward(self, state):
        x = torch.relu(self.layer1(state))
        x = torch.relu(self.layer2(x))
        x = torch.tanh(self.layer3(x)) * self.max_action
        return x

输入包含状态维度、动作维度和最大动作值。网络结构包含两层隐藏层（256 个神经元），输出层使用 tanh 激活函数将动作限制在 [-1, 1]，再乘以 max_action 缩放到实际范围。

3. 定义 Critic 网络

Critic 网络评估给定状态和动作的质量（即 Q 值）。

class Critic(nn.Module):
    def __init__(self, state_dim, action_dim):
        super(Critic, self).__init__()
        self.layer1 = nn.Linear(state_dim + action_dim, 256)
        self.layer2 = nn.Linear(256, 256)
        self.layer3 = nn.Linear(256, 1)

    def forward(self, state, action):
        x = torch.cat([state, action], dim=1)
        x = torch.relu(self.layer1(x))
        x = torch.relu(self.layer2(x))
        x = self.layer3(x)
        return x

状态和动作拼接后作为输入，经过两层隐藏层，输出标量 Q 值。

4. 定义经验回放池

作用：存储智能体与环境交互的经验数据，打破样本间的时间相关性。

class ReplayBuffer:
    def __init__(self, max_size):
        self.buffer = deque(maxlen=max_size)

    def add(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))

    def sample(self, batch_size):
        batch = random.sample(self.buffer, batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)
        return (np.array(states), np.array(actions), np.array(rewards), 
                np.array(next_states), np.array(dones))

    def size(self):
        return len(self.buffer)

5. 定义 DDPG 智能体

class DDPGAgent:
    def __init__(self, state_dim, action_dim, max_action, gamma=0.99, tau=0.005, buffer_size=100000, batch_size=64):
        self.actor = Actor(state_dim, action_dim, max_action)
        self.actor_target = Actor(state_dim, action_dim, max_action)
        self.actor_target.load_state_dict(self.actor.state_dict())
        self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=1e-4)

        self.critic = Critic(state_dim, action_dim)
        self.critic_target = Critic(state_dim, action_dim)
        self.critic_target.load_state_dict(self.critic.state_dict())
        self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=1e-3)

        self.max_action = max_action
        self.gamma = gamma
        self.tau = tau
        self.replay_buffer = ReplayBuffer(buffer_size)
        self.batch_size = batch_size

    def select_action(self, state):
        state = torch.FloatTensor(state.reshape(1, -1))
        action = self.actor(state).detach().cpu().numpy().flatten()
        return action

    def train(self):
        if self.replay_buffer.size() < self.batch_size:
            return

        states, actions, rewards, next_states, dones = self.replay_buffer.sample(self.batch_size)
        states = torch.FloatTensor(states)
        actions = torch.FloatTensor(actions)
        rewards = torch.FloatTensor(rewards).unsqueeze(1)
        next_states = torch.FloatTensor(next_states)
        dones = torch.FloatTensor(dones).unsqueeze(1)

        with torch.no_grad():
            next_actions = self.actor_target(next_states)
            target_q = self.critic_target(next_states, next_actions)
            target_q = rewards + (1 - dones) * self.gamma * target_q

        current_q = self.critic(states, actions)
        critic_loss = nn.MSELoss()(current_q, target_q)

        self.critic_optimizer.zero_grad()
        critic_loss.backward()
        self.critic_optimizer.step()

        actor_loss = -self.critic(states, self.actor(states)).mean()
        self.actor_optimizer.zero_grad()
        actor_loss.backward()
        self.actor_optimizer.step()

        for target_param, param in zip(self.critic_target.parameters(), self.critic.parameters()):
            target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)
        for target_param, param in zip(self.actor_target.parameters(), self.actor.parameters()):
            target_param.data.copy_(self.tau * param.data + (1 - self.tau) * target_param.data)

    def add_to_replay_buffer(self, state, action, reward, next_state, done):
        self.replay_buffer.add(state, action, reward, next_state, done)

这里初始化了 Actor 和 Critic 及其目标网络，并定义了训练循环。注意 Actor 的损失取负号是因为我们要最大化 Q 值，而优化器默认最小化损失。

6. 训练智能体

def train_ddpg(env_name, episodes=1000, max_steps=200):
    env = gym.make(env_name)
    state_dim = env.observation_space.shape[0]
    action_dim = env.action_space.shape[0]
    max_action = float(env.action_space.high[0])

    agent = DDPGAgent(state_dim, action_dim, max_action)
    rewards = []

    for episode in range(episodes):
        state, _ = env.reset()
        episode_reward = 0
        for step in range(max_steps):
            action = agent.select_action(state)
            next_state, reward, done, _, _ = env.step(action)
            agent.add_to_replay_buffer(state, action, reward, next_state, done)
            agent.train()
            state = next_state
            episode_reward += reward
            if done:
                break
        rewards.append(episode_reward)
        print(f"Episode: {episode + 1}, Reward: {episode_reward}")

    plt.plot(rewards)
    plt.title("Learning Curve")
    plt.xlabel("Episodes")
    plt.ylabel("Cumulative Reward")
    plt.show()
    env.close()

if __name__ == "__main__":
    env_name = "Pendulum-v1"
    episodes = 500
    train_ddpg(env_name, episodes=episodes)

主函数中创建了 Pendulum 环境，设置状态和动作维度，初始化智能体并开始训练。训练结束后会绘制学习曲线。

运行结果

训练完成后，可以看到累计奖励随回合数变化的曲线，通常随着训练进行，奖励会逐渐上升并趋于稳定。

注意事项

• 演员和评论家网络: 演员网络预测给定当前状态的动作，批评家网络评估状态 - 行为对的 q 值。
• Replay Buffer: 存储过去的经验，使有效的采样训练成为可能。
• 训练: Critic 使用 Bellman 方程更新，Actor 被更新以最大化期望 q 值。
• 目标网络: 平滑更新以稳定训练。
• 环境: 代理在 Pendulum-v1 环境中进行训练作为演示。

优势与应用

• 解决连续动作问题: 直接输出连续值动作，无需离散化。
• 样本效率高: 使用经验回放和目标网络，减少样本相关性，提高学习效率。

应用场景：机器人运动控制（机械臂、无人机）、自动驾驶中的连续控制任务、游戏中的复杂策略设计。

通俗类比

可以把 DDPG 算法想象成一个赛车手（Actor）和他的教练（Critic）：

• 赛车手（Actor）: 决定转弯的角度、加速的力度，直接控制赛车。
• 教练（Critic）: 观察赛车手的表现，告诉他哪些动作是好的，哪些需要改进。
• 经验回放池: 赛车手回看之前的比赛录像，找到改进的地方。
• 目标网络: 类似于赛车手的长期目标，比如平稳驾驶，而不是今天开得快、明天开得慢。