拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

DDPG信号灯控制Python源码:从训练到推理的完整工程实践

DDPG信号灯控制Python源码:从训练到推理的完整工程实践

简介:这份资源面向交通工程、智能控制方向的学生与研究者,提供一套基于DDPG深度强化学习算法解决交通信号灯控制问题的完整Python项目,适合作为毕业设计或算法实践参考。压缩包共1984个文件,约145.48MB,包含49个Python脚本、378个pth模型权重、578个sumocfg与524个xml仿真配置、290个txt日志及json数据文件,覆盖智能体定义、经验回放、训练与测试主流程、SUMO路网配置等模块。项目结合PyTorch与SUMO交通仿真工具,通过Traci和Sumolib接口实时采集交通流数据,训练智能体根据车流动态调整信号灯时序,以减少等待时间、提升通行效率,并附带性能评估脚本。目前已有210人学习下载,读者可获取可运行的训练测试代码、预训练模型与仿真环境配置,便于复现实验、调参优化与二次开发。

1. 信号灯配时靠 DDPG 智能体自己学:这套 Python 源码能跑出什么

路口排队长度忽长忽短,固定配时方案在早晚高峰和午间平峰之间来回打脸,这是做交通控制的人最熟悉的场景。这份资源给的是一个用深度确定性策略梯度(DDPG)算法训练信号灯控制智能体的 Python 工程,包含可运行的源码和训练好的模型文件。它要解决的核心问题是:让智能体根据实时车流状态,连续地输出各相位的绿灯时长或相位切换决策,而不是依赖人工标定的周期和绿信比。适合两类人:一类是学过强化学习基础、想找一个完整交通场景练手的算法工程师;另一类是做交通仿真或智慧路口项目、需要一套可复现基线代码的从业者。源码结构通常包含环境封装、DDPG 网络定义、经验回放、训练主循环和模型保存加载几个模块,拿到手就能在本地跑通训练和推理两条链路。

2. DDPG 控制信号灯的原理与工程选型:为什么不是 DQN 或 PPO

2.1 信号灯控制为什么落在连续动作空间

信号灯控制有两种建模方式。一种是离散动作,比如把动作定义为「切换相位」「保持当前相位」,或者把绿灯时长切成 5 秒、10 秒、15 秒几档,这种建模适合 DQN 这类输出 Q 值的算法。另一种是连续动作,直接输出某个相位的绿灯持续秒数,或者输出相位时长的调整量,取值范围是一个连续区间。真实路口的绿灯时长本质上是连续的,离散化会引入量化误差,档位太粗控制不精细,档位太细动作空间爆炸、Q 值估计方差变大。DDPG 属于 actor-critic 架构,actor 网络直接输出连续动作,critic 网络评估动作价值,天然适配「绿灯时长连续可调」这个需求。常见做法是把动作定义为当前相位绿灯时长的增量,再用一个动作边界裁剪到合理区间,避免智能体输出负时长或超长时长。

2.2 DDPG 四个网络与目标网络软更新

DDPG 的工程实现里有四个网络:actor 在线网络、actor 目标网络、critic 在线网络、critic 目标网络。actor 负责根据状态输出动作,critic 负责评估「状态-动作」对的价值。目标网络的作用是给 critic 计算目标 Q 值时提供一个相对稳定的参考,避免自举导致训练发散。目标网络的参数不是直接复制在线网络,而是用软更新(soft update)的方式缓慢跟随:

# 软更新:tau 通常取 0.001 到 0.005 for target_param, param in zip(self.actor_target.parameters(), self.actor.parameters()): target_param.data.copy_(tau * param.data + (1.0 - tau) * target_param.data) for target_param, param in zip(self.critic_target.parameters(), self.critic.parameters()): target_param.data.copy_(tau * param.data + (1.0 - tau) * target_param.data)

tau 越小,目标网络越稳定但跟随越慢;tau 越大,跟随快但训练容易震荡。交通信号灯场景里状态变化相对平缓,tau 取 0.001 到 0.002 比较常见。critic 的损失用均方贝尔曼误差,actor 的损失用 critic 对 actor 输出动作的评价值取负,也就是让 actor 朝着 critic 认为价值高的方向更新。

2.3 经验回放与探索噪声的配合

DDPG 是 off-policy 算法,训练数据存在经验回放缓冲区里,每次从缓冲区随机采样一个 batch 更新网络。回放缓冲区大小、batch size、采样策略都会影响训练稳定性。探索方面,DDPG 在 actor 输出动作上叠加奥恩斯坦-乌伦贝克(OU)噪声或高斯噪声,让智能体在训练初期多尝试不同动作。噪声方差通常随训练步数衰减,前期探索、后期利用。如果噪声不衰减,智能体收敛后会一直抖动,路口配时忽长忽短;如果噪声衰减太快,智能体可能过早陷入局部策略,某些相位组合永远试不到。

2.4 状态设计:排队长度、等待时间与相位 one-hot

状态向量一般包含各车道排队车辆数、车辆平均等待时间、当前相位 one-hot 编码、当前相位已持续时长。排队长度和等待时间反映路口拥堵程度,相位编码让智能体知道当前处于哪个相位,已持续时长帮助智能体判断是否该切换。状态归一化很关键,排队长度除以车道容量、等待时间除以一个参考上限,把各维度压到相近量级,否则网络训练时梯度会被大量纲维度主导。常见做法是每个仿真步返回一个固定长度的状态向量,维度等于车道数乘以统计指标数再加上相位编码维度。

3. 把源码跑起来:环境封装、训练循环与模型加载

3.1 环境接口对齐 Gym 风格

这份源码的环境封装通常遵循 Gym 风格的 reset / step 接口,方便替换成 SUMO、CityFlow 或其他交通仿真器。reset 返回初始状态,step 接收动作返回下一状态、奖励、done 标志和额外信息。如果你手头有自己的仿真环境,只要把接口对齐,DDPG 训练部分基本不用改。下面是一个典型的环境骨架:

class TrafficEnv: def __init__(self, config): self.lane_num = config["lane_num"] # 车道数 self.max_green = config["max_green"] # 单相位最大绿灯秒数 self.min_green = config["min_green"] # 单相位最小绿灯秒数 self.state_dim = self.lane_num * 2 + 4 # 排队+等待+相位编码+已持续时长 def reset(self): # 重置仿真,返回初始状态向量 self.current_phase = 0 self.phase_duration = 0 return self._get_state() def step(self, action): # action 是连续值,裁剪到 [min_green, max_green] green_time = np.clip(action[0], self.min_green, self.max_green) self._advance_simulation(green_time) next_state = self._get_state() reward = self._compute_reward() done = self._check_done() return next_state, reward, done, {}

_compute_reward一般用负的排队长度或负的累计等待时间,也可以把通行量作为正奖励加进去。奖励设计直接决定智能体学出什么行为:只罚排队,智能体可能频繁切换相位;只奖通行量,智能体可能让某些方向长期绿灯。常见做法是奖励等于负的加权排队长度加上一个切换惩罚项,抑制过于频繁的相位切换。

3.2 训练主循环与超参数落点

训练主循环负责采样、存回放、更新网络、软更新目标网络、保存模型。下面这段是核心结构:

for episode in range(max_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.replay_buffer.push(state, action, reward, next_state, done) if len(agent.replay_buffer) > batch_size: agent.update() # 采样 batch 更新四个网络 state = next_state episode_reward += reward if done: break # 噪声衰减 agent.noise_scale = max(agent.noise_scale * 0.999, 0.05) if episode % save_interval == 0: agent.save("ddpg_traffic.pth")

max_episodes和max_steps决定训练总步数,交通场景里一个 episode 通常对应一段仿真时长,比如一小时或一个高峰时段。batch_size常见取 64 到 256,回放缓冲区容量取 1e5 到 1e6。update()里先算 critic 目标值,再更新 critic,然后更新 actor,最后软更新目标网络。顺序不能乱,否则目标值用的是更新后的网络,自举会不稳定。

3.3 模型加载与推理模式

训练完保存的模型文件包含 actor 和 critic 的 state_dict。推理时只需要 actor 网络,不需要探索噪声:

agent = DDPGAgent(state_dim=env.state_dim, action_dim=1) agent.load("ddpg_traffic.pth") agent.actor.eval() # 切到推理模式 state = env.reset() for step in range(eval_steps): with torch.no_grad(): action = agent.actor(torch.FloatTensor(state)).numpy() green_time = np.clip(action[0], env.min_green, env.max_green) state, reward, done, _ = env.step([green_time]) if done: break

torch.no_grad()关掉梯度计算,减少显存占用和计算量。eval()把网络里的 dropout 和 batch norm 切到推理行为。如果你的模型保存时用了 DataParallel 或分布式训练,加载时注意 key 里可能带module.前缀,需要做一次 key 映射。

3.4 训练曲线看什么指标

训练过程中至少盯三个量:episode 累计奖励、平均排队长度、平均等待时间。累计奖励上升说明策略在优化目标上变好,但奖励设计有偏时可能和真实指标背离,所以排队长度和等待时间要单独记录。常见现象是训练前期奖励快速上升,中期震荡,后期缓慢收敛。如果奖励一直不涨,先检查状态归一化和奖励尺度;如果奖励涨但排队没降,检查奖励函数是不是被切换惩罚项主导了。

4. 避坑与排查:DDPG 训信号灯最容易翻车的几个点

4.1 现象:训练几百轮奖励不涨,动作输出恒定

原因通常是 actor 输出被初始化或激活函数压死了。DDPG 的 actor 最后一层一般用 tanh 把输出压到 [-1, 1],再按动作范围缩放。如果最后一层权重初始化太小,输出接近 0,缩放后动作始终在中间值附近,智能体等于没学。解决方法是检查 actor 最后一层权重初始化,常见做法是用均匀分布初始化到较小范围,同时确认动作缩放公式正确:action = (action + 1) / 2 * (max_action - min_action) + min_action。

4.2 现象:critic 损失爆炸或变成 NaN

原因多半是学习率太大、奖励尺度太大或者状态没归一化。critic 的 Q 值会随着奖励累积不断增大,如果奖励是负的排队长度且数值上百,Q 值量级很快上去,梯度爆炸。解决办法是把奖励缩放到 [-1, 1] 或 [-10, 10] 量级,critic 学习率取 1e-3 到 1e-4,actor 学习率比 critic 再小一个量级。另外可以在 critic 更新时做梯度裁剪,torch.nn.utils.clip_grad_norm_限制梯度范数。

4.3 现象:智能体学会一直保持同一相位不切换

这是奖励设计的经典坑。如果奖励只罚排队长度,而某个方向车流少,智能体发现保持该方向绿灯、其他方向排队涨得慢,总惩罚反而低,于是永远不切换。解决办法是在奖励里加入切换惩罚或者对长时间不切换的相位加惩罚项,也可以把各方向等待时间的最大值纳入奖励,逼智能体关注最堵的方向。常见做法是奖励等于负的(各车道排队长度之和 + 最大等待时间),再加一个相位切换的固定小惩罚。

4.4 现象:训练时奖励正常,推理时表现差很多

原因通常是训练和推理的状态分布不一致。训练时环境有随机性,推理时环境确定,如果状态归一化用了训练集的统计量而推理时没同步,输入分布就偏了。另一个常见原因是推理时忘了把 actor 切到 eval 模式,或者忘了关探索噪声。排查方法是打印训练和推理时同一场景下的状态向量,对比数值范围;确认推理代码里没有调用select_action而是直接走 actor 前向。

4.5 现象:换一个路口或车流文件后模型完全失效

DDPG 学到的策略和状态维度、车道数、车流分布强相关。换路口后如果状态维度变了,actor 网络输入对不上,直接报错;如果维度一样但车流分布变了,策略可能次优甚至很差。解决办法是把状态设计成与路口拓扑无关的统计量,或者在新路口上做少量微调训练。常见做法是保留 actor 和 critic 权重,用新路口的仿真数据以较小学习率继续训练几百轮,让策略适配新分布。

5. 进阶技巧:用固定随机种子和评估脚本验证策略稳定性

训练完一个 DDPG 信号灯智能体,最怕的是「这次跑出来好,下次跑出来崩」。强化学习对随机种子敏感是出了名的玄学,环境初始化、网络初始化、经验回放采样都带随机性。我一般会在训练脚本开头固定三个种子:

import random, numpy as np, torch def set_seed(seed=42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False

cudnn.deterministic = True会让 cuDNN 选确定性算法,速度可能慢一点,但换来的可复现性对调参很重要。benchmark = False关掉自动算法选择,避免同一份代码在不同机器上跑出不同结果。固定种子后,同一份配置跑三次,如果三次评估指标方差很大,说明策略本身不稳定,不是种子问题,得回去看网络结构或奖励设计。

评估脚本要单独写,不要复用训练循环。评估时跑多个 episode,每个 episode 用不同随机种子初始化车流,统计平均排队长度、平均等待时间、平均通行量三个指标,输出均值和标准差。下面是一个评估骨架:

def evaluate(agent, env, episodes=10): metrics = {"queue": [], "wait": [], "throughput": []} for ep in range(episodes): state = env.reset(seed=1000 + ep) # 每个 episode 不同车流 done = False while not done: with torch.no_grad(): action = agent.actor(torch.FloatTensor(state)).numpy() state, _, done, info = env.step(action) metrics["queue"].append(info["avg_queue"]) metrics["wait"].append(info["avg_wait"]) metrics["throughput"].append(info["throughput"]) for k, v in metrics.items(): print(f"{k}: mean={np.mean(v):.2f}, std={np.std(v):.2f}") return metrics

标准差比均值更能说明问题。如果某个指标标准差很大,说明策略对车流波动敏感,实际部署时可能时好时坏。我一般要求三个指标的标准差都在均值的 15% 以内,超过就继续调。另外评估时要把探索噪声彻底关掉,actor 切 eval,否则评估结果里混进了随机动作,指标会虚高或虚低。

还有一个容易忽略的点:模型保存不要只存最后一个 episode 的权重。训练后期策略可能在震荡,最后一个不一定最好。常见做法是每隔若干 episode 评估一次,保存评估指标最好的那个 checkpoint,同时保留最近几个 checkpoint 做对比。文件名带上 episode 号和评估指标,比如ddpg_traffic_ep1200_queue8.3.pth,回头排查时能直接定位到是哪一轮的模型。

从那以后我每次跑强化学习训练,都强制先固定种子、再跑评估脚本、最后才看训练曲线,顺序反了就容易把随机波动当成算法改进。希望帮到你。

本文还有配套的精品资源,点击获取

返回列表