1. 增强学习基础概念与核心思想
增强学习（Reinforcement Learning）是机器学习领域的一个重要分支，它研究的是智能体（Agent）如何在环境（Environment）中通过试错来学习最优的行为策略。与监督学习不同，增强学习没有预先标注好的"正确答案"，智能体只能通过与环境交互获得的延迟奖励来调整自己的行为策略。
1.1 增强学习的核心要素
增强学习系统由以下几个关键要素组成：
- 智能体（Agent）：做出决策的主体，可以是机器人、游戏AI或其他自主系统
- 环境（Environment）：智能体所处的外部世界，会对智能体的动作做出响应
- 状态（State）：环境的当前情况描述，如游戏画面、传感器读数等
- 动作（Action）：智能体可以执行的操作集合
- 奖励（Reward）：环境对智能体动作的即时反馈信号
- 策略（Policy）：从状态到动作的映射规则，即智能体的决策机制
1.2 增强学习的特点与挑战
增强学习有几个显著特点：
- 试错学习：智能体通过不断尝试来发现哪些动作能带来更好的结果
- 延迟奖励：当前动作的影响可能在很久之后才会显现
- 探索与利用的权衡：需要在尝试新动作（探索）和选择已知好动作（利用）之间取得平衡
这些特点也带来了几个主要挑战：
- 信用分配问题：如何将长期结果归因于特定的早期动作
- 部分可观测性：环境状态可能无法完全被智能体感知
- 大规模状态空间：当状态数量巨大时，如何有效学习和泛化
2. K臂老虎机问题与探索策略
2.1 K臂老虎机问题描述
K臂老虎机是增强学习中最简单的场景之一，它模拟了赌场中的老虎机，但有K个拉杆（动作）可供选择。每个拉杆对应不同的奖励概率分布，智能体的目标是在有限次尝试中最大化累积奖励。
这个问题很好地体现了增强学习中的"探索-利用"困境：是继续尝试已知能获得不错奖励的拉杆（利用），还是尝试其他可能更好的拉杆（探索）。
2.2 常见探索策略实现与比较
2.2.1 ε-贪心算法
ε-贪心算法是最简单的探索策略，它以ε的概率随机选择动作（探索），以1-ε的概率选择当前估值最高的动作（利用）。
python复制def epsilon_greedy(bandit, iterations=1000, epsilon=0.1):
"""ε-贪心算法实现"""
q_est = np.zeros(bandit.k) # 动作价值估计
action_counts = np.zeros(bandit.k) # 动作选择次数
rewards = [] # 记录每一步的奖励
for _ in range(iterations):
# ε-贪心选择动作
if np.random.random() < epsilon:
arm = np.random.choice(bandit.k) # 探索
else:
arm = np.argmax(q_est) # 利用
# 执行动作并获取奖励
reward = bandit.pull(arm)
rewards.append(reward)
# 更新动作价值估计（增量式）
action_counts[arm] += 1
q_est[arm] += (reward - q_est[arm]) / action_counts[arm]
return np.array(rewards)
2.2.2 UCB算法
上置信界（Upper Confidence Bound, UCB）算法通过为每个动作的估值添加一个不确定性奖励来平衡探索与利用。不确定性大的动作会获得更高的探索倾向。
python复制def ucb(bandit, iterations=1000, c=2):
"""UCB算法实现"""
q_est = np.zeros(bandit.k)
action_counts = np.zeros(bandit.k)
rewards = []
# 先每个臂都试一次
for arm in range(bandit.k):
reward = bandit.pull(arm)
rewards.append(reward)
action_counts[arm] += 1
q_est[arm] = reward
# 正式迭代
for t in range(bandit.k, iterations):
# 计算UCB值
ucb_values = q_est + c * np.sqrt(np.log(t + 1) / (action_counts + 1e-6))
arm = np.argmax(ucb_values)
reward = bandit.pull(arm)
rewards.append(reward)
# 更新估计
action_counts[arm] += 1
q_est[arm] += (reward - q_est[arm]) / action_counts[arm]
return np.array(rewards)
2.2.3 策略性能对比
通过实验对比不同探索策略的表现：
| 策略 | 平均奖励 | 最优动作选择率 | 特点 |
|---|---|---|---|
| 贪心(ε=0) | 中等 | 低 | 容易陷入局部最优 |
| ε-贪心(ε=0.1) | 较高 | 中 | 平衡探索与利用 |
| ε-贪心(ε=0.01) | 高 | 高 | 后期表现好但初期慢 |
| UCB(c=2) | 最高 | 最高 | 智能探索，综合表现最佳 |
实际应用中，UCB算法通常表现最好，但ε-贪心更简单易实现。对于非平稳环境（奖励分布会变化），还需要考虑使用衰减的ε或自适应调整的探索策略。
3. 基于模型的增强学习方法
3.1 马尔可夫决策过程（MDP）
基于模型的增强学习通常假设环境是一个马尔可夫决策过程，即下一个状态和奖励只取决于当前状态和动作，与历史无关。MDP可以用五元组(S,A,P,R,γ)表示：
- S: 状态集合
- A: 动作集合
- P: 状态转移概率 P(s'|s,a)
- R: 奖励函数 R(s,a,s')
- γ: 折扣因子(0≤γ≤1)
3.2 价值迭代算法
价值迭代通过不断更新状态价值函数来寻找最优策略。其核心思想是贝尔曼最优方程：
V*(s) = max_a Σ P(s'|s,a)[R(s,a,s') + γV*(s')]
算法步骤：
- 初始化所有状态价值V(s)
- 对每个状态s，更新V(s) = max_a Σ P(s'|s,a)[R(s,a,s') + γV(s')]
- 重复步骤2直到价值函数收敛
- 从最优价值函数导出最优策略：π*(s) = argmax_a Σ P(s'|s,a)[R(s,a,s') + γV*(s')]
python复制def value_iteration(env, gamma=0.9, theta=1e-6):
"""价值迭代算法实现"""
V = np.zeros(env.size)
policy = np.zeros(env.size, dtype=int)
while True:
delta = 0
for s in env.states:
if env.is_terminal(s):
continue
# 计算每个动作的Q值
q_values = []
for a in env.actions:
q = 0
for s_next in env.states:
q += env.transition_prob(s, a, s_next) * \
(env.reward(s, a, s_next) + gamma * V[s_next])
q_values.append(q)
# 更新价值函数
new_v = max(q_values)
delta = max(delta, abs(new_v - V[s]))
V[s] = new_v
if delta < theta:
break
# 从最优价值函数导出策略
for s in env.states:
if env.is_terminal(s):
continue
q_values = []
for a in env.actions:
q = 0
for s_next in env.states:
q += env.transition_prob(s, a, s_next) * \
(env.reward(s, a, s_next) + gamma * V[s_next])
q_values.append(q)
policy[s] = np.argmax(q_values)
return V, policy
3.3 策略迭代算法
策略迭代交替进行策略评估和策略改进：
- 策略评估：固定当前策略π，计算其价值函数V^π
- 策略改进：基于V^π，贪心地更新策略π' = argmax_a Σ P(s'|s,a)[R(s,a,s') + γV^π(s')]
- 重复上述步骤直到策略收敛
python复制def policy_iteration(env, gamma=0.9, theta=1e-6):
"""策略迭代算法实现"""
# 初始化随机策略
policy = np.random.choice(env.actions, size=env.size)
V = np.zeros(env.size)
while True:
# 策略评估
while True:
delta = 0
for s in env.states:
if env.is_terminal(s):
continue
v = V[s]
a = policy[s]
new_v = 0
for s_next in env.states:
new_v += env.transition_prob(s, a, s_next) * \
(env.reward(s, a, s_next) + gamma * V[s_next])
V[s] = new_v
delta = max(delta, abs(v - V[s]))
if delta < theta:
break
# 策略改进
policy_stable = True
for s in env.states:
if env.is_terminal(s):
continue
old_a = policy[s]
q_values = []
for a in env.actions:
q = 0
for s_next in env.states:
q += env.transition_prob(s, a, s_next) * \
(env.reward(s, a, s_next) + gamma * V[s_next])
q_values.append(q)
new_a = np.argmax(q_values)
policy[s] = new_a
if old_a != new_a:
policy_stable = False
if policy_stable:
break
return V, policy
3.4 两种方法的比较
| 方法 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 价值迭代 | 收敛快，不需要完整策略评估 | 每次迭代计算量大 | 状态空间中等，需要快速收敛 |
| 策略迭代 | 策略改进明确，通常迭代次数少 | 每次迭代需要完整策略评估 | 状态空间较小，策略变化明显 |
在实际应用中，价值迭代通常更常用，特别是当状态空间较大时。策略迭代在策略变化明显时可能收敛更快，但每次迭代的计算成本更高。
4. 无模型增强学习方法
4.1 时间差分学习（TD Learning）
时间差分学习结合了蒙特卡洛方法和动态规划的思想，能够从经验中在线学习，不需要完整的环境模型。最基本的TD(0)算法更新规则为：
V(S_t) ← V(S_t) + α[R_{t+1} + γV(S_{t+1}) - V(S_t)]
其中α是学习率，γ是折扣因子，δ = R_{t+1} + γV(S_{t+1}) - V(S_t)称为TD误差。
4.1.1 SARSA算法
SARSA是一种on-policy的TD控制算法，其名称来源于状态(S)-动作(A)-奖励(R)-下一个状态(S)-下一个动作(A)的序列。更新规则为：
Q(S_t,A_t) ← Q(S_t,A_t) + α[R_{t+1} + γQ(S_{t+1},A_{t+1}) - Q(S_t,A_t)]
python复制def sarsa(env, episodes=1000, alpha=0.1, gamma=0.9, epsilon=0.1):
"""SARSA算法实现"""
Q = np.zeros((env.n_states, env.n_actions))
for _ in range(episodes):
state = env.reset()
action = epsilon_greedy(Q[state], epsilon)
while True:
next_state, reward, done = env.step(action)
next_action = epsilon_greedy(Q[next_state], epsilon)
# SARSA更新
Q[state, action] += alpha * (reward + gamma * Q[next_state, next_action] - Q[state, action])
state, action = next_state, next_action
if done:
break
# 导出确定性策略
policy = np.argmax(Q, axis=1)
return Q, policy
4.1.2 Q-Learning算法
Q-Learning是一种off-policy的TD控制算法，其更新规则为：
Q(S_t,A_t) ← Q(S_t,A_t) + α[R_{t+1} + γmax_a Q(S_{t+1},a) - Q(S_t,A_t)]
与SARSA不同，Q-Learning直接使用最大Q值进行更新，而不考虑实际采取的策略。
python复制def q_learning(env, episodes=1000, alpha=0.1, gamma=0.9, epsilon=0.1):
"""Q-Learning算法实现"""
Q = np.zeros((env.n_states, env.n_actions))
for _ in range(episodes):
state = env.reset()
while True:
action = epsilon_greedy(Q[state], epsilon)
next_state, reward, done = env.step(action)
# Q-Learning更新
best_next_action = np.argmax(Q[next_state])
Q[state, action] += alpha * (reward + gamma * Q[next_state, best_next_action] - Q[state, action])
state = next_state
if done:
break
policy = np.argmax(Q, axis=1)
return Q, policy
4.2 资格迹与TD(λ)
资格迹是一种将TD误差向后传播的机制，可以加速学习。λ是资格迹衰减参数(0≤λ≤1)，λ=0退化为TD(0)，λ=1接近蒙特卡洛方法。
4.2.1 SARSA(λ)算法
python复制def sarsa_lambda(env, episodes=1000, alpha=0.1, gamma=0.9, epsilon=0.1, lambda_=0.7):
"""SARSA(λ)算法实现"""
Q = np.zeros((env.n_states, env.n_actions))
for _ in range(episodes):
E = np.zeros_like(Q) # 资格迹
state = env.reset()
action = epsilon_greedy(Q[state], epsilon)
while True:
next_state, reward, done = env.step(action)
next_action = epsilon_greedy(Q[next_state], epsilon)
# 计算TD误差
td_error = reward + gamma * Q[next_state, next_action] - Q[state, action]
# 更新资格迹
E[state, action] += 1
# 更新Q值和资格迹
Q += alpha * td_error * E
E *= gamma * lambda_
state, action = next_state, next_action
if done:
break
policy = np.argmax(Q, axis=1)
return Q, policy
4.3 深度Q网络（DQN）
对于大规模状态空间，可以使用神经网络来近似Q函数，这就是深度Q网络（Deep Q-Network, DQN）。DQN引入了几个关键技巧：
- 经验回放（Experience Replay）：存储转移样本(s,a,r,s')在回放缓冲区，训练时随机采样
- 目标网络（Target Network）：使用独立的网络生成目标Q值，提高稳定性
python复制class DQN:
def __init__(self, state_dim, action_dim, hidden_dim=64):
self.model = self._build_network(state_dim, action_dim, hidden_dim)
self.target_model = self._build_network(state_dim, action_dim, hidden_dim)
self.update_target()
self.memory = deque(maxlen=10000) # 经验回放缓冲区
self.batch_size = 64
self.gamma = 0.95
self.epsilon = 1.0
self.epsilon_min = 0.01
self.epsilon_decay = 0.995
def _build_network(self, state_dim, action_dim, hidden_dim):
model = Sequential([
Dense(hidden_dim, input_dim=state_dim, activation='relu'),
Dense(hidden_dim, activation='relu'),
Dense(action_dim, activation='linear')
])
model.compile(optimizer=Adam(), loss='mse')
return model
def update_target(self):
self.target_model.set_weights(self.model.get_weights())
def remember(self, state, action, reward, next_state, done):
self.memory.append((state, action, reward, next_state, done))
def act(self, state):
if np.random.rand() <= self.epsilon:
return np.random.randint(self.action_dim)
q_values = self.model.predict(state[np.newaxis])[0]
return np.argmax(q_values)
def replay(self):
if len(self.memory) < self.batch_size:
return
minibatch = random.sample(self.memory, self.batch_size)
states = np.array([x[0] for x in minibatch])
actions = np.array([x[1] for x in minibatch])
rewards = np.array([x[2] for x in minibatch])
next_states = np.array([x[3] for x in minibatch])
dones = np.array([x[4] for x in minibatch])
# 计算目标Q值
targets = self.model.predict(states)
q_next = self.target_model.predict(next_states)
targets[range(self.batch_size), actions] = rewards + self.gamma * np.max(q_next, axis=1) * (1 - dones)
# 训练模型
self.model.train_on_batch(states, targets)
# 衰减探索率
if self.epsilon > self.epsilon_min:
self.epsilon *= self.epsilon_decay
def train(self, env, episodes=1000):
for e in range(episodes):
state = env.reset()
total_reward = 0
done = False
while not done:
action = self.act(state)
next_state, reward, done, _ = env.step(action)
self.remember(state, action, reward, next_state, done)
state = next_state
total_reward += reward
self.replay()
if e % 10 == 0:
self.update_target()
print(f"Episode: {e+1}, Total Reward: {total_reward}, Epsilon: {self.epsilon:.2f}")
5. 策略梯度方法
5.1 策略梯度定理
策略梯度方法直接参数化策略π(a|s;θ)，并通过梯度上升来优化策略性能。策略梯度定理给出了目标函数J(θ)（如平均奖励）关于参数θ的梯度：
∇J(θ) ∝ Σ_s μ(s) Σ_a ∇π(a|s;θ) Q^π(s,a)
其中μ(s)是状态分布，Q^π(s,a)是状态-动作价值函数。
5.2 REINFORCE算法
REINFORCE是最基本的策略梯度算法，使用蒙特卡洛方法估计回报：
- 使用当前策略π_θ生成一个轨迹τ=(s_0,a_0,r_1,...,s_T)
- 对每个时间步t，计算回报G_t = Σ_{k=t}^T γ^{k-t} r_k
- 更新参数θ ← θ + αγ^t G_t ∇lnπ(a_t|s_t;θ)
python复制class REINFORCE:
def __init__(self, state_dim, action_dim, hidden_dim=64):
self.model = self._build_network(state_dim, action_dim, hidden_dim)
self.gamma = 0.99
self.optimizer = Adam()
def _build_network(self, state_dim, action_dim, hidden_dim):
model = Sequential([
Dense(hidden_dim, input_dim=state_dim, activation='relu'),
Dense(action_dim, activation='softmax')
])
return model
def act(self, state):
probs = self.model.predict(state[np.newaxis])[0]
action = np.random.choice(len(probs), p=probs)
return action
def train(self, env, episodes=1000):
for e in range(episodes):
state = env.reset()
states, actions, rewards = [], [], []
done = False
total_reward = 0
# 生成轨迹
while not done:
action = self.act(state)
next_state, reward, done, _ = env.step(action)
states.append(state)
actions.append(action)
rewards.append(reward)
state = next_state
total_reward += reward
# 计算每个时间步的回报
returns = []
G = 0
for r in reversed(rewards):
G = r + self.gamma * G
returns.insert(0, G)
# 归一化回报
returns = np.array(returns)
returns = (returns - np.mean(returns)) / (np.std(returns) + 1e-9)
# 计算梯度并更新
with tf.GradientTape() as tape:
loss = 0
for s, a, G in zip(states, actions, returns):
prob = self.model(s[np.newaxis])[0][a]
loss += -tf.math.log(prob) * G
grads = tape.gradient(loss, self.model.trainable_variables)
self.optimizer.apply_gradients(zip(grads, self.model.trainable_variables))
print(f"Episode: {e+1}, Total Reward: {total_reward}")
5.3 Actor-Critic方法
Actor-Critic方法结合了策略梯度（Actor）和价值函数近似（Critic）的优点。Actor负责选择动作，Critic负责评估动作的好坏。
python复制class ActorCritic:
def __init__(self, state_dim, action_dim, hidden_dim=64):
# Actor网络（策略）
self.actor = self._build_actor(state_dim, action_dim, hidden_dim)
# Critic网络（价值函数）
self.critic = self._build_critic(state_dim, hidden_dim)
self.gamma = 0.99
self.actor_optimizer = Adam()
self.critic_optimizer = Adam()
def _build_actor(self, state_dim, action_dim, hidden_dim):
model = Sequential([
Dense(hidden_dim, input_dim=state_dim, activation='relu'),
Dense(action_dim, activation='softmax')
])
return model
def _build_critic(self, state_dim, hidden_dim):
model = Sequential([
Dense(hidden_dim, input_dim=state_dim, activation='relu'),
Dense(1, activation='linear')
])
return model
def act(self, state):
probs = self.actor.predict(state[np.newaxis])[0]
action = np.random.choice(len(probs), p=probs)
return action
def train(self, env, episodes=1000):
for e in range(episodes):
state = env.reset()
done = False
total_reward = 0
while not done:
with tf.GradientTape(persistent=True) as tape:
# 选择动作
action_probs = self.actor(state[np.newaxis])[0]
action = np.random.choice(len(action_probs), p=action_probs)
# 执行动作
next_state, reward, done, _ = env.step(action)
total_reward += reward
# Critic评估状态价值
value = self.critic(state[np.newaxis])[0]
next_value = self.critic(next_state[np.newaxis])[0] if not done else 0
# 计算TD误差
td_error = reward + self.gamma * next_value - value
# Actor损失（策略梯度）
actor_loss = -tf.math.log(action_probs[action]) * td_error
# Critic损失（均方误差）
critic_loss = tf.square(td_error)
# 更新Actor
actor_grads = tape.gradient(actor_loss, self.actor.trainable_variables)
self.actor_optimizer.apply_gradients(zip(actor_grads, self.actor.trainable_variables))
# 更新Critic
critic_grads = tape.gradient(critic_loss, self.critic.trainable_variables)
self.critic_optimizer.apply_gradients(zip(critic_grads, self.critic.trainable_variables))
state = next_state
print(f"Episode: {e+1}, Total Reward: {total_reward}")
6. 高级增强学习算法
6.1 近端策略优化（PPO）
PPO是一种流行的策略梯度算法，通过限制策略更新的幅度来提高训练稳定性。其目标函数为：
L(θ) = E[min(r(θ)A, clip(r(θ),1-ε,1+ε)A)]
其中r(θ)=π_θ(a|s)/π_old(a|s)是新旧策略的概率比，A是优势函数估计。
python复制class PPO:
def __init__(self, state_dim, action_dim, hidden_dim=64, clip_ratio=0.2):
self.actor = self._build_actor(state_dim, action_dim, hidden_dim)
self.critic = self._build_critic(state_dim, hidden_dim)
self.clip_ratio = clip_ratio
self.gamma = 0.99
self.lam = 0.95
self.actor_optimizer = Adam(3e-4)
self.critic_optimizer = Adam(1e-3)
def _build_actor(self, state_dim, action_dim, hidden_dim):
inputs = Input(shape=(state_dim,))
x = Dense(hidden_dim, activation='tanh')(inputs)
x = Dense(hidden_dim, activation='tanh')(x)
outputs = Dense(action_dim, activation='softmax')(x)
return Model(inputs, outputs)
def _build_critic(self, state_dim, hidden_dim):
inputs = Input(shape=(state_dim,))
x = Dense(hidden_dim, activation='tanh')(inputs)
x = Dense(hidden_dim, activation='tanh')(x)
outputs = Dense(1)(x)
return Model(inputs, outputs)
def act(self, state):
probs = self.actor.predict(state[np.newaxis])[0]
action = np.random.choice(len(probs), p=probs)
return action, probs
def compute_advantages(self, rewards, values, dones, next_value):
"""计算广义优势估计(GAE)"""
advantages = np.zeros_like(rewards)
last_advantage = 0
for t in reversed(range(len(rewards))):
if t == len(rewards) - 1:
next_non_terminal = 1.0 - dones[t]
next_value = next_value
else:
next_non_terminal = 1.0 - dones[t]
next_value = values[t+1]
delta = rewards[t] + self.gamma * next_value * next_non_terminal - values[t]
advantages[t] = delta + self.gamma * self.lam * next_non_terminal * last_advantage
last_advantage = advantages[t]
returns = advantages + values
return advantages, returns
def train(self, env, episodes=1000, steps_per_episode=1000, epochs=10, batch_size=64):
for e in range(episodes):
states, actions, old_probs, rewards, dones, values = [], [], [], [], [], []
state = env.reset()
total_reward = 0
# 收集数据
for _ in range(steps_per_episode):
action, probs = self.act(state)
next_state, reward, done, _ = env.step(action)
value = self.critic.predict(state[np.newaxis])[0]
states.append(state)
actions.append(action)
old_probs.append(probs[action])
rewards.append(reward)
dones.append(done)
values.append(value)
state = next_state
total_reward += reward
if done:
state = env.reset()
# 计算最后一个状态的价值
next_value = self.critic.predict(state[np.newaxis])[0] if not done else 0
# 计算优势和回报
advantages, returns = self.compute_advantages(rewards, values, dones, next_value)
# 转换为numpy数组
states = np.array(states)
actions = np.array(actions)
old_probs = np.array(old_probs)
returns = np.array(returns)
advantages = (advantages - np.mean(advantages)) / (np.std(advantages) + 1e-8)
# 训练多个epoch
for _ in range(epochs):
# 随机打乱数据
indices = np.arange(len(states))
np.random.shuffle(indices)
# 小批量更新
for start in range(0, len(states), batch_size):
end = start + batch_size
idx = indices[start:end]
batch_states = states[idx]
batch_actions = actions[idx]
batch_old_probs = old_probs[idx]
batch_advantages = advantages[idx]
batch_returns = returns[idx]
# 计算策略梯度损失
with tf.GradientTape() as tape:
# 新策略的概率
new_probs = self.actor(batch_states)
new_probs = tf.gather(new_probs, batch_actions, batch_dims=1)
# 概率比
ratio = new_probs / batch_old_probs
# 裁剪目标
surr1 = ratio * batch_advantages
surr2 = tf.clip_by_value(ratio, 1-self.clip_ratio, 1+self.clip_ratio) * batch_advantages
actor_loss = -tf.reduce_mean(tf.minimum(surr1, surr2))
# 熵奖励（鼓励探索）
entropy = -tf.reduce_mean(tf.reduce_sum(new_probs * tf.math.log(new_probs + 1e-10), axis=1))
actor_loss -= 0.01 * entropy
actor_grads = tape.gradient(actor_loss, self.actor.trainable_variables)
self.actor_optimizer.apply_gradients(zip(actor_grads, self.actor.trainable_variables))
# 计算价值函数损失
with tf.GradientTape() as tape:
value_pred = self.critic(batch_states)
critic_loss = tf.reduce_mean(tf.square(batch_returns - value_pred))
critic_grads = tape.gradient(critic_loss, self.critic.trainable_variables)
self.critic_optimizer.apply_gradients(zip(critic_grads, self.critic.trainable_variables))
print(f"Episode: {e+1}, Total Reward: {total_reward}")
6.2 深度确定性策略梯度（DDPG）
DDPG是一种用于连续动作空间的Actor-Critic算法，结合了DQN和策略梯度的思想。
python复制class DDPG:
def __init__(self, state_dim, action_dim, action_high, hidden_dim=64):
self.actor = self._build_actor(state_dim, action_dim, hidden_dim)
self.actor_target = self._build_actor(state_dim, action_dim, hidden_dim)
self.actor_optimizer = Adam(1e-4)
self.critic = self._build_critic(state_dim, action_dim, hidden_dim)
self.critic_target = self._build_critic(state_dim, action_dim, hidden_dim)
self.critic_optimizer = Adam(1e-3)
self.action_high = action_high
self.gamma = 0.99
self.tau = 0.001 # 目标网络软更新参数
self.memory = deque(maxlen=100000)
self.batch_size = 64
self.noise = OUNoise(action_dim)
self.update_target_networks(tau=1.0) # 初始硬更新
def _build_actor(self, state_dim, action_dim, hidden_dim):
inputs = Input(shape=(state_dim,))
x = Dense(hidden_dim, activation='relu')(inputs)
x = Dense(hidden_dim, activation='relu')(x)
outputs = Dense(action_dim, activation='tanh')(x)
model = Model(inputs, outputs)
return model
def _build_critic(self, state_dim, action_dim, hidden_dim):
state_input = Input(shape=(state_dim,))
action_input = Input(shape=(action_dim,))
state_h = Dense(hidden_dim, activation='relu')(state_input)
action_h = Dense(hidden_dim, activation='relu')(action_input)
concat = Concatenate()([state_h, action_h])
x = Dense(hidden_dim, activation='relu')(concat)
outputs = Dense(1)(x)
model = Model([state_input, action_input], outputs)
return model
def update_target_networks(self, tau=None):
tau = tau or self.tau
# 更新Actor目标网络
actor_weights = self.actor.get_weights()
actor_target_weights = self.actor_target.get_weights()
for i in range(len(actor_weights)):
actor_target_weights[i] = tau * actor_weights[i] + (1 - tau) * actor_target_weights[i]
self.actor_target.set_weights(actor_target_weights)
# 更新Critic目标网络
critic_weights = self.critic.get_weights()
critic_target_weights = self.critic_target.get_weights()
for i in range(len(critic_weights)):
critic_target_weights[i] = tau * critic_weights[i] + (1 - tau) * critic_target_weights[i]
self.critic_target.set_weights(critic_target_weights)
def act(self, state, add_noise=True):
action = self.actor.predict(state[np.newaxis])[0]
if add_noise:
action += self.noise.sample()
return np.clip(action, -self.action_high, self.action_high)
def remember(self, state, action, reward, next_state, done):
self.memory.append((state, action, reward, next_state, done))
def replay(self):
if len(self.memory) < self.batch_size:
return
minibatch = random.sample(self.memory, self.batch_size)