1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214
| """ 论文:ForeDrive: Foresight-Guided E2E Autonomous Driving arXiv: 2609.26299
核心方法:规划相关潜在世界模型 + DiT规划器异步耦合 """
import torch import torch.nn as nn import torch.nn.functional as F
class ForeDriveModel(nn.Module): """ ForeDrive: 先知感知端到端驾驶 1. 共享编码器提取当前观测表征 2. JEPA世界模型预测多视野未来 3. DiT规划器使用当前+未来表征生成轨迹 4. TAB注入未来引导而不覆盖当前 """ def __init__(self, latent_dim: int = 512, num_future_horizons: int = 4, num_dit_layers: int = 12): super().__init__() self.online_encoder = OnlineEncoder(latent_dim) self.world_model = JEPAPredictor( latent_dim=latent_dim, num_horizons=num_future_horizons ) self.dit_planner = DiTPlanner( latent_dim=latent_dim, num_layers=num_dit_layers, num_future_horizons=num_future_horizons ) self.gated_fusion = GatedVisualFusion(latent_dim, num_future_horizons) self.future_status_injection = FutureStatusInject(latent_dim) self.tab = TrajectoryAdaptiveBias(latent_dim) def forward(self, current_obs, trajectory_queries=None): """ Args: current_obs: (B, C, H, W) 当前前视图 trajectory_queries: (B, T, 2) 轨迹查询点 Returns: trajectory: (B, T, 2) 预测轨迹 future_latents: (B, num_horizons, latent_dim) 未来表征 """ z_current = self.online_encoder(current_obs) with torch.no_grad(): future_latents = self.world_model(z_current) fused = self.gated_fusion(z_current, future_latents) fused = self.future_status_injection(fused, future_latents) fused = self.tab(fused, trajectory_queries) trajectory = self.dit_planner(fused, trajectory_queries) return trajectory, future_latents
class OnlineEncoder(nn.Module): """共享在线编码器:当前观测→潜在表征""" def __init__(self, latent_dim): super().__init__() self.conv = nn.Sequential( nn.Conv2d(3, 64, 7, stride=4, padding=3), nn.GELU(), nn.Conv2d(64, 128, 3, stride=2, padding=1), nn.GELU(), nn.Conv2d(128, 256, 3, stride=2, padding=1), nn.GELU(), nn.Conv2d(256, 512, 3, stride=2, padding=1), nn.GELU(), nn.AdaptiveAvgPool2d(1), nn.Flatten(), nn.Linear(512, latent_dim) ) def forward(self, x): return self.conv(x)
class JEPAPredictor(nn.Module): """JEPA潜在世界模型:预测多视野未来""" def __init__(self, latent_dim, num_horizons): super().__init__() self.num_horizons = num_horizons self.predictors = nn.ModuleList([ nn.Linear(latent_dim, latent_dim) for _ in range(num_horizons) ]) def forward(self, z_current): futures = [] for i, pred in enumerate(self.predictors): z_future = pred(z_current) futures.append(z_future) return torch.stack(futures, dim=1)
class GatedVisualFusion(nn.Module): """门控视觉融合:当前+未来表征""" def __init__(self, latent_dim, num_horizons): super().__init__() self.gates = nn.ModuleList([ nn.Sequential( nn.Linear(latent_dim * 2, latent_dim), nn.Sigmoid() ) for _ in range(num_horizons) ]) self.projections = nn.ModuleList([ nn.Linear(latent_dim, latent_dim) for _ in range(num_horizons) ]) def forward(self, z_current, future_latents): B, H, D = future_latents.shape fused = z_current for h in range(H): gate = self.gates[h]( torch.cat([z_current, future_latents[:, h]], dim=-1) ) projected = self.projections[h](future_latents[:, h]) fused = fused + gate * projected return fused
class TrajectoryAdaptiveBias(nn.Module): """TAB: 轨迹自适应偏置""" def __init__(self, latent_dim): super().__init__() self.bias_net = nn.Sequential( nn.Linear(latent_dim + 4, latent_dim), nn.GELU(), nn.Linear(latent_dim, latent_dim) ) def forward(self, fused, trajectory_queries): if trajectory_queries is not None: traj_info = torch.cat([ trajectory_queries.mean(dim=1), trajectory_queries.std(dim=1), ], dim=-1) bias = self.bias_net(torch.cat([fused, traj_info], dim=-1)) return fused + bias return fused
class DiTPlanner(nn.Module): """Diffusion Transformer规划器""" def __init__(self, latent_dim, num_layers, num_future_horizons): super().__init__() self.num_layers = num_layers self.layers = nn.TransformerEncoderLayer( d_model=latent_dim, nhead=8, dim_feedforward=latent_dim*4, batch_first=True ) self.transformer = nn.TransformerEncoder(self.layers, num_layers) self.trajectory_head = nn.Linear(latent_dim, 2) def forward(self, fused, trajectory_queries): x = fused.unsqueeze(1) out = self.transformer(x) trajectory = self.trajectory_head(out.squeeze(1)) return trajectory
if __name__ == "__main__": B = 4 model = ForeDriveModel(latent_dim=512, num_future_horizons=4) obs = torch.randn(B, 3, 224, 224) traj_q = torch.randn(B, 10, 2) traj, futures = model(obs, traj_q) print(f"输入: {obs.shape}") print(f"轨迹: {traj.shape}") print(f"未来表征: {futures.shape}") print(f"\n模型参数量: {sum(p.numel() for p in model.parameters()):,}") print("\n=== 论文性能 ===") print("NAVSIM v1 PDMS: 89.9") print("NAVSIM v2 EPDMS: 90.0") print("视觉输入: 仅前视单图") print("强化学习: 不需要") print("外部评分器: 不需要")
|