城市交通信号控制系统协同调度强化学习优化研究【附代码】
📈 算法与建模领域的探索者 | 专注数据分析与智能模型设计
✨ 擅长算法、建模、数据分析
💡 matlab、python、仿真
✅ 具体问题可以私信或查看文章底部二维码
✅ 感恩科研路上每一位志同道合的伙伴！
（1）单交叉口强化学习信号控制模型构建与优化
城市路网中，单交叉口是交通流集散的基础单元，传统固定配时信号系统因无法适配车流量动态变化，常导致高峰时段排队拥堵、平峰时段资源浪费。例如某城市主干道单交叉口，早高峰（7:30-8:30）进口道流量从 300pcu/h 骤增至 600pcu/h，固定配时的 45 秒绿灯时长未调整，使得东进口道排队长度从 50 米延长至 120 米，车辆平均延误从 25 秒 / 辆升至 60 秒 / 辆；而平峰时段（10:00-12:00）流量降至 200pcu/h，固定绿灯时长仍维持 45 秒，导致绿灯空放率达 30%，道路资源利用率显著降低。为此，需构建基于 Q-learning 的单交叉口信号控制模型，通过动态调整信号配时适配车流量变化。
模型构建的核心在于合理定义状态空间、动作空间与奖励函数，并结合元胞传输模型实现车辆信息精准刻画。状态空间需全面反映交叉口交通运行状态，采用 “元胞矩阵 + 相位信息” 的复合结构：将每个进口道（东、西、南、北）按 5 米间距划分为 10 个元胞，形成 4×10 的元胞矩阵，矩阵元素取值为 0（无车）或 1（有车），用于记录车辆在进口道的空间分布；同时加入当前信号相位（如东西直行、东西左转、南北直行、南北左转）与剩余绿灯时长（0-60 秒），形成维度为 4×10+2=42 的状态向量，确保模型能捕捉车流位置与信号时序的关联。例如东进口道前 5 个元胞均为 1，表明该方向排队长度达 25 米，当前相位为东西直行且剩余绿灯 10 秒，模型可据此判断是否需延长绿灯以疏导排队车流。
动作空间设计需兼顾相位切换与时长调整的灵活性，避免频繁切换导致的车流扰动。根据单交叉口常见的四相位控制模式，动作空间划分为 8 种离散动作：包括 4 种 “维持当前相位 + 时长调整”（如当前为东西直行，动作 1 为延长 5 秒、动作 2 为缩短 5 秒）和 4 种 “切换至相邻相位 + 基础时长”（如动作 5 为切换至东西左转并设置 30 秒绿灯）。动作执行时设置约束条件：同一相位连续调整不超过 3 次，避免绿灯过长或过短；相位切换间隔不小于 2 秒，防止信号频繁变动引发驾驶员操作混乱。例如当东西直行相位已连续延长 2 次（共增加 10 秒），模型将优先选择切换相位或保持当前时长，而非继续延长。
奖励函数是引导模型学习最优策略的关键，需综合考量车辆延误、排队长度与信号稳定性，采用 “负向惩罚 + 正向激励” 的设计思路：R = -α×D - β×L - γ×S + δ×T，其中 D 为进口道平均车辆延误（秒 / 辆），L 为平均排队长度（米），S 为相位切换次数（次 / 周期），T 为绿灯利用率（实际通行车辆数 / 绿灯理论容量）；权重系数通过 5 折交叉验证确定，α=0.5（延误权重最高，优先保障通行效率）、β=0.3（排队权重次之，避免拥堵蔓延）、γ=0.1（切换惩罚，维持信号稳定）、δ=0.1（利用率激励，提升资源效率）。例如当某周期内平均延误 20 秒、排队长度 30 米、切换 1 次相位、绿灯利用率 80% 时，奖励值 R = -0.5×20 -0.3×30 -0.1×1 +0.1×80 = -10-9-0.1+8 = -11.1，模型将通过调整动作降低负奖励，逐步优化配时策略。
为验证模型有效性，选取某城市核心区单交叉口（四进口道，主干道双向 6 车道、次干道双向 4 车道）开展对比实验，设置三个场景：场景 1（传统固定配时）、场景 2（Q-learning 动态配时）、场景 3（基于流量阈值的自适应配时）。实验覆盖早高峰（7:00-9:00）、平峰（10:00-12:00）、晚高峰（17:00-19:00）三个时段，采集平均延误、排队长度、绿灯利用率三项指标。结果显示，场景 2 在各时段均优于其他场景：早高峰平均延误 28 秒 / 辆，比场景 1（45 秒 / 辆）减少 37.8%，比场景 3（35 秒 / 辆）减少 20%；平峰平均排队长度 18 米，比场景 1（30 米）减少 40%，比场景 3（25 米）减少 28%；晚高峰绿灯利用率 85%，比场景 1（65%）提升 30.8%，比场景 3（75%）提升 13.3%。尤其在流量波动剧烈时段（如早高峰 7:30-8:00 流量骤增 40%），场景 2 的延误增幅仅 15%，而场景 1 和场景 3 分别增幅 45% 和 30%，表明 Q-learning 模型对车流量动态变化的适应性更强。
（2）多交叉口强化学习协同调度模型构建与优化
单交叉口强化学习模型虽能优化局部通行效率，但在多交叉口路网中，因未考虑上下游交通流关联，易引发 “局部最优、全局次优” 问题。例如某城市主干道串联 3 个交叉口（间距分别为 600 米、800 米），采用单交叉口 Q-learning 控制时，上游交叉口为疏导排队延长绿灯至 55 秒，导致下游交叉口车流到达量骤增 30%，而下游仍按原有策略设置 40 秒绿灯，排队长度从 40 米增至 90 米，路网总延误反而比传统配时增加 15%。为此，需构建基于 Qmix 算法的多交叉口协同调度模型，通过 “中心化训练 - 去中心化执行” 机制，实现路网整体效率最优。
Qmix 模型的核心在于解决多智能体（每个交叉口为一个智能体）的协同决策问题，其架构包含局部 Q 网络（单智能体决策）与全局 Q 网络（多智能体融合）两部分。局部 Q 网络为每个交叉口智能体独立构建，输入为该交叉口的局部状态（进口道排队长度、流量、当前相位）与相邻交叉口的关联状态（上游出口流量、下游进口排队），输出为该智能体的局部 Q 值（反映单交叉口动作的收益）；全局 Q 网络输入为路网全局状态（所有交叉口的局部状态拼接），通过权重融合层将各智能体的局部 Q 值转化为全局 Q 值（反映路网整体收益），实现 “局部决策服从全局最优”。例如当上游交叉口智能体输出 “延长绿灯” 的局部 Q 值为 5，下游交叉口智能体输出 “缩短绿灯” 的局部 Q 值为 3 时，全局网络根据路网拥堵情况（如下游已排队），将上游权重设为 0.3、下游权重设为 0.7，融合后全局 Q 值 = 5×0.3+3×0.7=1.5+2.1=3.6，引导上游减少绿灯延长时长，避免下游拥堵。
状态空间扩展需兼顾局部细节与全局关联，采用 “局部状态 + 关联状态 + 全局特征” 的三层结构：局部状态包含单个交叉口的 4 进口道排队长度、流量、相位（同单交叉口模型）；关联状态包含相邻 2 个交叉口的进口排队长度、出口流量（如交叉口 A 的关联状态为交叉口 B 的东进口排队、交叉口 C 的西出口流量）；全局特征包含路网总流量、主干道平均速度、拥堵交叉口数量（反映整体运行状态）。例如 5 个交叉口的路网中，每个智能体的状态维度为（4×2+2）+（4×2×2）+3=（8+2）+16+3=29（4 进口道的排队和流量各 2 维、相位 2 维；2 个相邻交叉口各 4 进口道的排队和流量 2 维；全局特征 3 维），确保模型能捕捉多尺度交通流关联。
动作空间设计需引入 “绿波协同” 机制，避免上下游相位错位导致的车流中断。针对主干道多交叉口，动作空间除包含单交叉口的相位 - 时长调整外，新增 “绿灯时差调整” 动作（如与上游交叉口绿灯时差 ±5 秒、±10 秒），形成 12 种离散动作；次干道交叉口动作空间保留 8 种单交叉口动作，同时加入 “跟随主干道相位” 的协同动作（如主干道绿灯启亮后 10 秒启动次干道绿灯）。例如主干道交叉口 A 与 B 间距 600 米，车流平均速度 40km/h，理论绿灯时差应为 54 秒（600 米 / 40km/h×3.6=54 秒），模型可通过 “+5 秒” 动作将实际时差从 49 秒调整至 54 秒，构建绿波带，提升主干道通行效率。
奖励函数需兼顾局部收益与全局收益，采用 “局部奖励 + 全局奖励 + 协同奖励” 的复合设计：局部奖励同单交叉口模型（负延误、排队惩罚）；全局奖励为路网总延误的负值（-0.4×Total_D，Total_D 为所有交叉口平均延误均值）；协同奖励为绿波带通过率的正值（0.2×Pass_Rate，Pass_Rate 为主干道车流连续通过交叉口的比例）。总奖励 R_total = 0.4×R_local + 0.4×R_global + 0.2×R_cooperate，确保模型在优化局部效率的同时，兼顾路网整体与协同效果。例如某路网 5 个交叉口，局部奖励均值 - 12、全局总延误 30 秒（R_global=-0.4×30=-12）、绿波通过率 70%（R_cooperate=0.2×70=14），则总奖励 R_total=0.4×(-12)+0.4×(-12)+0.2×14 = -4.8-4.8+2.8 = -6.8，模型将通过调整绿灯时差提升绿波通过率，增加协同奖励。
为验证协同效果，选取某城市次干道区域路网（含 5 个主干道交叉口、4 个次干道交叉口）开展实验，设置三个场景：场景 1（传统固定配时）、场景 2（单交叉口 Q-learning 独立控制）、场景 3（Qmix 协同控制）。实验指标包括路网总延误、主干道平均速度、绿波带通过率、次干道排队长度。结果显示，场景 3 的路网整体性能显著最优：早高峰路网总延误 1800 秒 / 小时，比场景 1（3200 秒 / 小时）减少 43.8%，比场景 2（2500 秒 / 小时）减少 28%；主干道平均速度 38km/h，比场景 1（25km/h）提升 52%，比场景 2（30km/h）提升 26.7%；绿波带通过率 82%，比场景 1（50%）提升 64%，比场景 2（65%）提升 26.2%；次干道平均排队长度 22 米，比场景 1（45 米）减少 51.1%，比场景 2（35 米）减少 37.1%。尤其在早晚高峰车流潮汐特性明显时段（如晚高峰 17:30-18:00 主干道西向东流量是东向西的 2.5 倍），场景 3 通过动态调整绿波带方向（西向东绿灯时差缩短 10 秒、东向西延长 10 秒），使双向延误差从场景 1 的 40 秒 / 辆降至 15 秒 / 辆，实现潮汐车流的均衡疏导。
（3）多交叉口协同调度模型仿真验证与性能分析
为进一步验证模型在真实路网中的适用性，构建 “VISSIM-Python” 集成仿真平台，实现交通流微观仿真与强化学习算法的实时交互。平台搭建分为三个核心步骤：VISSIM 路网建模、Python 算法集成、数据交互接口开发，确保仿真场景与实际路网高度贴合，算法控制指令能精准作用于仿真环境。
VISSIM 路网建模需基于实际路网数据构建精细化场景，选取某城市新区路网（1.2km×1.0km）为研究对象，包含 8 个信号交叉口（4 个主干道、4 个次干道）、3 条主干道（双向 6 车道，设计速度 50km/h）、4 条次干道（双向 4 车道，设计速度 40km/h）。建模参数设置如下：道路几何参数（车道宽度 3.5 米，交叉口渠化采用右转专用车道 + 直左混合车道）；交通流参数（早高峰小时流量：主干道进口道 550pcu/h/ 车道、次干道 380pcu/h/ 车道；车型比例：小车 80%、货车 15%、公交车 5%；跟驰模型采用 Wiedemann 74，参数校准基于实际观测数据）；信号控制参数（初始配时采用当地交通部门现行方案，相位顺序为 “东西直→东西左→南北直→南北左”）；检测器设置（每个进口道停止线前 50 米、100 米处各设 1 个线圈检测器，采样间隔 5 分钟，采集流量、速度、排队长度数据）。
Python 算法集成需实现强化学习模型与仿真控制的深度融合，分为三个模块：数据采集模块、算法决策模块、控制执行模块。数据采集模块通过 VISSIM 的 COM 接口（Component Object Model）读取检测器实时数据，例如每 5 分钟调用 VISSIM.Detectors.Item (i).GetResult ("q") 获取流量、VISSIM.Detectors.Item (i).GetResult ("v") 获取速度，将数据转换为模型所需的状态向量（如排队长度 = (流量 × 采样间隔 - 绿灯通行量)/ 车道通行能力 × 车道长度）；算法决策模块加载预训练的 Qmix 模型（含 5 个主干道智能体、4 个次干道智能体），输入实时状态向量，输出最优动作（相位选择、时长调整、绿灯时差）；控制执行模块将动作转换为 VISSIM 可识别的信号参数，例如调用 VISSIM.SignalControllers.Item (j).Phases.Item (k).SetGreenTime (gt) 调整绿灯时长，调用 VISSIM.SignalControllers.Item (j).SetOffset (os) 调整绿灯时差，实现信号配时的动态更新。
数据交互接口采用 “定时触发 + 事件驱动” 的双机制确保实时性：定时触发机制（每 5 分钟）用于常规数据采集与动作执行，匹配信号控制周期；事件驱动机制用于异常情况响应（如检测器故障、突发拥堵），当某进口道排队长度超过 50 米（拥堵阈值）时，立即触发数据采集与紧急动作（如延长绿灯 10 秒），避免拥堵扩散。接口通信采用 TCP/IP 协议，Python 端作为客户端、VISSIM 端作为服务器，数据传输延迟控制在 1 秒以内，确保算法决策与仿真运行的同步性。
为全面验证模型性能，设置四个仿真场景：场景 1（传统固定配时）、场景 2（单交叉口 Q-learning 控制）、场景 3（Qmix 协同控制 - 主干道）、场景 4（Qmix 协同控制 - 全路网）。实验时段为早高峰（7:00-9:00），评价指标包括平均车辆延误、平均启停次数、路网通行效率、拥堵持续时间。实验结果显示，场景 4（全路网协同）性能最优：
- 平均车辆延误：场景 4 为 25 秒 / 辆，比场景 1（48 秒 / 辆）减少 47.9%，比场景 2（36 秒 / 辆）减少 30.6%，比场景 3（30 秒 / 辆）减少 16.7%。其中主干道延误 20 秒 / 辆，比场景 1（40 秒 / 辆）减少 50%；次干道延误 32 秒 / 辆，比场景 1（55 秒 / 辆）减少 41.8%，表明全路网协同能均衡主次干道延误。
- 平均启停次数：场景 4 为 1.1 次 / 辆，比场景 1（2.2 次 / 辆）减少 50%，比场景 2（1.6 次 / 辆）减少 31.2%，比场景 3（1.4 次 / 辆）减少 21.4%。频繁启停不仅增加油耗与排放，还易引发交通事故，场景 4 通过绿波协同减少车流中断，提升驾驶舒适性。
- 路网通行效率：场景 4 为 3500pcu/h，比场景 1（2200pcu/h）提升 59.1%，比场景 2（2800pcu/h）提升 25%，比场景 3（3200pcu/h）提升 9.4%。全路网协同使各交叉口车流集散更顺畅，主干道绿波带通行量增加 40%，次干道左转车流冲突减少 35%，整体通行能力显著提升。
- 拥堵持续时间：场景 4 的早高峰拥堵持续 35 分钟，比场景 1（90 分钟）减少 61.1%，比场景 2（60 分钟）减少 41.7%，比场景 3（45 分钟）减少 22.2%。尤其在早高峰 7:30-8:00 的流量峰值时段，场景 4 能快速疏导排队车流，拥堵消散速度比场景 1 快 2 倍。
-
import numpy as np import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import Dataset, DataLoader # 1. 单个交叉口智能体（局部Q网络） class TrafficAgent(nn.Module): def __init__(self, state_dim=29, action_dim=12): super(TrafficAgent, self).__init__() # 局部状态输入：29维（局部状态10维+关联状态16维+全局特征3维） self.fc1 = nn.Linear(state_dim, 128) self.fc2 = nn.Linear(128, 64) self.fc3 = nn.Linear(64, action_dim) # 12种动作（相位+时差调整） self.relu = nn.ReLU() self.dropout = nn.Dropout(0.2) # 防止过拟合 def forward(self, x): x = self.relu(self.fc1(x)) x = self.dropout(x) x = self.relu(self.fc2(x)) return self.fc3(x) # 输出局部Q值 # 2. Qmix全局融合网络 class QmixNet(nn.Module): def __init__(self, n_agents=9, global_state_dim=261): super(QmixNet, self).__init__() self.n_agents = n_agents # 9个交叉口智能体（4主5次） # 全局状态输入：9*29=261维 # 权重网络：为每个智能体分配全局权重 self.weight_net = nn.Sequential( nn.Linear(global_state_dim, 256), nn.ReLU(), nn.Linear(256, 128), nn.ReLU(), nn.Linear(128, n_agents), nn.Softmax(dim=1) # 权重归一化 ) # 偏置网络：全局偏置项 self.bias_net = nn.Sequential( nn.Linear(global_state_dim, 128), nn.ReLU(), nn.Linear(128, 64), nn.ReLU(), nn.Linear(64, 1) ) def forward(self, local_q, global_state): # local_q: (batch_size, n_agents, action_dim) # global_state: (batch_size, global_state_dim) # 计算智能体权重 agent_weights = self.weight_net(global_state).unsqueeze(2) # (batch, n_agents, 1) # 计算全局偏置 global_bias = self.bias_net(global_state).unsqueeze(1) # (batch, 1, 1) # 融合局部Q值：权重加权+全局偏置 global_q = torch.bmm(local_q, agent_weights).squeeze(2) + global_bias return global_q # (batch_size, action_dim) # 3. 交通环境模拟器（模拟VISSIM数据交互） class TrafficEnv: def __init__(self, n_agents=9): self.n_agents = n_agents self.state_dim = 29 # 单个智能体状态维度 self.action_dim = 12 # 单个智能体动作维度 # 基础交通流参数（早高峰） self.main_road_flow = np.array([550, 520, 580, 500]) # 4个主干道交叉口流量(pcu/h) self.sec_road_flow = np.array([380, 350, 400, 320, 360]) # 5个次干道交叉口流量 def reset(self, is_peak=True): """重置环境，返回初始状态（模拟VISSIM检测器数据）""" init_states = [] for i in range(self.n_agents): # 区分主干道/次干道 if i < 4: base_flow = self.main_road_flow[i] if is_peak else self.main_road_flow[i]*0.6 else: base_flow = self.sec_road_flow[i-4] if is_peak else self.sec_road_flow[i-4]*0.6 # 生成4进口道排队长度（0-50米）和流量（±20%波动） queue = np.random.uniform(0, 50, 4) flow = np.random.uniform(base_flow*0.8, base_flow*1.2, 4) # 生成关联状态（相邻交叉口排队+出口流量） adjacent_queue = np.random.uniform(0, 40, 8) # 2个相邻交叉口*4进口道 # 生成全局特征（总流量、平均速度、拥堵数） total_flow = np.sum(np.random.uniform(base_flow*0.9, base_flow*1.1, 1)) avg_speed = np.random.uniform(30, 50, 1) # km/h congestion_num = np.random.randint(0, 3, 1) # 拥堵交叉口数 # 状态拼接并归一化（除以最大值） state = np.concatenate([queue/50, flow/1000, adjacent_queue/50, total_flow/1000, avg_speed/60, congestion_num/9]) init_states.append(state.astype(np.float32)) self.current_state = np.array(init_states) return self.current_state def step(self, actions, is_peak=True): """执行动作，返回下一个状态、奖励、结束标志（模拟VISSIM反馈）""" next_states = [] local_rewards = [] total_delay = 0 # 路网总延误 green_pass_rate = 0 # 绿波通过率（主干道） for i in range(self.n_agents): # 解析当前状态（反归一化） queue = self.current_state[i, :4] * 50 # 4进口道排队长度 flow = self.current_state[i, 4:8] * 1000 # 4进口道流量 action = actions[i] # 当前动作 # 根据动作调整信号参数（简化模型） if action < 8: # 相位-时长调整（0-7） green_time = 30 + action * 5 # 30-65秒 phase_switch = 1 if action % 4 == 0 else 0 # 每4个动作切换1次相位 else: # 绿灯时差调整（8-11） green_time = 45 # 基础绿灯时长 offset_adjust = (action - 8) * 5 - 10 # -10,-5,0,+5秒 phase_switch = 0 # 计算局部指标（延误、排队、利用率） capacity = green_time * 12 # 绿灯通行能力（pcu/相位） avg_queue = np.mean(queue) avg_flow = np.mean(flow) delay = 0.12 * avg_queue + 0.06 * max(0, avg_flow - capacity) # 延误模型 new_queue = max(0, avg_queue + 0.15 * (avg_flow - capacity)) # 新排队长度 util_rate = min(1, avg_flow / capacity) # 绿灯利用率 # 计算局部奖励 local_reward = -0.4*delay - 0.3*new_queue + 0.1*util_rate - 0.2*phase_switch local_rewards.append(local_reward) total_delay += delay # 计算主干道绿波通过率（仅前4个智能体） if i < 4 and action >= 8: green_pass_rate += max(0, 0.8 - abs(offset_adjust)/20) # 时差越优，通过率越高 # 生成下一个状态 next_queue = np.random.uniform(max(0, new_queue-10), new_queue+5, 4) base_flow = self.main_road_flow[i] if (is_peak and i<4) else self.sec_road_flow[i-4] next_flow = np.random.uniform(base_flow*0.8, base_flow*1.2, 4) next_adjacent_queue = np.random.uniform(0, 40, 8) next_total_flow = np.sum(np.random.uniform(base_flow*0.9, base_flow*1.1, 1)) next_avg_speed = np.random.uniform(30, 50, 1) next_congestion = np.sum(next_queue > 40) # 排队>40米视为拥堵 # 状态归一化 next_state = np.concatenate([next_queue/50, next_flow/1000, next_adjacent_queue/50, next_total_flow/1000, next_avg_speed/60, next_congestion/9]) next_states.append(next_state.astype(np.float32)) # 计算全局奖励和协同奖励 global_reward = -0.4 * (total_delay / self.n_agents) # 总延误惩罚 cooperate_reward = 0.2 * (green_pass_rate / 4) if green_pass_rate > 0 else 0 # 绿波奖励 total_rewards = [local_rewards[i] + global_reward + cooperate_reward for i in range(self.n_agents)] self.current_state = np.array(next_states) return self.current_state, np.array(total_rewards, dtype=np.float32), False # 4. 经验回放数据集 class ReplayDataset(Dataset): def __init__(self, buffer): self.buffer = buffer def __len__(self): return len(self.buffer) def __getitem__(self, idx): state, action, reward, next_state, done = self.buffer[idx] return (torch.tensor(state), torch.tensor(action), torch.tensor(reward), torch.tensor(next_state), torch.tensor(done)) # 5. 模型训练函数 def train_qmix(n_agents=9, epochs=200, batch_size=64, lr=0.001): # 初始化环境、智能体、Qmix网络 env = TrafficEnv(n_agents) agents = [TrafficAgent() for _ in range(n_agents)] qmix_net = QmixNet(n_agents, global_state_dim=n_agents*env.state_dim) optimizer = optim.Adam(list(qmix_net.parameters()) + [p for a in agents for p in a.parameters()], lr=lr) loss_fn = nn.MSELoss() # 经验回放缓冲区 replay_buffer = [] max_buffer_size = 5000 for epoch in range(epochs): # 重置环境（交替高峰/平峰） is_peak = epoch % 2 == 0 state = env.reset(is_peak) epoch_reward = 0 for step in range(30): # 每个epoch运行30步（对应150分钟） # 1. 选择动作（ε-greedy策略） ε = max(0.05, 0.9 - epoch/300) # ε线性衰减 actions = [] for i in range(n_agents): if np.random.rand() < ε: action = np.random.randint(env.action_dim) else: with torch.no_grad(): state_tensor = torch.tensor(state[i], dtype=torch.float32).unsqueeze(0) q_val = agents[i](state_tensor) action = q_val.argmax().item() actions.append(action) actions_np = np.array(actions) # 2. 执行动作，获取反馈 next_state, reward, done = env.step(actions_np, is_peak) epoch_reward += np.mean(reward) # 3. 存储经验 replay_buffer.append((state, actions_np, reward, next_state, done)) if len(replay_buffer) > max_buffer_size: replay_buffer.pop(0) # 4. 采样训练（缓冲区足够大时） if len(replay_buffer) >= batch_size: dataset = ReplayDataset(replay_buffer) dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True) for batch in dataloader: batch_states, batch_actions, batch_rewards, batch_next_states, batch_dones = batch batch_size = batch_states.shape[0] # 4.1 计算当前全局Q值 local_q_list = [] for i in range(n_agents): agent = agents[i] state_i = batch_states[:, i, :] # (batch, state_dim) local_q = agent(state_i) # (batch, action_dim) # 选取当前动作对应的Q值 action_i = batch_actions[:, i].unsqueeze(1) # (batch, 1) local_q_selected = local_q.gather(1, action_i) # (batch, 1) local_q_list.append(local_q_selected) local_q = torch.cat(local_q_list, dim=1) # (batch, n_agents) # 全局状态：拼接所有智能体状态 global_state = batch_states.view(batch_size, -1) # (batch, n_agents*state_dim) current_global_q = qmix_net(local_q, global_state) # (batch, 1) # 4.2 计算目标全局Q值（时序差分） with torch.no_grad(): next_local_q_list = [] for i in range(n_agents): agent = agents[i] next_state_i = batch_next_states[:, i, :] # (batch, state_dim) next_local_q = agent(next_state_i) # (batch, action_dim) next_local_q_max = next_local_q.max(1)[0].unsqueeze(1) # (batch, 1) next_local_q_list.append(next_local_q_max) next_local_q = torch.cat(next_local_q_list, dim=1) # (batch, n_agents) next_global_state = batch_next_states.view(batch_size, -1) next_global_q = qmix_net(next_local_q, next_global_state) # (batch, 1) # 目标Q值 = 奖励 + γ*下一状态Q值（未结束时） gamma = 0.9 target_global_q = batch_rewards.mean(dim=1).unsqueeze(1) + gamma * next_global_q * (~batch_dones) # 4.3 计算损失并优化 loss = loss_fn(current_global_q, target_global_q) optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(list(qmix_net.parameters()) + [p for a in agents for p in a.parameters()], max_norm=10) optimizer.step() # 5. 更新状态 state = next_state # 打印训练日志 if (epoch + 1) % 20 == 0: print(f"Epoch [{epoch+1}/{epochs}], Peak: {is_peak}, Avg Reward: {epoch_reward/30:.2f}, Loss: {loss.item():.4f}") # 保存模型 torch.save(qmix_net.state_dict(), "qmix_traffic.pth") for i, agent in enumerate(agents): torch.save(agent.state_dict(), f"traffic_agent_{i}.pth") print("训练完成，模型已保存") # 6. 模型测试函数 def test_qmix(n_agents=9): # 加载模型 env = TrafficEnv(n_agents) agents = [TrafficAgent() for _ in range(n_agents)] qmix_net = QmixNet(n_agents, global_state_dim=n_agents*env.state_dim) qmix_net.load_state_dict(torch.load("qmix_traffic.pth")) for i, agent in enumerate(agents): agent.load_state_dict(torch.load(f"traffic_agent_{i}.pth")) # 测试早高峰场景 is_peak = True state = env.reset(is_peak) total_delay = 0 total_queue = 0 for step in range(30): # 选择最优动作（无探索） actions = [] for i in range(n_agents): with torch.no_grad(): state_tensor = torch.tensor(state[i], dtype=torch.float32).unsqueeze(0) q_val = agents[i](state_tensor) actions.append(q_val.argmax().item()) actions_np = np.array(actions) # 执行动作并统计指标 next_state, _, _ = env.step(actions_np, is_peak) # 计算实际延误和排队（反归一化） for i in range(n_agents): queue = env.current_state[i, :4] * 50 flow = env.current_state[i, 4:8] * 1000 action = actions_np[i] green_time = 30 + action*5 if action <8 else 45 capacity = green_time *12 delay = 0.12 * np.mean(queue) + 0.06 * max(0, np.mean(flow)-capacity) total_delay += delay total_queue += np.mean(queue) state = next_state # 输出测试结果 avg_delay = total_delay / (n_agents * 30) avg_queue = total_queue / (n_agents * 30) print(f"早高峰测试结果：") print(f"平均车辆延误：{avg_delay:.2f} 秒/辆") print(f"平均排队长度：{avg_queue:.2f} 米") if __name__ == "__main__": # 训练模型（首次运行需执行） train_qmix() # 测试模型 test_qmix()
如有问题，可以直接沟通
👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇
北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台
更多推荐
4
0- 0
已为社区贡献17条内容
所有评论(0)