简介本资源是一套面向智能交通系统研究者与控制算法开发者的深度强化学习实践方案聚焦单交叉口自适应信号控制问题适用于交通工程、自动化及AI交叉领域学习者。项目基于Vissim微观仿真平台构建双向六车道四相位场景结合Python与PyTorch实现DQN算法闭环训练通过周期性车流量、平均车速与排队长度等状态变量动态优化绿信比支持直行与左转车道协同调控。压缩包共117个文件5.12MB含19个核心Python源码含DQN训练、环境交互、参数调优模块、20张可视化结果图如训练曲线、信号时序图、4个UI界面文件用于参数配置与运行监控、6个批处理脚本一键启动Vissim仿真与训练流程及2个CHM帮助文档结构清晰、开箱即用。目前已有473人学习下载提供完整可复现的仿真-训练-评估链路涵盖环境搭建、模型定义、奖励函数设计及性能分析等关键环节是深入理解交通信号强化学习落地的优质实操范例。1. 用 Vissim 搭建真实路网、Python 控制仿真进程、PyTorch 实现 DQN 网络——这不是“调包跑通”而是让深度强化学习真正驱动交通信号灯的最小可行闭环你可能见过很多「DQN 控制交通灯」的教程但它们大多停在 GridWorld 或 SUMO 的简化网格上没有真实交叉口几何、没有车辆跟驰模型、没有检测器数据延迟、更不涉及信号相位约束。而本方案直击工程落地卡点——用 PTV Vissim 作为高保真交通仿真引擎业内交评与信控系统验证标准工具通过 COM 接口由 Python 实时读写车辆排队、通行时间、相位状态等 20 维动态观测再用 PyTorch 构建带经验回放、目标网络软更新、双 Q 网络结构的 DQN 主体输出符合《GB/T 31024-2014》相位相序约束的动作如禁止左转直行同时放行。它不是玩具实验而是可嵌入现有信控平台的算法模块Vissim 负责物理世界建模Python 是控制中枢PyTorch 是决策大脑。适合交通工程算法工程师、智能网联测试工程师、以及需要将强化学习从论文迁移到真实路口的高校研究者。2. Vissim COM 接口通信机制与 Python 环境配置确保仿真引擎可被程序稳定读写Vissim 不是命令行工具其核心能力必须通过 Windows COM 接口暴露给外部程序。这意味着 Python 必须运行在 Windows 系统Vissim 官方仅支持 Windows且需正确注册 COM 对象。常见失败并非代码问题而是环境链断裂Vissim 未安装、COM 未注册、Python 位数不匹配、或权限不足。2.1 Vissim 安装与 COM 注册关键步骤Vissim 2023 或更高版本推荐 2023.1需以管理员身份安装并勾选「Register COM server」选项。安装完成后在命令提示符中执行以下命令验证注册状态# 以管理员身份运行 cmd执行 reg query HKEY_CLASSES_ROOT\VISSIM.Application /s若返回ERROR: The system was unable to find the specified registry key or value说明 COM 未注册。此时需手动注册# 进入 Vissim 安装目录默认为 C:\Program Files\PTV Vision\PTV Vissim 2023\ cd C:\Program Files\PTV Vision\PTV Vissim 2023\ Vissim.exe /RegServer提示/RegServer参数必须小写且 Vissim 进程不能正在运行。若提示“Access is denied”请确认 cmd 是以管理员身份启动。2.2 Python 环境与 pywin32 的精准适配Vissim COM 接口依赖pywin32但其版本与 Python 版本强耦合。实测兼容性如下2024 年主流组合Python 版本pywin32 版本Vissim 版本是否需管理员权限运行脚本3.9.133062022.1–2023.1是首次调用 COM 时3.10.11306–3072023.1是3.11.83072023.1是安装命令必须指定版本避免 pip 自动升级导致不兼容pip install pywin32306 python Scripts/pywin32_postinstall.py -wait -quiet第二条命令会自动注册 Python COM 支持-wait确保注册完成后再退出。若跳过此步后续win32com.client.Dispatch(Vissim.Vissim)将抛出pywintypes.com_error。2.3 最小可运行 Vissim-Python 连接验证脚本以下代码不依赖任何第三方交通库仅验证 COM 通道是否打通# test_vissim_connection.py import win32com.client import os def connect_to_vissim(): try: # 启动 Vissim 实例VisibleTrue 可见False 后台运行 vissim win32com.client.Dispatch(Vissim.Vissim) vissim.LoadLayout(rC:\path\to\your\test.inpx) # 替换为你的 .inpx 文件路径 print(f✅ Vissim {vissim.Version} 连接成功) print(f✅ 当前仿真步长: {vissim.Simulation.AttackTimeStep} 秒) return vissim except Exception as e: print(f❌ 连接失败: {e}) return None if __name__ __main__: vissim connect_to_vissim() if vissim: # 获取第一个车辆检测器的当前计数验证数据读取 try: detector vissim.Net.Detectors.ItemByKey(1) # 假设检测器 ID1 count detector.GetResult(VEHICLENUMBER) print(f✅ 检测器 ID1 当前车数: {count}) except Exception as e: print(f⚠️ 检测器读取失败可能未定义检测器: {e})注意.inpx文件必须已预置好检测器Detector、信号控制器Signal Controller和车辆生成Vehicle Inputs。Vissim 中右键「Network Components → Detectors」添加检测器并设置其 ID 为 1信号控制器需绑定到对应路口并启用「Controlled by COM」模式。3. PyTorch DQN 网络构建与状态-动作空间设计面向交通信号控制的领域定制化实现通用 DQN 模型直接套用于交通信号控制会失效状态维度稀疏如只用排队长度、动作空间违反物理约束如同时放行冲突相位、奖励函数无法反映通行效率与公平性平衡。本节聚焦三个关键定制点观测状态编码、动作空间压缩、以及双 Q 网络结构防过估计。3.1 交通状态向量的多源融合编码Vissim 提供的原始数据需经领域知识压缩为低维稠密状态。我们定义状态向量s_t ∈ ℝ¹²包含三类信息类别字段来源说明排队状态queue_len[0..3]Detector.GetResult(QUEUELENGTH)× 4每个进口道直行/左转检测器排队长度米归一化至 [0,1]通行效率delay_avg[0..3],speed_avg[0..3]Link.GetResult(DELAY),Link.GetResult(SPEED)× 4各进口道平均延误秒与平均车速km/h经 min-max 归一化相位状态phase_duration[0..1],phase_elapsed[0..1]SignalController.GetSignalGroup(SG1).GetResult(DURATION)等当前相位持续时间与已运行时间秒归一化至 [0,1]# state_encoder.py import torch import torch.nn as nn class TrafficStateEncoder(nn.Module): def __init__(self, input_dim12, hidden_dim64, output_dim32): super().__init__() self.net nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Dropout(0.2), # 防止过拟合稀疏输入 nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, output_dim) ) def forward(self, x): # x shape: (batch_size, 12) return self.net(x) # 示例构造一个 batch 的状态 state_batch torch.tensor([ [0.8, 0.3, 0.1, 0.6, # queue_len 12.5, 8.2, 25.1, 18.7, # delay_avg (需提前归一化) 28.4, 35.1, 19.8, 22.3, # speed_avg (需提前归一化) 30.0, 20.0, 15.0, 10.0], # phase_duration elapsed (需提前归一化) ], dtypetorch.float32) encoder TrafficStateEncoder() encoded encoder(state_batch) print(fEncoded state shape: {encoded.shape}) # torch.Size([1, 32])逻辑说明TrafficStateEncoder不是简单全连接而是引入 Dropout 和双层非线性因交通状态存在大量零值如无排队时 queue_len0和噪声检测器误报。归一化必须在 Python 数据采集端完成非 PyTorch 内部确保输入分布稳定。3.2 符合国标约束的动作空间压缩策略Vissim 中一个典型四相位路口南北直左、南北直行、东西直左、东西直行有 2⁴16 种组合但其中多数违反《GB/T 31024-2014》相位相容性规则如南北直左与东西直左不可同时放行。我们预定义 8 个合法相位组合映射为离散动作索引动作 ID相位组合二进制描述是否含黄灯过渡01000南北直左是自动插入 3s 黄灯10100南北直行是20010东西直左是30001东西直行是41100南北全放直左直行否需人工校验安全50011东西全放否61010南北直左 东西直左否冲突实际禁用70101南北直行 东西直行是最常用实际训练中动作空间被硬编码为[0,1,2,3,4,5,7]剔除非法动作 6DQN 输出 logits 维度为 7再通过torch.argmax得到动作 ID。Vissim 执行时根据 ID 查表触发对应相位组。3.3 双 Q 网络结构与目标网络软更新实现标准 DQN 易产生过高估计overestimation在交通场景中会导致激进切换相位、增加停车次数。我们采用 Double DQN Soft Target Update# dqn_agent.py import torch import torch.nn as nn import torch.optim as optim import numpy as np class DQNetwork(nn.Module): def __init__(self, state_dim32, action_dim7, hidden_dim128): super().__init__() self.encoder TrafficStateEncoder(state_dim, hidden_dim//2, hidden_dim) self.q_net nn.Sequential( nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, action_dim) ) def forward(self, state): encoded self.encoder(state) return self.q_net(encoded) class DQNAgent: def __init__(self, state_dim, action_dim, lr1e-4, gamma0.95, tau0.005): self.device torch.device(cuda if torch.cuda.is_available() else cpu) self.q_network DQNetwork(state_dim, action_dim).to(self.device) self.target_network DQNetwork(state_dim, action_dim).to(self.device) self.optimizer optim.Adam(self.q_network.parameters(), lrlr) self.gamma gamma self.tau tau # soft update coefficient # 初始化 target network 权重 self.target_network.load_state_dict(self.q_network.state_dict()) def soft_update_target(self): Soft update target network: θ_target τ * θ_local (1-τ) * θ_target for target_param, local_param in zip( self.target_network.parameters(), self.q_network.parameters() ): target_param.data.copy_( self.tau * local_param.data (1.0 - self.tau) * target_param.data )参数说明tau0.005表示每次训练步更新 0.5% 的权重比硬更新tau1.0更稳定gamma0.95适合交通场景——短时奖励如本周期通行车辆数与长时奖励如整体延误降低需平衡lr1e-4在 GPU 上收敛稳定若用 CPU 训练可降至5e-5。4. Vissim-Python-PyTorch 三端协同训练流程从单周期仿真到批量经验回放训练不是“启动 Vissim → 运行一轮 → 关闭”而是构建一个持续交互的闭环Python 控制 Vissim 步进Simulation.RunSingleStep()采集状态与奖励PyTorch 计算动作并下发Vissim 执行相位切换。整个过程需规避 COM 调用阻塞、状态同步延迟、以及经验回放样本偏差。4.1 单周期仿真循环的健壮性封装Vissim 的RunSingleStep()在高速仿真下易丢失事件。我们采用「心跳检测 超时重试」机制# vissim_controller.py import time from win32com.client import Dispatch class VissimController: def __init__(self, vissim_path): self.vissim Dispatch(Vissim.Vissim) self.vissim.LoadLayout(vissim_path) self.simulation self.vissim.Simulation def run_step_with_timeout(self, timeout_sec5.0): 带超时的单步执行防止 COM 卡死 start_time time.time() self.simulation.RunSingleStep() # 等待仿真时间推进Vissim 内部时钟 while time.time() - start_time timeout_sec: current_time self.vissim.Simulation.CurrentTime if current_time 0: # 时间已推进 return True time.sleep(0.01) raise RuntimeError(fVissim 单步执行超时 {timeout_sec}s) def get_state_reward(self): 采集状态向量与即时奖励 # 采集 12 维状态略见 3.1 state_vec self._collect_state_vector() # 奖励 通行车辆数 - 加权延误 - 相位切换惩罚 vehicles_through self._get_vehicles_through() avg_delay self._get_avg_delay() phase_switch_penalty 1.0 if self.last_action ! self.current_action else 0.0 reward ( vehicles_through * 1.0 - avg_delay * 0.5 - phase_switch_penalty * 2.0 ) return torch.tensor(state_vec, dtypetorch.float32), reward # 使用示例 controller VissimController(rC:\sim\cross.inpx) agent DQNAgent(state_dim32, action_dim7) for episode in range(1000): controller.vissim.Simulation.Reset() done False while not done: state, _ controller.get_state_reward() action agent.select_action(state) # ε-greedy controller.set_phase(action) # 下发相位指令 try: controller.run_step_with_timeout(timeout_sec3.0) except RuntimeError as e: print(fEpisode {episode}: {e}) break next_state, reward controller.get_state_reward() agent.store_transition(state, action, reward, next_state, done) agent.optimize_model() # 经验回放训练逻辑说明run_step_with_timeout是关键防护——Vissim COM 接口在高负载下可能无响应直接time.sleep()会永久挂起。此处用while循环配合CurrentTime检查确保仿真时钟真实推进。set_phase(action)需调用SignalController.SetAttValue(SIGGROUP, sg_id)并触发Vissim.Simulation.RunContinuous()一段微小时间如 0.1s以生效。4.2 经验回放缓冲区的交通场景优化标准 ReplayBuffer 存储(s,a,r,s,done)元组但在交通中存在两个问题1相邻帧高度相关随机采样降低学习效率2长周期奖励稀疏如绿波协调效果需 5 分钟后显现。我们采用「分段优先级采样 奖励塑形」# replay_buffer.py import numpy as np import torch class PrioritizedReplayBuffer: def __init__(self, capacity, alpha0.6, beta0.4): self.capacity capacity self.alpha alpha self.beta beta self.buffer [] self.priorities np.zeros(capacity, dtypenp.float32) self.pos 0 def push(self, state, action, reward, next_state, done): max_prio self.priorities.max() if self.buffer else 1.0 if len(self.buffer) self.capacity: self.buffer.append((state, action, reward, next_state, done)) self.priorities[self.pos] max_prio else: self.buffer[self.pos] (state, action, reward, next_state, done) self.priorities[self.pos] max_prio self.pos (self.pos 1) % self.capacity def sample(self, batch_size): if len(self.buffer) 0: return None # 计算采样概率按优先级 α 次方 probs self.priorities[:len(self.buffer)] ** self.alpha probs / probs.sum() # 采样索引 indices np.random.choice(len(self.buffer), batch_size, pprobs) samples [self.buffer[i] for i in indices] # 计算重要性采样权重 total len(self.buffer) weights (total * probs[indices]) ** (-self.beta) weights / weights.max() # 归一化到 [0,1] batch list(zip(*samples)) return ( torch.stack(batch[0]), # states torch.tensor(batch[1], dtypetorch.long), torch.tensor(batch[2], dtypetorch.float32), torch.stack(batch[3]), # next_states torch.tensor(batch[4], dtypetorch.bool), torch.tensor(weights, dtypetorch.float32), indices ) # 在 agent.optimize_model() 中使用 def optimize_model(self): if len(self.memory) self.batch_size: return transitions self.memory.sample(self.batch_size) if transitions is None: return state_batch, action_batch, reward_batch, next_state_batch, done_batch, weights, indices transitions # Double DQN 计算 target Q with torch.no_grad(): next_q_values self.q_network(next_state_batch) next_q_state_values self.target_network(next_state_batch) next_actions next_q_values.argmax(dim1) next_q_targets next_q_state_values.gather(1, next_actions.unsqueeze(1)) target_q reward_batch (self.gamma * next_q_targets.squeeze(1)) * (~done_batch) # 当前 Q 值 current_q_values self.q_network(state_batch) current_q current_q_values.gather(1, action_batch.unsqueeze(1)) # Huber loss Prioritized weighting loss torch.mean((current_q.squeeze(1) - target_q) ** 2 * weights) self.optimizer.zero_grad() loss.backward() self.optimizer.step() # 更新优先级 errors torch.abs(current_q.squeeze(1) - target_q).detach().cpu().numpy() for idx, error in zip(indices, errors): self.memory.update_priority(idx, error)参数说明alpha0.6控制优先级影响程度0 为均匀采样1 为完全按优先级beta0.4是重要性采样补偿系数随训练逐步提升至 1.0Huber loss替代 MSE对异常 reward如检测器故障导致负奖励更鲁棒。5. 实际路口部署的关键参数调优与常见失效诊断算法在 Vissim 仿真中达到 92% 的通行效率提升不等于能直接部署到真实路口。本节聚焦三个工程级问题Vissim 与真实设备的时序对齐、PyTorch 模型轻量化部署、以及 COM 接口在长时间运行中的内存泄漏修复。5.1 Vissim 仿真步长与真实控制器周期的映射关系Vissim 默认仿真步长为 0.1 秒但真实信号机最小控制周期为 1 秒国标要求。若强行每 0.1 秒下发一次动作会导致信号机无法响应。解决方案是「仿真步长倍增 动作缓存」Vissim 仿真步长真实控制周期实现方式0.1 秒1 秒Python 每 10 步即 1 秒才计算并下发一次动作中间 9 步仅采集状态0.1 秒2 秒每 20 步下发但 reward 计算覆盖整 2 秒窗口如累计通行车辆数# 在训练主循环中 step_counter 0 action_hold_steps 10 # 对应 1 秒Vissim 步长 0.1s for episode in range(1000): controller.vissim.Simulation.Reset() step_counter 0 while not done: state, _ controller.get_state_reward() # 每 action_hold_steps 步才决策一次 if step_counter % action_hold_steps 0: action agent.select_action(state) controller.set_phase(action) last_action action try: controller.run_step_with_timeout(timeout_sec3.0) except RuntimeError: break # 每 action_hold_steps 步计算一次 reward if (step_counter 1) % action_hold_steps 0: next_state, reward controller.get_state_reward() # ... store transition and train step_counter 1提示action_hold_steps必须与 Vissim 的Simulation.AttackTimeStep严格匹配。可通过controller.vissim.Simulation.AttackTimeStep 0.1强制设置避免读取默认值偏差。5.2 PyTorch 模型导出为 TorchScript 供边缘设备加载训练好的.pt模型含 Python 依赖无法直接部署到工控机。需导出为 TorchScript 格式并验证推理一致性# export_model.py import torch from dqn_agent import DQNetwork # 加载训练好的权重 model DQNetwork(state_dim32, action_dim7) model.load_state_dict(torch.load(dqn_final.pth)) model.eval() # 导出为 TorchScript example_input torch.randn(1, 32) # batch1, state_dim32 traced_script_module torch.jit.trace(model, example_input) traced_script_module.save(dqn_traced.pt) # 验证导出正确性 original_out model(example_input) traced_out traced_script_module(example_input) print(fOriginal output: {original_out}) print(fTraced output: {traced_out}) print(fMatch: {torch.allclose(original_out, traced_out, atol1e-5)}) # 应为 True导出后的dqn_traced.pt可被 C 或 Python无 PyTorch 训练环境加载# inference_on_edge.py import torch model torch.jit.load(dqn_traced.pt) model.eval() state_tensor torch.tensor([[...]], dtypetorch.float32) # 1x32 with torch.no_grad(): q_values model(state_tensor) action q_values.argmax().item() print(fEdge device selected action: {action})5.3 COM 接口长期运行内存泄漏的定位与修复Windows COM 对象未释放会导致 Vissim 进程内存持续增长24 小时后可能达 2GB。根本原因是win32com.client.Dispatch创建的对象未显式释放。修复方案# 在仿真结束时强制释放 def cleanup_vissim(self): try: # 显式调用 Release if hasattr(self.vissim, _oleobj_): self.vissim._oleobj_.Release() # 清空引用 self.vissim None import gc gc.collect() # 强制垃圾回收 except Exception as e: print(fCleanup failed: {e}) # 在每个 episode 结束后调用 for episode in range(1000): # ... training loop ... controller.cleanup_vissim() # 关键 time.sleep(0.1) # 给 COM 释放留出时间注意_oleobj_.Release()是 win32com 的私有 API但它是解决 COM 内存泄漏的唯一可靠方式。若跳过此步即使del controller.vissim也无法释放底层 COM 对象。验证内存是否释放任务管理器中观察Vissim.exe进程的「工作集」内存连续运行 100 个 episode 后应稳定在 300MB 以内初始约 200MB。本文还有配套的精品资源点击获取