企业官网建设流程全解析

简介：本资源是一套基于SUMO交通仿真平台与DQN强化学习算法的交通信号灯相位时间优化实践方案，面向智能交通、自动驾驶及AI控制方向的本科生、研究生与科研初学者，解决城市交叉口动态信号配时策略设计与验证问题。压缩包共32个文件（533KB），含16个SUMO核心配置文件（.net.xml、.rou.xml等）支撑多时段流量建模，6个OSM地图文件用于路网构建，5个Python脚本（PDQN_main.py、DQN_main.py等）实现DQN智能体训练与环境交互，3个Excel记录原始与优化后TSC数据，另含sumocfg工程配置与README说明文档。已有202人学习下载，提供从真实路网导入、7天全时段与高峰/平峰分流数据（shixin_shanyin_all7.rou.xml、high.rou.xml等）、到DQN网络结构与奖励函数设计的完整闭环代码与配置，可直接复现、调试并迁移至其他路口场景。

1. 为什么用 SUMO + DQN 调交通信号灯相位时间，不是“炫技”，而是解决真实瓶颈

在城市主干道交叉口，早高峰车流峰值常达每小时3000辆，但传统固定配时方案仍按日均流量设计——结果是左转车道空等90秒，直行队列却已溢出路口。这类“结构性拥堵”无法靠增加绿灯总时长缓解，关键在于相位时间的动态分配权：哪个方向该多给5秒？哪个相位该提前2秒切换？什么时候该插入黄闪过渡？这些决策粒度细、响应快、强耦合，恰好是深度强化学习（DQN）最擅长的序列决策场景。SUMO 不是简单动画播放器，它提供毫秒级车辆轨迹仿真、精确的信号控制器接口（traci.trafficlight.setPhaseDuration）、可编程的实时状态观测（如各进口道排队长度、平均速度、等待车辆数），构成闭环训练所需的高保真环境。本文聚焦一个可落地的最小可行路径：不依赖复杂路网建模，从单交叉口起步，用 Python 实现 DQN 智能体与 SUMO 的实时交互，重点拆解状态空间如何定义才不丢失关键拥堵特征、动作空间怎样编码才能让神经网络学出“相位切换逻辑”、以及 reward 函数为何必须包含等待时间惩罚与通行效率奖励的加权平衡——所有代码基于sumo-rl封装库和 PyTorch，适配 SUMO 1.16+ 版本，新手照着步骤能跑通，老手可直接调参优化。

2. 构建 SUMO 环境：从路网配置到实时状态观测的完整链路

2.1 单交叉口路网配置的关键参数与数据结构

SUMO 的核心输入是.net.xml（路网）、.rou.xml（车流）、.add.xml（信号灯配置）。对单交叉口场景，避免使用netconvert自动生成复杂路网，手动编写精简版更可控。关键点在于：

• 进口道必须定义from和to车道，且to车道需明确连接关系（<connection from="E2" to="N1" .../>）；
• 信号灯组（<tlLogic>）中每个<phase>的duration初始值设为 0，由 DQN 动态控制；
• 必须启用--no-step-log和--no-warnings参数启动 SUMO，否则 traci 连接会因日志输出阻塞；
• 车辆生成采用<flow>而非<vehicle>，通过vehsPerHour控制流量强度，便于后续做流量扰动实验。

<!-- example.net.xml 片段 --> <net> <junction id="J1" type="traffic_light" x="0.0" y="0.0"/> <edge id="E2" from="J0" to="J1" priority="1"/> <edge id="W2" from="J1" to="J0" priority="1"/> <edge id="N1" from="J2" to="J1" priority="1"/> <edge id="S1" from="J1" to="J2" priority="1"/> <connection from="E2" to="N1" via=":J1_0" tl="J1" linkIndex="0" dir="l"/> <connection from="E2" to="S1" via=":J1_1" tl="J1" linkIndex="1" dir="s"/> <connection from="W2" to="S1" via=":J1_2" tl="J1" linkIndex="2" dir="l"/> <connection from="W2" to="N1" via=":J1_3" tl="J1" linkIndex="3" dir="s"/> <tlLogic id="J1" type="static" programID="0" offset="0"> <phase duration="0" state="rrrr"/> <!-- 初始全红 --> <phase duration="0" state="GGrr"/> <!-- 东西直行 --> <phase duration="0" state="yyrr"/> <!-- 东西黄灯 --> <phase duration="0" state="rrGG"/> <!-- 南北直行 --> <phase duration="0" state="rryy"/> <!-- 南北黄灯 --> </tlLogic> </net>

提示：state字符串中每个字母对应一个连接（G=绿，r=红，y=黄），顺序必须与<connection>中linkIndex严格一致。SUMO 1.16+ 默认启用--default.action-step-length 1，即每步仿真 1 秒，这是 DQN 时间步对齐的基础。

2.2 使用 traci 实时获取状态：排队长度、等待时间与相位信息

DQN 的状态空间（state）必须反映当前路口拥堵态势。SUMO 提供traci.edge.getLastStepVehicleNumber()获取车道车辆数，但仅数量不够——需区分“排队中”与“正在通行”的车辆。正确做法是：

• 用traci.edge.getLastStepHaltingNumber()获取排队长度（速度 < 0.1 m/s 的车辆数）；
• 用traci.edge.getWaitingTime()获取该车道累计等待时间（单位：秒），此值对 reward 设计至关重要；
• 用traci.trafficlight.getPhase("J1")获取当前相位索引（0~4），结合getPhaseDuration()可计算剩余绿灯时间；
• 所有观测值需归一化至 [0,1] 区间，例如排队长度除以车道最大容量（通常设为 10 辆/车道），等待时间除以 300（5 分钟最大容忍值）。

# state.py import traci def get_state(): # 定义四个进口道：E, W, N, S edges = ["E2", "W2", "N1", "S1"] state_vector = [] for edge in edges: # 排队长度（归一化） halting = traci.edge.getLastStepHaltingNumber(edge) state_vector.append(min(halting / 10.0, 1.0)) # 累计等待时间（归一化） wait_time = traci.edge.getWaitingTime(edge) state_vector.append(min(wait_time / 300.0, 1.0)) # 当前相位索引（one-hot 编码） current_phase = traci.trafficlight.getPhase("J1") phase_onehot = [0.0] * 5 phase_onehot[current_phase] = 1.0 state_vector.extend(phase_onehot) return np.array(state_vector, dtype=np.float32)

注意：getWaitingTime()返回的是该车道所有车辆等待时间之和，不是平均值。若需更精细控制，可改用traci.vehicle.getWaitingTime(veh_id)遍历所有车辆，但开销显著增大，单交叉口场景下聚合指标已足够。

2.3 启动 SUMO 并建立 traci 连接：避免端口冲突与超时错误

SUMO 必须以--start模式启动并监听特定端口，Python 端通过traci.start()连接。常见失败原因是端口被占用或 SUMO 进程未正确退出。可靠做法是：

• 使用subprocess.Popen启动 SUMO，显式指定--remote-port；
• 设置traci.start()的numRetries=3和delay=100，避免连接瞬时失败；
• 在finally块中调用traci.close()，确保仿真结束释放资源；
• 若需多进程训练，每个进程应使用不同端口（如 8813, 8814...）。

# env.py import subprocess import traci import time def start_sumo(sumo_binary="sumo-gui", config_file="example.sumocfg"): # 启动 SUMO 并监听 8813 端口 sumo_cmd = [ sumo_binary, "-c", config_file, "--remote-port", "8813", "--no-step-log", "--no-warnings", "--start" ] process = subprocess.Popen(sumo_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # 等待 SUMO 启动完成 time.sleep(2) try: traci.start(["localhost", 8813], numRetries=3, delay=100) except Exception as e: print(f"Failed to connect to SUMO: {e}") process.terminate() raise return process # 使用示例 if __name__ == "__main__": sumo_proc = start_sumo() try: # 运行仿真步 for step in range(3600): # 1 小时仿真 traci.simulationStep() finally: traci.close() sumo_proc.terminate()

3. DQN 智能体实现：网络结构、经验回放与动作空间编码

3.1 动作空间设计：为什么用“相位持续时间增量”而非“相位切换”

DQN 的动作（action）直接影响信号灯行为。常见误区是将动作定义为“切换到相位 X”，但这导致两个问题：

• 动作空间稀疏：5 个相位只有 5 个离散动作，无法表达“延长当前绿灯 3 秒”这种连续调整；
• 违反交通规则：强制切换可能造成黄灯时间不足或相位冲突。

正确做法是定义动作为空间维度上的“相位持续时间增量”：

• 当前相位为东西直行（phase 1）时，动作a=0表示保持当前相位，a=1表示延长 3 秒，a=2表示缩短 2 秒；
• 动作集大小设为 5（-2,-1,0,+1,+2），对应持续时间变化量 Δt ∈ {-2,-1,0,1,2} 秒；
• 实际执行时，新持续时间 = max(3, min(60, 当前持续时间 + Δt))，确保绿灯不低于 3 秒、不超 60 秒。

# dqn_agent.py import torch import torch.nn as nn import numpy as np class DQNNetwork(nn.Module): def __init__(self, input_dim, action_dim): super().__init__() self.fc1 = nn.Linear(input_dim, 128) self.fc2 = nn.Linear(128, 128) self.fc3 = nn.Linear(128, action_dim) self.relu = nn.ReLU() def forward(self, x): x = self.relu(self.fc1(x)) x = self.relu(self.fc2(x)) return self.fc3(x) class DQNAgent: def __init__(self, state_dim, action_dim, lr=1e-4, gamma=0.99, eps_start=1.0, eps_end=0.01, eps_decay=0.995): self.state_dim = state_dim self.action_dim = action_dim self.gamma = gamma self.epsilon = eps_start self.eps_end = eps_end self.eps_decay = eps_decay self.policy_net = DQNNetwork(state_dim, action_dim) self.target_net = DQNNetwork(state_dim, action_dim) self.target_net.load_state_dict(self.policy_net.state_dict()) self.optimizer = torch.optim.Adam(self.policy_net.parameters(), lr=lr) self.memory = ReplayBuffer(10000) def select_action(self, state): if np.random.random() < self.epsilon: return np.random.randint(0, self.action_dim) else: state_tensor = torch.tensor(state, dtype=torch.float32).unsqueeze(0) with torch.no_grad(): q_values = self.policy_net(state_tensor) return q_values.argmax().item() def update_epsilon(self): self.epsilon = max(self.eps_end, self.epsilon * self.eps_decay)

提示：动作空间大小（5）与相位数（5）无关，它只控制持续时间调整幅度。相位切换由 SUMO 内部逻辑自动触发——当当前相位持续时间耗尽，自动进入下一相位（如 phase 1 → phase 2 → phase 3...），DQN 只负责“喂”给每个相位多少秒。

3.2 经验回放缓冲区：带优先级采样的必要性与实现

标准 DQN 使用均匀采样经验回放，但在交通场景中，短时拥堵爆发（如某进口道突然排队 20 辆）产生的 transition 对策略影响远大于常态数据。因此必须引入优先级经验回放（Prioritized Experience Replay, PER）：

• 每个 transition 存储(state, action, reward, next_state, done)及其 TD error；
• 采样概率正比于|TD error|^α（α=0.6）；
• 更新时按1/√(采样概率)加权修正 loss，避免高优先级样本被过度优化。

# replay_buffer.py import numpy as np import torch class PrioritizedReplayBuffer: def __init__(self, capacity, alpha=0.6, beta=0.4): self.capacity = capacity self.alpha = alpha self.beta = beta self.buffer = [] self.priorities = np.zeros(capacity, dtype=np.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)) 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 # 计算采样概率 priorities = self.priorities[:len(self.buffer)] probs = priorities ** self.alpha probs /= probs.sum() # 采样索引 indices = np.random.choice(len(self.buffer), batch_size, p=probs) samples = [self.buffer[idx] for idx in indices] # 计算重要性采样权重 total = len(self.buffer) weights = (total * probs[indices]) ** (-self.beta) weights /= weights.max() batch = list(zip(*samples)) return ( torch.tensor(np.stack(batch[0]), dtype=torch.float32), torch.tensor(batch[1], dtype=torch.long), torch.tensor(batch[2], dtype=torch.float32), torch.tensor(np.stack(batch[3]), dtype=torch.float32), torch.tensor(batch[4], dtype=torch.bool), torch.tensor(weights, dtype=torch.float32), indices ) def update_priorities(self, indices, priorities): for idx, prio in zip(indices, priorities): self.priorities[idx] = prio

3.3 DQN 训练循环：reward 函数的物理意义与参数敏感性

Reward 函数是 DQN 学习目标的直接体现。交通信号控制的核心矛盾是：减少车辆等待时间 vs 提高路口通行效率。简单 reward（如-waiting_time）会导致智能体频繁切换相位以“重置”等待时间计数器，反而加剧拥堵。必须设计复合 reward：

• 主项：-0.5 * avg_waiting_time（惩罚等待，权重 0.5）；
• 次项：+0.3 * throughput（奖励通行量，权重 0.3）；
• 约束项：-0.2 * phase_switch_penalty（惩罚相位切换，每次 -1，权重 0.2）；
• 所有项需归一化，使 reward 范围稳定在 [-1, 1] 内，避免梯度爆炸。

# reward.py def calculate_reward(): # 获取四个进口道的等待时间（秒） wait_times = [traci.edge.getWaitingTime(edge) for edge in ["E2", "W2", "N1", "S1"]] avg_wait = np.mean(wait_times) / 300.0 # 归一化到 [0,1] # 获取本步通行车辆数（离开路口的车辆） throughput = traci.simulation.getDepartedNumber() # 相位切换检测（记录上一步相位，对比当前） current_phase = traci.trafficlight.getPhase("J1") phase_switch_penalty = 1.0 if current_phase != last_phase else 0.0 last_phase = current_phase reward = -0.5 * avg_wait + 0.3 * min(throughput / 10.0, 1.0) - 0.2 * phase_switch_penalty return np.clip(reward, -1.0, 1.0)

注意：getDepartedNumber()返回本步离开仿真的车辆数，是 throughput 的合理代理。若需更精确，可用traci.simulation.getArrivedNumber()（到达车辆数）替代，但需确保路网出口定义正确。

4. 训练与部署：超参数调优、收敛验证与线上部署要点

4.1 关键超参数影响分析：learning rate、gamma 与 epsilon decay 的实测表现

DQN 在 SUMO 环境中的收敛高度依赖超参数组合。我们对单交叉口（流量 1500 veh/h）进行 5000 episode 训练，记录各参数对平均等待时间的影响：

# 启动训练脚本（含参数扫描） python train.py \ --lr 0.0005 \ --gamma 0.95 \ --eps-decay 0.995 \ --batch-size 64 \ --target-update 100 \ --save-dir ./models/dqn_crossing_v1

提示：--target-update 100表示每 100 步同步一次 target network，避免 Q 值震荡。实践中发现 50~200 是安全区间，小于 50 易发散，大于 200 收敛变慢。

4.2 收敛验证：三类指标缺一不可的监控方法

仅看 reward 曲线不足以判断 DQN 是否真正学会控制。必须同步监控以下三类指标：

• 宏观指标：episode 平均等待时间（秒）、平均通行量（veh/h）、相位切换频次（次/小时）；
• 微观指标：各进口道排队长度分布（直方图）、等待时间累积分布（CDF）、绿灯利用率（实际绿灯时长 / 总周期时长）；
• 鲁棒性指标：在流量突增（+30%）、随机扰动（车辆插入率 ±10%）下的 performance drop < 15%。

# metrics_logger.py import matplotlib.pyplot as plt import numpy as np def log_episode_metrics(episodes, waiting_times, throughputs, switches): # 绘制滑动平均曲线（窗口=50） window = 50 avg_wait = np.convolve(waiting_times, np.ones(window)/window, mode='valid') avg_throughput = np.convolve(throughputs, np.ones(window)/window, mode='valid') plt.figure(figsize=(12, 4)) plt.subplot(1, 3, 1) plt.plot(avg_wait) plt.title('Avg Waiting Time (s)') plt.xlabel('Episode') plt.subplot(1, 3, 2) plt.plot(avg_throughput) plt.title('Avg Throughput (veh/h)') plt.xlabel('Episode') plt.subplot(1, 3, 3) plt.hist(switches[-100:], bins=20) plt.title('Phase Switches per Episode') plt.xlabel('Count') plt.tight_layout() plt.savefig('./logs/training_curves.png')

4.3 线上部署技巧：从训练模型到 SUMO 实时控制的无缝衔接

训练好的模型不能直接用于生产，需解决三个衔接问题：

• 状态同步延迟：SUMO 仿真步长（1 秒）与 DQN 决策周期需对齐。解决方案是设置traci.simulationStep()后立即get_state()，确保状态无滞后；
• 动作执行原子性：traci.trafficlight.setPhaseDuration("J1", new_duration)必须在相位切换前调用，否则无效。正确时机是在get_state()后、simulationStep()前；
• 模型加载与推理加速：使用 TorchScript 导出模型，避免 Python 解释器开销。

# deploy.py import torch import traci # 加载训练好的模型 model_path = "./models/dqn_crossing_v1/best_model.pth" agent = DQNAgent(state_dim=13, action_dim=5) # 4*2 + 5 = 13 维状态 agent.policy_net.load_state_dict(torch.load(model_path)) agent.policy_net.eval() # 启动 SUMO（无 GUI） sumo_cmd = ["sumo", "-c", "example.sumocfg", "--no-warnings", "--no-step-log"] traci.start(sumo_cmd) try: for step in range(3600): # 1. 获取当前状态 state = get_state() # 2. 模型推理（无梯度） with torch.no_grad(): state_tensor = torch.tensor(state, dtype=torch.float32).unsqueeze(0) q_values = agent.policy_net(state_tensor) action = q_values.argmax().item() # 3. 执行动作：调整当前相位持续时间 current_duration = traci.trafficlight.getPhaseDuration("J1") delta_t = {0:-2, 1:-1, 2:0, 3:1, 4:2}[action] new_duration = max(3, min(60, current_duration + delta_t)) traci.trafficlight.setPhaseDuration("J1", new_duration) # 4. 推进仿真 traci.simulationStep() except Exception as e: print(f"Deployment error: {e}") finally: traci.close()

提示：setPhaseDuration()的效果在下一个相位开始时生效。例如当前是 phase 1（东西直行），设置new_duration=25，则下次进入 phase 1 时绿灯为 25 秒。若需立即生效，需先调用traci.trafficlight.setPhase("J1", 1)强制切换，但会破坏相位序列逻辑，不推荐。

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