2. 集中式训练分布式执行基础框架
任务目标
搭建一个最小可运行的CTDE（Centralized Training with Decentralized Execution）框架，包含集中式Critic与分布式Actor，用于多智能体协作任务。训练时Critic利用全局信息（所有智能体的观测与动作）评估联合动作价值；执行时每个Actor仅依赖自身局部观测独立决策。
环境与依赖
- Python 3.8+
- PyTorch 1.12+
- NumPy
实现思路
采用两个智能体协作的简单环境：每个智能体观测自身位置（一维坐标），目标为同时到达同一目标点。奖励为两智能体与目标距离之和的负值，促使协作。Actor网络输入自身观测，输出动作（连续值，表示移动量）；Critic网络输入拼接后的全局状态（两个观测）与两个动作，输出联合Q值。训练时使用集中式Critic计算TD误差更新两个Actor；执行时仅使用各自的Actor网络。
关键完整代码
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
# ---------- 环境定义 ----------
class CooperativeEnv:
"""两个智能体协作到达目标点。观测: 自身位置(1维)。动作: 移动量(连续)。"""
def __init__(self):
self.target = 5.0
self.reset()
def reset(self):
self.pos = np.array([0.0, 0.0])
return self._get_obs()
def step(self, actions):
self.pos += np.array(actions, dtype=np.float32)
dist = np.abs(self.pos - self.target)
reward = -float(np.sum(dist)) # 协作奖励: 距离越小越好
done = bool(np.all(dist < 0.1))
return self._get_obs(), reward, done
def _get_obs(self):
return self.pos.astype(np.float32).reshape(2, 1)
# ---------- 网络定义 ----------
class Actor(nn.Module):
"""分布式Actor: 输入自身观测, 输出动作。"""
def __init__(self, obs_dim=1, act_dim=1):
super().__init__()
self.net = nn.Sequential(nn.Linear(obs_dim, 32), nn.ReLU(),
nn.Linear(32, act_dim), nn.Tanh())
def forward(self, obs):
return self.net(obs) * 2.0 # 动作范围[-2, 2]
class Critic(nn.Module):
"""集中式Critic: 输入全局状态(2个观测)与2个动作, 输出联合Q值。"""
def __init__(self, obs_dim=1, act_dim=1, n_agents=2):
super().__init__()
input_dim = obs_dim * n_agents + act_dim * n_agents
self.net = nn.Sequential(nn.Linear(input_dim, 64), nn.ReLU(),
nn.Linear(64, 64), nn.ReLU(),
nn.Linear(64, 1))
def forward(self, obs_all, act_all):
x = torch.cat([obs_all.flatten(), act_all.flatten()])
return self.net(x)
# ---------- 训练参数 ----------
n_agents = 2
actor_lr, critic_lr = 1e-3, 1e-3
gamma = 0.99
episodes = 300
actors = [Actor() for _ in range(n_agents)]
critic = Critic()
actor_optims = [optim.Adam(a.parameters(), lr=actor_lr) for a in actors]
critic_optim = optim.Adam(critic.parameters(), lr=critic_lr)
loss_fn = nn.MSELoss()
env = CooperativeEnv()
# ---------- 训练循环 ----------
for ep in range(episodes):
obs = env.reset()
obs_t = torch.tensor(obs, dtype=torch.float32)
total_reward = 0.0
done = False
while not done:
# 分布式执行: 每个Actor仅用自身观测
acts = []
for i in range(n_agents):
with torch.no_grad():
a = actors[i](obs_t[i].unsqueeze(0))
acts.append(a)
act_t = torch.cat(acts) # shape [2,1]
# 环境交互
obs_next, reward, done = env.step(act_t.detach().numpy().flatten())
obs_next_t = torch.tensor(obs_next, dtype=torch.float32)
total_reward += reward
# 集中式训练: Critic使用全局信息
q = critic(obs_t, act_t)
with torch.no_grad():
if done:
target = torch.tensor([reward], dtype=torch.float32)
else:
# 计算下一时刻动作(目标策略)
next_acts = []
for i in range(n_agents):
na = actors[i](obs_next_t[i].unsqueeze(0))
next_acts.append(na)
next_act_t = torch.cat(next_acts)
q_next = critic(obs_next_t, next_act_t)
target = torch.tensor([reward], dtype=torch.float32) + gamma * q_next
# 更新Critic
critic_loss = loss_fn(q, target)
critic_optim.zero_grad()
critic_loss.backward()
critic_optim.step()
# 更新Actor (最大化Q值)
for i in range(n_agents):
actor_loss = -critic(obs_t, act_t) # 梯度上升
actor_optims[i].zero_grad()
actor_loss.backward()
actor_optims[i].step()
obs_t = obs_next_t
if (ep+1) % 50 == 0:
print(f"Episode {ep+1}, Total Reward: {total_reward:.2f}")
# ---------- 验证: 分布式执行 ----------
print("\n验证分布式执行(仅用局部观测):")
obs = env.reset()
done = False
steps = 0
while not done and steps < 50:
acts = []
for i in range(n_agents):
with torch.no_grad():
a = actors[i](torch.tensor(obs[i], dtype=torch.float32).unsqueeze(0))
acts.append(a.item())
obs, _, done = env.step(acts)
steps += 1
print(f"最终位置: {obs.flatten()}, 目标位置: {env.target}, 到达: {done}")运行方式
python ctde_basic.py验证方法与判定标准
- 训练收敛指标：训练过程中每50个episode打印的Total Reward应呈上升趋势，最终稳定在接近0的负值（如-0.5~0），表示两个智能体已学会协作到达目标点。
- 执行验证：训练结束后，代码自动运行一次分布式执行测试，若输出“到达: True”且最终位置接近[5.0, 5.0]，则证明训练成功且Actor可独立决策。
- 判定通过条件：满足上述两点即可判定框架搭建与训练成功。若训练不收敛，可调整学习率或增加episode数至500。
常见问题
- 训练不收敛：检查Critic的target计算是否使用了detach()（已处理），或增大gamma至0.99以上。
- 动作过大导致震荡：Tanh输出乘以2限制了动作范围，若仍震荡可减小至1.0。
- 扩展至更多智能体：只需修改n_agents并确保环境返回对应维度的观测即可，网络结构无需改动。