佛山骏域网站建设专家,开通域名,网站seo的主要优化内容,wordpress iframe插件今天看的是论文Colight:学习网络级合作进行交通信号控制
论文提出的CoLight模型是一种基于强化学习和图注意力网络的交通信号灯控制方法#xff0c;旨在解决城市道路网络中的交通信号的写作问题#xff0c;提升车辆通行效率。
问题定义为#xff1a;
将交通信号控制问题建…今天看的是论文Colight:学习网络级合作进行交通信号控制
论文提出的CoLight模型是一种基于强化学习和图注意力网络的交通信号灯控制方法旨在解决城市道路网络中的交通信号的写作问题提升车辆通行效率。
问题定义为
将交通信号控制问题建模为马尔可夫博弈每个路口由一个智能体控制智能体通过观察部分系统状态当前相位和各车道车辆数选择动作下一时间段的相位目标是最小化路口周围车道的平均队列长度。问题主要由系统状态空间观察空间动作集合转移概率奖励函数策略和折扣因子等部分组成。
模型结构为 观察嵌入层利用多层感知机MLP)将路口的原始观察数据车道车辆数和当前相位嵌入到一个低维的潜在空间生成隐藏状态代表当前路口的交通状况。 图注意力网络层GAT):通过注意力机制实现智能体之间的通信和协作学习邻居路口对目标路口的影响首先计算观察交互分数再进行归一化得到注意力分数然后基于此进行无索引的邻居协作将邻居路口的信息进行加权聚合还使用了多头注意力机制从不同子空间关注邻居信息。 Q值预测层根据前面学习到的邻居表示通过一系列GAT层和全连接层预测每个动作的Q值用于指导智能体的决策通过最小化损失函数来优化当前策略。
关键动态通信和无索引建模
动态通信利用图注意力网络学习邻居路口影响的动态变化避免直接拼接相邻路口交通信息的方式能更好地捕捉交通的时空变化例如在不同时间段相邻路口见的影响方向和程度不同模型可根据实时交通状况调整注意力分配。
无索引模型学习通过学习注意力权重对邻居路口影响进行平均而不是使用固定索引解决了多智能体共享模型参数时因邻居索引固定带来的学习冲突问题减少了学习模型的整体参数。
实验评估部分
实验设置论文在 CityFlow 开源交通模拟器上进行实验使用合成数据不同规模的动脉和网格网络和真实世界数据纽约、杭州和济南的交通数据将模型与传统交通方法如 Fixedtime、MaxPressure和其他强化学习方法如 CGRL、Individual RL 等进行对比。评估指标论文中采用平均旅行时间作为主要评估指标衡量不同模型在控制交通信号时的性能。实验结果论文得到的结果为CoLight 在不同路网和交通模式下相比传统方法和其他 RL 方法平均旅行时间显著降低收敛速度更快在大规模路网中表现出良好的可扩展性。例如在合成数据上平均提升 6.98% 在真实世界数据上平均提升 11.69%。 下面是我的理解 传统方法
1.固定配时比如红灯60秒绿灯40秒不管路上多少车到店就换高峰期容易堵效率低。
2.MaxPressure等传统算法通过计算“压力”比如上下游排队长度来选绿灯方向但它假设“车道永远能容纳车”“车流稳定”但是现实中一堵车就失灵不会灵活应变。
RL方法
让每个路口自己学怎么变灯但是老RL方法也有坑比如把邻居路口的信息直接堆在一起不管这些邻居是主干道还是小路也不管早上和晚上谁的影响更大就像把所有快递都堆在门口分不清哪个急哪个缓。
CoLight方法核心大招让交通灯会看会想会合作。
CoLight就像是给每个交通灯装了个聪明的大脑主要干三件事 1.先看清周围路况观察嵌入层
每个路口的智能体交通灯控制器会收集两个信息------当前灯的状态比如东西方向绿灯、各个车道的车辆数比如东进口道有 20 辆车排队。
然后用一个翻译器--MLP多层感知机把这些数据变成大脑能理解的信号隐藏状态就行人看见堵车信号会提高警惕一样。
2.再想明白哪个更重要图注意力网络GAT
这是CoLight最厉害的地方解决了“邻居影响动态变化”的问题。
比如早晚高峰的区别早上上班时上游路口A往路口B的车多A对B的影响大晚上下班时车流反方向B对A的影响大CoLight会像人一样根据实时车流给不同邻居“打分”也就是注意力权重车少的得分低。
那有人就要问了具体时怎么打分的呢 1.先算关系分比如路口i和邻居j把它们的路况信号相乘算出j对i的重要新初值eij. 2.然后再归一化关注度用softmax函数把初值转化为0-1之间的权重aij类似在所有邻居李j的重要性占比多少。 3.最后加权汇总把所有邻居的路况按权重加起来得到对i整体影响比如主干道邻居权重高小路邻居权重低避免“捡了芝麻丢了西瓜” 使用多头注意力用多个视角看问题
就像找不同的人打听路况有人关注车流量有人关注事故CoLight用5个多头注意力并行计算最后把结果平均让判断更准确。
3.最后“合作决策”无所有建模与参数共享
传统方法给邻居排固定顺序比如东南西北但是不同的路口的重要邻居位置不同比如主干道路口更关注东西方向小路口更关注南北方向。如果强行按统一的顺序学就会打架。
CoLight的解决方法为
不按固定顺序只看权重不管邻居在哪个方向谁的权重高就听谁的就行开会时谁的建议靠谱就听谁的。
所有路口共享学习经验比如路口A和路口B都用同一套”打分规则“但根据自己的邻居动态调整权重打打减少了需要学的参数就行全班同用一套教材但是各自记着不同的笔记。 传统交通灯像 “按剧本演戏”老 RL 方法像 “各顾各的傻瓜”而 CoLight 能根据实时车流给不同邻居路口 “动态分配关注度”比如主干道优先、早高峰关注上游不用死记硬背邻居位置只靠 “谁重要就听谁” 来合作能同时管几百个路口还不怎么增加计算量最终让车跑得更快、堵得更少。
这就好比一群协管员每个人都能快速判断周围哪个方向最堵然后互相 “通气”按重要性分享信息一起把车流疏导得明明白白。
示例代码
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from torch_geometric.nn import GATConv # 或自定义GAT层class ObservationEmbedding(nn.Module):观察嵌入层将原始交通数据转换为隐藏状态def __init__(self, input_dim, embed_dim):super(ObservationEmbedding, self).__init__()self.mlp nn.Sequential(nn.Linear(input_dim, embed_dim),nn.ReLU(),nn.Linear(embed_dim, embed_dim),nn.ReLU())self.input_dim input_dimself.embed_dim embed_dimdef forward(self, x):x: 输入张量形状为[batch_size, input_dim]对应论文公式2h_i Embed(o_i^t) σ(o_i W_e b_e)return self.mlp(x) # 输出形状[batch_size, embed_dim]class GraphAttentionLayer(nn.Module):单头部图注意力层计算邻居路口的注意力权重def __init__(self, in_dim, out_dim, alpha0.2):super(GraphAttentionLayer, self).__init__()self.w_t nn.Parameter(torch.FloatTensor(in_dim, out_dim))self.w_s nn.Parameter(torch.FloatTensor(in_dim, out_dim))self.leaky_relu nn.LeakyReLU(alpha)self.softmax nn.Softmax(dim1)self.in_dim in_dimself.out_dim out_dimdef forward(self, h, adj):h: 隐藏状态张量形状为[num_nodes, in_dim]adj: 邻接矩阵形状为[num_nodes, num_nodes]1表示邻居1. 计算观察交互分数 e_ij (h_i W_t) · (h_j W_s)^T 2. 归一化得到注意力权重 α_ij softmax(e_ij) 3. 加权聚合邻居表示 h_si σ(W_q · Σ(α_ij h_j W_c) b_q) # 特征变换h_t torch.matmul(h, self.w_t) # [num_nodes, out_dim]h_s torch.matmul(h, self.w_s) # [num_nodes, out_dim]# 计算注意力分数attn_scores []for i in range(h.size(0)):# 对每个节点i计算与所有邻居j的分数e_ij torch.matmul(h_t[i], h_s.T) # [num_nodes]# 仅保留邻居节点的分数e_ij e_ij * adj[i].float()e_ij self.leaky_relu(e_ij)# 归一化alpha_ij self.softmax(e_ij) # [num_nodes]attn_scores.append(alpha_ij)attn_scores torch.stack(attn_scores) # [num_nodes, num_nodes]# 加权聚合邻居表示简化版假设W_c为单位矩阵h_neighbors torch.matmul(attn_scores, h) # [num_nodes, in_dim]# 论文中还包含W_c和W_q的变换此处简化为线性层h_si F.relu(torch.matmul(h_neighbors, nn.Parameter(torch.FloatTensor(in_dim, out_dim))))return h_si, attn_scoresclass MultiHeadAttention(nn.Module):多头部注意力机制并行执行多个注意力层def __init__(self, in_dim, out_dim, num_heads5, alpha0.2):super(MultiHeadAttention, self).__init__()self.heads nn.ModuleList([GraphAttentionLayer(in_dim, out_dim, alpha) for _ in range(num_heads)])self.num_heads num_headsself.out_dim out_dimdef forward(self, h, adj):h: [num_nodes, in_dim]adj: [num_nodes, num_nodes]执行H个注意力头输出加权和 head_outputs []attn_weights []for head in self.heads:h_head, attn head(h, adj)head_outputs.append(h_head)attn_weights.append(attn)# 平均多头输出 h_merged torch.mean(torch.stack(head_outputs), dim0)return h_merged, torch.mean(torch.stack(attn_weights), dim0)class CoLightModel(nn.Module):CoLight完整模型观察嵌入多层GATQ值预测def __init__(self, obs_dim, embed_dim, num_phases, num_heads5, num_layers2):super(CoLightModel, self).__init__()self.obs_embed ObservationEmbedding(obs_dim, embed_dim)self.gat_layers nn.ModuleList([MultiHeadAttention(embed_dim, embed_dim, num_heads) for _ in range(num_layers)])self.q_value nn.Linear(embed_dim, num_phases)self.num_layers num_layersdef forward(self, obs, adj):obs: 观察张量形状[num_nodes, obs_dim]adj: 邻接矩阵形状[num_nodes, num_nodes]前向传播流程1. 观察嵌入 2. 多层GAT通信 3. Q值预测 h self.obs_embed(obs) # [num_nodes, embed_dim]for i in range(self.num_layers):h, _ self.gat_layers[i](h, adj) # 每层GAT更新隐藏状态q_values self.q_value(h) # [num_nodes, num_phases]return q_valuesclass CoLightTrainer:CoLight训练器实现强化学习训练逻辑def __init__(self, model, gamma0.99, lr1e-4):self.model modelself.optimizer torch.optim.Adam(model.parameters(), lrlr)self.gamma gammaself.loss_fn nn.MSELoss() # 对应论文公式1和10def train_step(self, obs, adj, actions, next_obs, rewards, dones):单步训练计算Q值损失并更新参数obs: 当前观察 [batch_size, num_nodes, obs_dim]adj: 邻接矩阵 [batch_size, num_nodes, num_nodes]actions: 执行的动作 [batch_size, num_nodes]next_obs: 下一时刻观察 [batch_size, num_nodes, obs_dim]rewards: 奖励 [batch_size, num_nodes]dones: 结束标志 [batch_size, num_nodes]# 展平批量数据batch_size, num_nodes obs.shape[:2]obs_flat obs.reshape(-1, obs.shape[-1])adj_flat adj.reshape(-1, adj.shape[-1])actions_flat actions.reshape(-1)rewards_flat rewards.reshape(-1)dones_flat dones.reshape(-1)# 预测当前Q值和下一时刻Q值current_q self.model(obs_flat, adj_flat) # [batch*nodes, num_phases]next_q self.model(next_obs.reshape(-1, next_obs.shape[-1]), adj_flat).detach() # 目标网络或冻结参数# 计算目标Q值r γ * max(next_q) max_next_q next_q.max(dim1)[0]target_q rewards_flat self.gamma * max_next_q * (1 - dones_flat)# 提取当前动作的Q值current_action_q current_q[torch.arange(batch_size*num_nodes), actions_flat]# 计算损失并优化loss self.loss_fn(current_action_q, target_q)self.optimizer.zero_grad()loss.backward()self.optimizer.step()return loss.item()# 示例构建和使用CoLight模型
def build_colight_model(obs_dim10, num_intersections20, num_phases4):构建CoLight模型实例embed_dim 64model CoLightModel(obs_dimobs_dim,embed_dimembed_dim,num_phasesnum_phases,num_heads5,num_layers2)return model# 模拟数据生成实际应用中从模拟器或真实数据获取
def generate_sample_data(batch_size32, num_nodes20, obs_dim10):生成模拟训练数据obs torch.randn(batch_size, num_nodes, obs_dim)# 构建邻接矩阵简化为距离最近的5个邻居adj torch.zeros(batch_size, num_nodes, num_nodes)for i in range(num_nodes):# 模拟邻居关系实际中基于地理距离或路网拓扑neighbors np.random.choice(num_nodes, 5, replaceFalse)adj[:, i, neighbors] 1actions torch.randint(0, 4, (batch_size, num_nodes))next_obs torch.randn(batch_size, num_nodes, obs_dim)rewards torch.randn(batch_size, num_nodes)dones torch.zeros(batch_size, num_nodes, dtypetorch.bool)return obs, adj, actions, next_obs, rewards, dones
如需更完整的实现可参考论文提供的开源代码https://github.com/wingsweihua/colight 其中包含数据集处理、模拟器对接及完整训练流程。
