## 作者信息 > **作者**: 方向 SoloFang > **邮箱**: solofang@aithisway.shop / solofang@gmail.com > **项目**: 空间大脑 · 3DWMTP · Growing Minds 项目组 --- # 3DWMT 原生系统 — 整体技术框架 v1.0 > 基于 3DWMT 协议 + 3D-GDPM 世界模型 + SE(3)-GNN 的机器原生语言完整系统框架 > 版本：v1.0 | 日期：2026-06-14 | 状态：设计阶段 --- ## 目录 1. [系统总览](#1-系统总览) 2. [五层架构详细设计](#2-五层架构详细设计) 3. [层间接口定义](#3-层间接口定义) 4. [数据流向](#4-数据流向) 5. [技术选型与依赖](#5-技术选型与依赖) 6. [项目目录结构](#6-项目目录结构) 7. [实施路线图](#7-实施路线图) 8. [关键模块设计](#8-关键模块设计) 9. [评估体系](#9-评估体系) 10. [风险与缓解](#10-风险与缓解) --- ## 1. 系统总览 ### 1.1 核心命题 ``` 传统机器人系统的问题： 自然语言 → VLM → 1D token → 机器人执行器 ↑ ↑ 维度税 1 维度税 2 本系统的命题： 自然语言 → L5 语言桥 → 3DWMT token → L4 世界模型 → L3 协议栈 → L2 空间感知 → L1 执行器 核心差异：所有层之间的数据交换都使用 3DWMT 协议 → 零维度税（3D 信息不压缩为 1D） ``` ### 1.2 系统边界 ``` 系统包含： ✅ 3DWMT 协议定义（已发布 v0.1） ✅ L1-L5 五层架构定义 ✅ 层间接口定义（Protobuf + gRPC） ✅ 3D-GDPM 世界模型架构 ✅ SE(3)-等变 GNN 实现指南 ✅ Genesis 数据生成管线 ✅ 评估体系 系统不包含（后续独立项目）： ❌ 具体机器人硬件采购与集成 ❌ 真实世界大规模部署 ❌ 商业化产品化 ``` ### 1.3 系统架构总图 ``` ┌─────────────────────────────────────────────────────────────────────┐ │ 用户交互层 │ │ "把桌子上红色的杯子拿过来" │ └────────────────────────┬────────────────────────────────────────────┘ │ 自然语言 ┌────────────────────────▼────────────────────────────────────────────┐ │ L5 语言桥（Language Bridge） │ │ GPT-4V / Qwen-VL + Structured Extraction │ │ 输入：自然语言指令 │ │ 输出：3DWMT Command Token（结构化） │ │ 关键：指令 → 机器可理解的 3D 空间指令 │ └────────────────────────┬────────────────────────────────────────────┘ │ 3DWMT Command Token ┌────────────────────────▼────────────────────────────────────────────┐ │ L4 世界模型（World Model — 3D-GDPM） │ │ SE(3)-Equivariant GNN + Sparse Delta Prediction │ │ 输入：当前 3D 场景（3DWMT tokens）+ 动作 │ │ 输出：预测的未来 3D 场景（3DWMT tokens） │ │ 关键：物理精确的预测（不是像素预测） │ └────────────────────────┬────────────────────────────────────────────┘ │ 3DWMT Scene Tokens ┌────────────────────────▼────────────────────────────────────────────┐ │ L3 3DWMT 协议栈（Protocol Stack） │ │ Protobuf 序列化 + gRPC 传输 + 空间索引 │ │ 输入：上层 3DWMT tokens │ │ 输出：序列化字节流 / 反序列化 tokens │ │ 关键：标准化的 3D 原生通信协议 │ └────────────────────────┬────────────────────────────────────────────┘ │ 3DWMT Scene Tokens ┌────────────────────────▼────────────────────────────────────────────┐ │ L2 空间感知（Spatial Perception） │ │ SAM 2 + DUSt3R + OWLv2 + Spatial Fusion Graph │ │ 输入：多视角 RGB-D 图像 │ │ 输出：3D 场景图（3DWMT tokens） │ │ 关键：像素 → 3D 物件 + 语义标签 │ └────────────────────────┬────────────────────────────────────────────┘ │ 3DWMT Scene Tokens ┌────────────────────────▼────────────────────────────────────────────┐ │ L1 传感执行（Sensing & Actuation） │ │ 摄像头 + 深度传感器 + 机器人执行器 │ │ 输入：物理世界 │ │ 输出：RGB-D 图像 / 关节控制信号 │ │ 关键：标准化硬件接口 │ └────────────────────────┬────────────────────────────────────────────┘ │ 物理交互 ┌────────────────────────▼────────────────────────────────────────────┐ │ 物理世界 │ └─────────────────────────────────────────────────────────────────────┘ 反馈回路（所有层都有）： 执行结果 → L2 重新感知 → L4 更新世界模型 → L5 修正理解 ``` --- ## 2. 五层架构详细设计 ### 2.1 L1 传感执行层 #### 功能定义 ``` 职责： 1. 从物理世界获取感知数据（RGB-D 图像、关节角度、力觉） 2. 将高层的动作指令转换为物理执行器的控制信号 3. 提供标准化的硬件抽象接口 设计原则： - 硬件抽象：上层不需要知道具体硬件型号 - 实时性：控制回路 ≥ 200Hz - 鲁棒性：传感器失效时的降级策略 ``` #### 硬件配置（推荐 PoC 配置） ``` 视觉传感： 主摄像头：Intel RealSense D435i（RGB-D，90Hz） 腕部摄像头：Intel RealSense D405（近距离，60Hz） （可选）全局摄像头：Azure Kinect DK（大视野，30Hz） 机器人执行器： 主机械臂：Franka Emika Panda（7-DoF，位置控制 + 力矩控制） 末端执行器：Robotiq 2F-85（二指夹爪） （可选）移动基座：Clearpath Ridgeback（差速驱动） 计算平台： 边缘计算：NVIDIA Jetson Orin AGX（64GB，200 TOPS） 云端训练：NVIDIA A100 × 4（80GB） ``` #### 软件接口（ROS 2） ```python # L1 对外接口（ROS 2 消息定义） # 文件：l1_interface.msg # 感知数据输出 Geometryase3D[] objects # 检测到的 3D 物件（L2 输入） float32[] joint_angles # 机器人关节角度 float32[] joint_torques # 机器人关节力矩 geometry_msgs/Pose ee_pose # 末端执行器位姿 # 动作指令输入 geometry_msgs/Twist ee_velocity_cmd # 末端速度指令 float32[] joint_torque_cmd # 关节力矩指令 std_msgs/Bool gripper_cmd # 夹爪开合指令 ``` --- ### 2.2 L2 空间感知层 #### 功能定义 ``` 职责： 1. 从 L1 获取 RGB-D 图像 2. 进行 3D 场景理解： - 实例分割（哪个像素属于哪个物体） - 3D 重建（物体的 3D 位置和形状） - 语义标注（这是什么物体） 3. 输出标准化的 3DWMT Scene Token 核心挑战： - 多视角融合：多个摄像头看到同一物体 → 怎么合并？ - 遮挡处理：物体被遮挡一部分 → 怎么推断完整形状？ - 语义对齐：SAM 2 的 mask + OWLv2 的 label → 怎么对应？ ``` #### 处理管线 ``` 输入：N 个视角的 RGB-D 图像 ↓ Step 1: 实例分割（SAM 2） 每个视角输出：M_i 个 mask（2D 像素级分割） ↓ Step 2: 3D 重建（DUSt3R） 每个 mask → 3D 点云（相机坐标系） ↓ Step 3: 语义标注（OWLv2） 每个 mask → 物体类别（"cup", "table", ...） ↓ Step 4: 多视角融合（Spatial Fusion Graph） 不同视角检测到的同一物体 → 合并为一个 3DWMT token ↓ 输出：K 个 3DWMT Scene Tokens ``` #### Spatial Fusion Graph 详细设计 ```python class SpatialFusionGraph: """ 空间融合图：将多视角检测结果合并为统一的 3DWMT token 核心思想： 每个检测结果 = 图的一个节点 节点之间的边 = "可能是同一物体"的置信度 用 GNN 在图上做消息传递 → 输出融合后的 token """ def __init__(self, iou_threshold=0.5, conf_threshold=0.3): self.iou_threshold = iou_threshold self.conf_threshold = conf_threshold def fuse(self, detections: List[Detection]) -> List[ThreeDWorldModelToken]: """ detections: 所有视角的所有检测结果 每个 Detection 有： - bbox_3d: (6,) [xmin, ymin, zmin, xmax, ymax, zmax] - class_name: str - confidence: float - camera_id: int 返回：融合后的 3DWMT tokens（每个物体一个 token） """ # Step 1: 构建图 # 节点：每个 Detection # 边：如果两个 Detection 的 3D IoU > threshold → 连边 graph = self._build_graph(detections) # Step 2: GNN 消息传递 # 目的：让"同一物体"的多个检测结果互相印证 node_features = self._gnn_message_passing(graph) # Step 3: 聚类 # 目的：将"同一物体"的节点聚类为一个 cluster clusters = self._cluster_nodes(node_features, graph) # Step 4: 每个 cluster → 一个 3DWMT token tokens = [] for cluster in clusters: token = self._cluster_to_token(cluster, detections) tokens.append(token) return tokens def _build_graph(self, detections): """构建空间融合图""" N = len(detections) adjacency = torch.zeros((N, N)) for i in range(N): for j in range(i+1, N): # 计算 3D IoU iou = self._compute_3d_iou( detections[i].bbox_3d, detections[j].bbox_3d ) # 如果 IoU 高 或 语义相同 且 视角不同 → 连边 if (iou > self.iou_threshold or (detections[i].class_name == detections[j].class_name and detections[i].camera_id != detections[j].camera_id)): adjacency[i, j] = 1 adjacency[j, i] = 1 return adjacency def _gnn_message_passing(self, adjacency): """GNN 消息传递（简化版）""" # 初始化节点特征 node_features = self._detection_to_feature(detections) # 2 层 GNN 消息传递 for layer in range(2): new_features = [] for i in range(N): # 聚合邻居特征 neighbors = [j for j in range(N) if adjacency[i, j] == 1] if len(neighbors) == 0: agg = torch.zeros_like(node_features[i]) else: agg = torch.stack([node_features[j] for j in neighbors]).mean(dim=0) # 更新节点特征 new_feat = torch.cat([node_features[i], agg]) new_feat = self.gnn_layer(new_feat) new_features.append(new_feat) node_features = torch.stack(new_features) return node_features def _cluster_to_token(self, cluster, detections): """将一个 cluster 转换为 3DWMT token""" # 取 cluster 中置信度最高的 detection 作为主检测 main_det = max(cluster, key=lambda i: detections[i].confidence) # 融合 3D 位置（加权平均） weights = torch.tensor([detections[i].confidence for i in cluster]) weights = weights / weights.sum() pos_3d = sum(detections[i].center_3d * w for i, w in zip(cluster, weights)) # 构建 3DWMT token token = ThreeDWorldModelToken() token.obj_id = f"obj_{main_det.track_id}" token.class_name = main_det.class_name token.pos_3d.x = pos_3d[0] token.pos_3d.y = pos_3d[1] token.pos_3d.z = pos_3d[2] token.confidence = weights.mean().item() return token ``` --- ### 2.3 L3 3DWMT 协议栈层 #### 功能定义 ``` 职责： 1. 3DWMT token 的序列化（token → 字节流） 2. 3DWMT token 的反序列化（字节流 → token） 3. 3DWMT token 的传输（gRPC / WebSocket / ROS 2） 4. 3DWMT token 的空间索引（加速空间查询） 核心特性： - 3D 原生：序列化格式保留 3D 坐标信息 - 高效：稀疏 token 的压缩编码 - 可扩展：支持未来新增 token 类型 ``` #### 序列化格式（基于 Protobuf） ```protobuf // 已有定义（3dwmtp_v0.1.proto），这里说明关键设计决策 // 关键设计 1：空间索引 // 每个 token 有 (x, y, z) 坐标 // 序列化时，按空间位置排序 → 提高压缩率 // 关键设计 2：稀疏编码 // 如果连续多个 token 位置相近 → 用 delta 编码 message SpatialDeltaEncoding { ThreeDWorldModelToken base_token = 1; repeated DeltaToken deltas = 2; // 相对于 base_token 的偏移 } message DeltaToken { float dx = 1; // 通常很小（-0.1 ~ 0.1）→ 高效编码 float dy = 2; float dz = 3; // ... 其他字段的 delta } // 关键设计 3：层次化编码 // 远处物体 → 低精度（用 16-bit float） // 近处物体 → 高精度（用 32-bit float） message AdaptivePrecisionEncoding { repeated ThreeDWorldModelToken near_tokens = 1 [packed=true]; // FP32 repeated CompactToken far_tokens = 2; // FP16，压缩表示 } ``` #### 传输协议选择 ``` 选项 1：gRPC（推荐，内部模块通信） 优势：类型安全、高性能、支持流式 劣势：需要维护连接 适用：L2 ↔ L3 ↔ L4 ↔ L5 之间的通信 选项 2：WebSocket（推荐，跨机器通信） 优势：防火墙友好、浏览器可访问 劣势：需要自己处理消息边界 适用：机器人 ↔ 云端 之间的通信 选项 3：ROS 2 Topic（兼容现有生态） 优势：机器人社区标准 劣势：不保留 3D 结构（flatten 为 1D message） 适用：与现有 ROS 2 系统互通（降级模式） ``` --- ### 2.4 L4 世界模型层（3D-GDPM） #### 功能定义 ``` 职责： 1. 预测：给定当前 3D 场景 + 动作 → 预测未来 3D 场景 2. 规划：给定当前场景 + 目标场景 → 生成动作序列 3. 学习：从数据中学习物理规律 核心创新： - 稀疏预测：只预测变化的物体（不是全部物体） - SE(3) 等变性：旋转场景 → 预测自动旋转 - 物理约束：训练目标包含物理定律（无穿透、动量守恒） ``` #### 3D-GDPM 完整架构 ```python class ThreeDGDPM(nn.Module): """ 3D Gaussian Delta Prediction Model 输入： s_t: 当前 3D 场景（N 个 3D Gaussian） a_t: 执行的动作 输出： delta: 稀疏变化量（只涉及变化的 Gaussian） s_{t+1}: 预测的未来场景 """ def __init__(self, d_model=256, n_layers=6, k_neighbors=10): super().__init__() self.d_model = d_model self.k_neighbors = k_neighbors # 1. Token Embedding # 将 3D Gaussian 参数 → 1D embedding self.token_embed = TokenEmbedding(d_model) # 2. SE(3)-Equivariant GNN # 关键：消息传递满足 SE(3) 等变性 self.se3_gnn = SE3EquivariantGNN(d_model, n_layers) # 3. 稀疏变化检测器 # 预测哪些 Gaussian 会变化 self.sparse_detector = SparseChangeDetector(d_model) # 4. Delta 预测器 # 只对被检测为"会变化"的 Gaussian 预测 delta self.delta_predictor = DeltaPredictor(d_model) # 5. 物理约束验证器（训练时用） self.physics_validator = PhysicsConstraintValidator() def forward(self, s_t, a_t): """ s_t: List[ThreeDWorldModelToken], 长度 N a_t: ActionToken 返回： delta: Dict[int, DeltaToken] # key = Gaussian ID, value = delta s_next_pred: List[ThreeDWorldModelToken] """ N = len(s_t) # === Step 1: Token Embedding === # 每个 Gaussian → 1D embedding + 3D 位置 embeddings = [] # (N, d_model) positions = [] # (N, 3) velocities = [] # (N, 3) for token in s_t: emb = self.token_embed(token) embeddings.append(emb) positions.append([token.pos_3d.x, token.pos_3d.y, token.pos_3d.z]) # 速度（如果有） if token.has_velocity(): velocities.append([token.velocity.vx, token.velocity.vy, token.velocity.vz]) else: velocities.append([0.0, 0.0, 0.0]) embeddings = torch.stack(embeddings) # (N, d_model) positions = torch.tensor(positions) # (N, 3) velocities = torch.tensor(velocities) # (N, 3) # === Step 2: 构建邻接图 === # 每个 Gaussian 连接 k 个最近的邻居 adjacency = self._build_knn_graph(positions, self.k_neighbors) # === Step 3: SE(3)-等变 GNN === # 关键：消息传递满足 SE(3) 等变性 # 即：旋转输入场景 → 输出自动旋转相同的量 updated_embeddings, updated_velocities = self.se3_gnn( embeddings, positions, velocities, adjacency ) # === Step 4: 稀疏变化检测 === # 预测哪些 Gaussian 会变化 change_logits = self.sparse_detector(updated_embeddings, a_t) change_prob = torch.sigmoid(change_logits) # (N,) # 训练时：用 ground truth 监督 # 推理时：用 threshold 决定 if self.training: changing_mask = self.gt_changing_mask # 从数据加载 else: changing_mask = (change_prob > 0.5) changing_indices = changing_mask.nonzero(as_tuple=True)[0] # === Step 5: Delta 预测（只对变化的 Gaussian） === delta_dict = {} for idx in changing_indices: delta = self.delta_predictor( updated_embeddings[idx], updated_embeddings, # 全局上下文 a_t ) delta_dict[idx.item()] = delta # === Step 6: 应用 Delta，生成预测场景 === s_next_pred = self._apply_delta(s_t, delta_dict, positions, updated_velocities) return delta_dict, s_next_pred def _apply_delta(self, s_t, delta_dict, positions, velocities): """应用 delta，生成预测的未来场景""" s_next = [] for i, token in enumerate(s_t): if i in delta_dict: # 这个 Gaussian 会变化 → 应用 delta delta = delta_dict[i] new_token = ThreeDWorldModelToken() new_token.CopyFrom(token) # 位置变化 new_token.pos_3d.x += delta.dx new_token.pos_3d.y += delta.dy new_token.pos_3d.z += delta.dz # 速度变化 new_token.velocity.vx += delta.dvx new_token.velocity.vy += delta.dvy new_token.velocity.vz += delta.dvz # 姿态变化（四元数） # ...（省略旋转更新细节） s_next.append(new_token) else: # 这个 Gaussian 不变 → 直接复制 s_next.append(token) return s_next def compute_loss(self, s_t, a_t, s_next_gt, delta_gt): """ 计算训练损失 s_next_gt: 真实的下一时刻场景（Genesis 仿真器提供） delta_gt: 真实的 delta（用于监督稀疏预测） """ delta_pred, s_next_pred = self.forward(s_t, a_t) # === 损失 1：位置预测精度 === L_pos = 0.0 for i in delta_pred.keys(): pred_pos = torch.tensor([ s_next_pred[i].pos_3d.x, s_next_pred[i].pos_3d.y, s_next_pred[i].pos_3d.z ]) gt_pos = torch.tensor([ s_next_gt[i].pos_3d.x, s_next_gt[i].pos_3d.y, s_next_gt[i].pos_3d.z ]) L_pos += torch.norm(pred_pos - gt_pos) # === 损失 2：稀疏变化检测精度 === changing_gt = torch.zeros(len(s_t)) for i in delta_gt.keys(): changing_gt[i] = 1.0 change_logits = self.sparse_detector(self._embed(s_t), a_t) L_sparse = F.binary_cross_entropy_with_logits( change_logits, changing_gt ) # === 损失 3：物理约束 === L_physics = self.physics_validator(s_next_pred) # === 总损失 === L_total = L_pos + 0.1 * L_sparse + 0.01 * L_physics return L_total ``` #### SE(3)-等变 GNN 消息传递详细实现 ```python class SE3EquivariantMessagePassing(nn.Module): """ SE(3)-等变消息传递层 核心原理： 消息 = 标量部分 × 等变部分 标量部分：在旋转下不变（例如：距离、质量） 等变部分：在旋转下等变（例如：相对位置向量、速度向量） 消息传递后： - 标量特征：通过 MLP 更新（不变） - 等变特征：通过 MLP 更新，然后乘以旋转矩阵（等变） """ def __init__(self, d_model): super().__init__() self.d_model = d_model # 标量特征更新 MLP（输入：不变特征） self.mlp_inv = nn.Sequential( nn.Linear(d_model + 1, d_model), # +1 是因为距离（标量） nn.ReLU(), nn.Linear(d_model, d_model) ) # 等变特征更新 MLP（输入：等变特征） self.mlp_equiv = nn.Sequential( nn.Linear(3, d_model), # 3D 向量 → embedding nn.ReLU(), nn.Linear(d_model, 3) # 输出还是 3D 向量（保持等变性） ) def forward(self, embeddings, positions, velocities, adjacency): """ embeddings: (N, d_model) 标量特征（不变） positions: (N, 3) 3D 位置（等变） velocities: (N, 3) 3D 速度（等变） adjacency: (N, N) 邻接矩阵 """ N = embeddings.shape[0] new_embeddings = embeddings.clone() new_velocities = velocities.clone() for i in range(N): # 找到节点 i 的邻居 neighbors = [j for j in range(N) if adjacency[i, j] == 1] if len(neighbors) == 0: continue # === 计算消息（对每个邻居 j） === messages_inv = [] messages_equiv = [] for j in neighbors: # 相对位置（等变特征） rel_pos = positions[j] - positions[i] # (3,) # 距离（不变特征） dist = torch.norm(rel_pos) # scalar # 相对速度（等变特征） rel_vel = velocities[j] - velocities[i] # (3,) # === 消息 = 标量部分 × 等变部分 === # 标量消息：embedding_i + embedding_j + dist msg_inv = torch.cat([ embeddings[i], embeddings[j], torch.tensor([dist]) ]) # (2*d_model + 1,) msg_inv = self.mlp_inv(msg_inv) # (d_model,) messages_inv.append(msg_inv) # 等变消息：rel_pos + rel_vel msg_equiv = torch.cat([rel_pos, rel_vel]) # (6,) msg_equiv = self.mlp_equiv(msg_equiv) # (3,) messages_equiv.append(msg_equiv) # === 聚合消息 === agg_inv = torch.stack(messages_inv).mean(dim=0) # (d_model,) agg_equiv = torch.stack(messages_equiv).mean(dim=0) # (3,) # === 更新节点 === new_embeddings[i] = embeddings[i] + agg_inv new_velocities[i] = velocities[i] + agg_equiv return new_embeddings, new_velocities ``` --- ### 2.5 L5 语言桥层 #### 功能定义 ``` 职责： 1. 理解自然语言指令（"把桌子上的红杯子拿过来"） 2. 将指令分解为可执行的子目标 3. 生成结构化的 3DWMT Command Token 核心挑战： - 歧义消除："红杯子" → 哪个红杯子？（需要 L2 的感知结果） - 空间推理："桌子上" → 具体坐标是多少？ - 动作规划："拿过来" → 抓取 + 抬起 + 平移 + 放置 ``` #### 处理管线 ``` 输入：自然语言指令 ↓ Step 1: 指令解析（VLM） 输入："把桌子上红色的杯子拿过来" 输出：结构化解析 { "action": "pick_and_place", "object": {"class": "cup", "color": "red"}, "source": {"ref": "table", "relation": "on_top"}, "target": {"ref": "robot_base", "relation": "near"} } ↓ Step 2: 空间消歧（需要 L2 感知结果） 输入：{"class": "cup", "color": "red"} + 当前场景的 3DWMT tokens 输出：目标物体的 obj_id（例如："obj_42"） ↓ Step 3: 生成 3DWMT Command Token 输入：动作 + 目标物体 ID + 目标位置 输出：3DWMT CommandToken（可以被 L4 世界模型消费） ↓ 输出：3DWMT CommandToken ``` #### 实现示例 ```python class LanguageBridge: def __init__(self, vlm_model="gpt-4-vision-preview"): self.vlm = vlm_model # 可以是 GPT-4V, Qwen-VL, LLaVA 等 def parse_instruction(self, instruction: str, scene_tokens: List[ThreeDWorldModelToken]): """ instruction: 自然语言指令 scene_tokens: 当前场景的 3DWMT tokens（来自 L2） 返回：3DWMT CommandToken """ # Step 1: 用 VLM 解析指令 parsed = self._vlm_parse(instruction, scene_tokens) # Step 2: 空间消歧 target_obj_id = self._resolve_object(parsed["object"], scene_tokens) # Step 3: 生成 Command Token cmd_token = ThreeDWorldModelToken() cmd_token.token_type = ThreeDWorldModelToken.TOKEN_COMMAND # 填充指令内容 cmd = CommandToken() cmd.command_id = str(uuid.uuid4()) cmd.action_type = self._action_to_enum(parsed["action"]) cmd.target_object_id = target_obj_id # 目标位置（如果需要放置） if "target" in parsed: target_pos = self._resolve_position(parsed["target"], scene_tokens) cmd.target_position.x = target_pos[0] cmd.target_position.y = target_pos[1] cmd.target_position.z = target_pos[2] cmd_token.command.CopyFrom(cmd) return cmd_token def _vlm_parse(self, instruction, scene_tokens): """ 用 VLM 解析自然语言指令 关键：给 VLM 提供当前场景的"文本描述"作为上下文 """ # 构建场景描述（将 3DWMT tokens 转为文本） scene_description = self._tokens_to_text(scene_tokens) prompt = f""" 当前场景： {scene_description} 用户指令： {instruction} 请输出结构化 JSON： {{ "action": "pick_and_place" | "push" | "pull" | ..., "object": {{"class": "...", "color": "...", ...}}, "source": {{"ref": "...", "relation": "..."}}, "target": {{"ref": "...", "relation": "..."}} // 可选 }} """ response = call_vlm(prompt, model=self.vlm) return json.loads(response) def _resolve_object(self, obj_description, scene_tokens): """空间消歧：找到符合描述的目标物体""" candidates = [] for token in scene_tokens: if self._match_description(token, obj_description): candidates.append(token) if len(candidates) == 0: raise ValueError(f"未找到符合描述的物体：{obj_description}") if len(candidates) == 1: return candidates[0].obj_id # 多个候选 → 需要进一步消歧（用空间关系） # 例如："桌子上的红杯子" 且有两个红杯子 → 选在桌子上的那个 return self._disambiguate_by_spatial_relation( candidates, obj_description.get("source") ) ``` --- ## 3. 层间接口定义 ### 3.1 接口总表 | 接口 | 上游层 | 下游层 | 数据格式 | 传输协议 | |------|--------|--------|---------|---------| | I1 | L5 | L4 | 3DWMT CommandToken | gRPC | | I2 | L4 | L3 | List[ThreeDWorldModelToken] | In-process（直接函数调用） | | I3 | L3 | L2 | 3DWMT Scene Token | gRPC / ROS 2 | | I4 | L2 | L1 | RGB-D Image | ROS 2 Topic | | I5 | L1 | 物理世界 | 关节控制信号 | ROS 2 Control | | I6（反馈） | L1 | L2 | 执行后图像 | ROS 2 Topic | | I7（反馈） | L2 | L4 | 更新后的 Scene Token | In-process | | I8（反馈） | L4 | L5 | 预测结果 | gRPC | ### 3.2 关键接口详细定义（Protobuf） ```protobuf // ========== 接口 I1: L5 → L4 ========== // 文件：interface_l5_l4.proto syntax = "proto3"; package threedwmtp.interface; import "3dwmtp_v0.1.proto"; // L5 发送给 L4 的指令 message L5ToL4Request { // 自然语言指令（调试用） string natural_language_instruction = 1; // 结构化指令（3DWMT Command Token） threedwmtp.CommandToken command = 2; // 当前场景（L2 感知结果） threedwmtp.SceneToken current_scene = 3; } // L4 返回给 L5 的执行计划 message L4ToL5Response { // 是否成功理解指令 bool success = 1; // 如果失败，原因 string failure_reason = 2; // 生成的动作序列（给 L1 执行） repeated threedwmtp.ActionToken action_sequence = 3; // 预测的后续场景（用于验证） repeated threedwmtp.SceneToken predicted_scenes = 4; } // ========== 接口 I2: L4 ↔ L3 ========== // 注意：L4 和 L3 通常在同一进程内，用直接函数调用 // 这里定义的是函数签名（Python 类型提示） """ L4 → L3 接口（函数签名）： def predict_next_scene( current_scene: List[ThreeDWorldModelToken], action: ActionToken ) -> Tuple[Dict[int, DeltaToken], List[ThreeDWorldModelToken]]: \""" 预测下一时刻场景 返回： delta_dict: 稀疏变化量（只涉及变化的物体） next_scene: 完整的预测场景 \""" pass def plan_action_sequence( current_scene: List[ThreeDWorldModelToken], goal_scene: List[ThreeDWorldModelToken], max_steps: int = 10 ) -> List[ActionToken]: \""" 规划动作序列（模型预测控制 MPC） 在 L4 世界模型中展开 K 步 → 找到最优动作序列 \""" pass """ // ========== 接口 I3: L3 → L2 ========== // 文件：interface_l3_l2.proto message L3ToL2Request { // 请求 L2 重新感知场景 bool request_refresh = 1; // 如果需要感知特定区域（空间查询） threedwmtp.BoundingBox3D region_of_interest = 2; } message L2ToL3Response { // 感知结果 threedwmtp.SceneToken scene = 1; // 感知置信度 float confidence = 2; // 感知耗时（毫秒） int32 latency_ms = 3; } ``` --- ## 4. 数据流向 ### 4.1 执行回路（闭环） ``` 时间序列：t = 0, 1, 2, ... t=0（初始状态）： 物理世界 ↓ L1 感知 RGB-D 图像 ↓ L2 感知 3DWMT Scene Token (s_0) ↓ L5 理解指令 3DWMT Command Token ↓ L4 规划 Action Sequence [a_0, a_1, ..., a_K] ↓ L1 执行 物理世界（执行 a_0） t=1（执行一步后）： 物理世界（已变化） ↓ L1 感知 RGB-D 图像 ↓ L2 重新感知 3DWMT Scene Token (s_1) ↓ L4 验证预测 对比 s_1_pred（L4 在 t=0 时的预测）vs s_1（真实） ↓ （如果误差大）在线微调 L4 L4 更新 ↓ L4 继续规划 后续动作序列 [a_1, a_2, ..., a_K'] （可能重新规划） ↓ L1 执行 物理世界（执行 a_1） ...（循环） ``` ### 4.2 数据流大小估算 ``` L1 → L2： RGB-D 图像：4 个视角 × 1280×720×3 字节（RGB）+ 1280×720×2 字节（Depth） ≈ 4 × 3.5 MB ≈ 14 MB / 帧 频率：30 Hz（实时）或 5 Hz（规划时） → 带宽：~100 MB/s（实时）或 ~17 MB/s（规划时） L2 → L3： 3DWMT Scene Token：假设场景中有 50 个物体 每个物体 token ≈ 200 字节（压缩后） → 50 × 200 B = 10 KB / 场景 频率：与感知频率相同（30 Hz 或 5 Hz） → 带宽：~300 KB/s（实时）或 ~50 KB/s（规划时） → 非常轻量！ L3 → L4： 同上：~10 KB / 场景 → 不是瓶颈 L4 内部： 世界模型推理：~50 MB（模型权重）+ ~10 MB（激活值） → 内存占用：~60 MB（可以接受） 推理延迟：~50 ms（A100）→ 20 Hz（够用） L5 → L4： VLM 推理：~1-2 秒（GPT-4V API）或 ~500 ms（本地小模型） → 这是瓶颈！指令理解很慢 → 缓解：指令理解只在任务开始时做一次（不是每帧都做） ``` --- ## 5. 技术选型与依赖 ### 5.1 软件依赖总表 | 组件 | 技术选型 | 版本 | 用途 | 替代方案 | |------|---------|------|------|---------| | **L1 传感** | ROS 2 | Humble (LTS) | 硬件抽象 + 实时控制 | MQTT, Apollo Cyber RT | | **L1 执行器驱动** | frankx | 0.0.12 | Franka Panda 控制 | libfranka, MoveIt 2 | | **L2 实例分割** | SAM 2 | 1.0 | 2D 实例分割 | MobileSAM, FastSAM | | **L2 3D 重建** | DUSt3R | 1.0 | 2D → 3D 点云 | MoGe, Depth Anything V2 | | **L2 语义标注** | OWLv2 | 1.0 | 开放词汇物体检测 | Grounding DINO, YOLO-World | | **L2 融合 GNN** | PyTorch Geometric | 2.4 | Spatial Fusion Graph | DGL | | **L3 序列化** | Protobuf | 4.25 | 3DWMT token 序列化 | Cap'n Proto, FlatBuffers | | **L3 传输** | gRPC | 1.60 | 层间通信 | ROS 2, ZeroMQ | | **L4 世界模型** | PyTorch | 2.1 | 3D-GDPM 实现 | JAX, TensorFlow | | **L4 GNN** | PyTorch Geometric | 2.4 | SE(3)-等变 GNN | DGL | | **L4 物理约束** | Genesis | 0.1.0 | 物理模拟器（训练数据生成） | Isaac Gym, MuJoCo | | **L5 VLM** | GPT-4V API | 2024-04-09 | 指令解析 | Qwen-VL, LLaVA（本地） | | **L5 结构化输出** | guidance | 0.1.0 | 约束 VLM 输出格式 | outlines, jsonformer | ### 5.2 硬件依赖 ``` PoC 阶段（最小配置）： - 计算机：MacBook Pro M3 Max 64GB（开发）+ 云 GPU（训练） - 机器人：Franka Panda + Robotiq 2F-85（租赁：~$2000/月） - 摄像头：Intel RealSense D435i × 2（~$400） - 总预估：$3000（租赁）+ $400（硬件）= $3400 完整系统（推荐配置）： - 边缘计算：NVIDIA Jetson Orin AGX × 1（~$2000） - 云端训练：Lambda Labs A100 × 4（~$5/小时，训练 1 个月 ≈ $3600） - 机器人：Franka Panda 采购（~$60000）+ 备用部件 - 摄像头：Intel RealSense D435i × 4 + Azure Kinect DK × 1（~$2000） - 总预估：~$70000（不含人力） ``` --- ## 6. 项目目录结构 ``` 3dwmtp-native-system/ ├── README.md # 项目总览 ├── requirements.txt # Python 依赖 ├── setup.py # 安装脚本 │ ├── docs/ # 文档 │ ├── architecture.md # 架构设计文档 │ ├── interface_spec.md # 接口详细规范 │ ├── data_format.md # 数据格式说明 │ └── deployment_guide.md # 部署指南 │ ├── proto/ # Protobuf 定义（协议栈） │ ├── 3dwmtp_v0.1.proto # 已有：3DWMT 协议核心定义 │ ├── interface_l2_l3.proto # L2-L3 接口 │ ├── interface_l3_l4.proto # L3-L4 接口 │ ├── interface_l4_l5.proto # L4-L5 接口 │ └── gen/ # 生成的 Python 代码 │ ├── l1_sensing_actuation/ # L1 层实现 │ ├── sensors/ │ │ ├── realsense_wrapper.py # RealSense 摄像头驱动 │ │ └── robotiq_wrapper.py # Robotiq 夹爪驱动 │ ├── actuators/ │ │ ├── franka_controller.py # Franka Panda 控制器 │ │ └── mobile_base_controller.py # 移动基座控制器 │ └── ros_nodes/ │ ├── sensor_node.py # ROS 2 节点：发布传感器数据 │ └── actuator_node.py # ROS 2 节点：接收并执行动作 │ ├── l2_spatial_perception/ # L2 层实现 │ ├── segmentation/ │ │ ├── sam2_wrapper.py # SAM 2 实例分割 │ │ └── fastsam_fallback.py # FastSAM（实时备选） │ ├── reconstruction/ │ │ ├── dust3r_wrapper.py # DUSt3R 3D 重建 │ │ └── depth_anything.py # Depth Anything V2（单目深度估计） │ ├── semantic_labeling/ │ │ ├── owl_v2_wrapper.py # OWLv2 语义标注 │ │ └── grounding_dino_wrapper.py # Grounding DINO（备选） │ ├── fusion/ │ │ ├── spatial_fusion_graph.py # Spatial Fusion Graph │ │ └── gnn_model.py # GNN 模型定义 │ └── l2_node.py # L2 主节点（ROS 2 / gRPC） │ ├── l3_protocol_stack/ # L3 层实现 │ ├── serialization/ │ │ ├── protobuf_serializer.py # Protobuf 序列化 │ │ └── delta_encoder.py # 稀疏 Delta 编码 │ ├── transport/ │ │ ├── grpc_server.py # gRPC 服务端 │ │ ├── grpc_client.py # gRPC 客户端 │ │ └── ros2_bridge.py # ROS 2 ↔ 3DWMT 桥接 │ ├── spatial_index/ │ │ ├── kdtree_index.py # KD-Tree 空间索引 │ │ └── octree_index.py # Octree 空间索引（大规模场景） │ └── l3_node.py # L3 主节点 │ ├── l4_world_model/ # L4 层实现（3D-GDPM） │ ├── model/ │ │ ├── three_d_gdpm.py # 3D-GDPM 主模型 │ │ ├── se3_gnn.py # SE(3)-等变 GNN │ │ ├── sparse_detector.py # 稀疏变化检测器 │ │ └── delta_predictor.py # Delta 预测器 │ ├── physics_constraints/ │ │ ├── penetration_free.py # 无穿透约束 │ │ ├── momentum_conservation.py # 动量守恒 │ │ └── contact_model.py # 接触模型 │ ├── training/ │ │ ├── train.py # 训练脚本 │ │ ├── data_loader.py # Genesis 数据加载器 │ │ └── loss_functions.py # 损失函数定义 │ ├── inference/ │ │ ├── predict.py # 推理脚本 │ │ └── mpc_planner.py # 模型预测控制规划器 │ └── l4_node.py # L4 主节点 │ ├── l5_language_bridge/ # L5 层实现 │ ├── vlm/ │ │ ├── gpt4v_wrapper.py # GPT-4V API 封装 │ │ ├── qwen_vl_wrapper.py # Qwen-VL 本地部署 │ │ └── structured_output.py # 结构化输出（guidance） │ ├── instruction_parsing/ │ │ ├── parser.py # 指令解析器 │ │ └── spatial_grounding.py # 空间消歧 │ └── l5_node.py # L5 主节点 │ ├── data/ # 数据 │ ├── genesis_dataset/ # Genesis 仿真数据 │ ├── real_world_dataset/ # 真实世界数据（少量） │ └── benchmarks/ # 评估基准 │ ├── scripts/ # 脚本 │ ├── setup/ # 安装脚本 │ │ ├── install_deps.sh # 安装依赖 │ │ └── setup_ros2.sh # 安装 ROS 2 │ ├── data_generation/ # 数据生成脚本 │ │ ├── genesis_gen.py # Genesis 数据生成 │ │ └── token_convert.py # Genesis → 3DWMT token 转换 │ ├── training/ # 训练脚本 │ │ ├── train_l4.sh # 训练 L4 世界模型 │ │ └── finetune_l2.sh # 微调 L2 感知模型 │ └── evaluation/ # 评估脚本 │ ├── eval_l4_prediction.py # 评估 L4 预测精度 │ └── eval_end_to_end.py # 端到端评估 │ ├── tests/ # 测试 │ ├── unit_tests/ # 单元测试 │ ├── integration_tests/ # 集成测试 │ └── system_tests/ # 系统测试 │ └── deployment/ # 部署配置 ├── docker/ # Docker 镜像 ├── ros2_launch/ # ROS 2 启动文件 └── config/ # 配置文件 ``` --- ## 7. 实施路线图 ### 7.1 阶段划分 ``` Phase 1（月 1-3）：基础建设 目标：跑通 L1-L2-L3 链路（感知 → 协议 → 可视化） 月 1： ✅ 安装 ROS 2 + 依赖 ✅ 实现 L1 传感器驱动（RealSense + Franka 仿真） ✅ 实现 L2 SAM 2 + DUSt3R + OWLv2 管线 ✅ 实现 L3 Protobuf 序列化/反序列化 ✅ 端到端测试：摄像头图像 → 3DWMT tokens → 可视化 月 2： ✅ 实现 Spatial Fusion Graph（多视角融合） ✅ 优化 L2 延迟（目标：< 100ms） ✅ 实现 L3 空间索引（KD-Tree） ✅ 测试：多物体场景的感知精度 月 3： ✅ 完成 L1-L2-L3 集成测试 ✅ 撰写 Phase 1 报告 ✅ 发布 v0.1 演示视频 Phase 2（月 4-6）：L4 世界模型 目标：实现 3D-GDPM 并验证稀疏预测 月 4： ✅ 安装 Genesis 物理模拟器 ✅ 实现 Genesis → 3DWMT token 转换脚本 ✅ 生成小规模数据集（1000 episodes，单立方体滑动） ✅ 实现 3D-GDPM 模型骨架 月 5： ✅ 实现 SE(3)-等变 GNN ✅ 实现稀疏变化检测器 ✅ 在单立方体任务上训练 ✅ 验证 SE(3) 等变性 月 6： ✅ 扩展到多物体场景（10 个物体） ✅ 加入物理约束损失 ✅ 评估：稀疏预测 vs 全量预测（速度 + 精度） ✅ 撰写 Phase 2 报告 Phase 3（月 7-9）：L5 + 集成 目标：完整的 L1-L5 端到端系统 月 7： ✅ 实现 L5 语言桥（GPT-4V + 结构化输出） ✅ 实现空间消歧（L2 tokens → 物体 ID） ✅ 测试：自然语言指令 → 3DWMT Command Token 月 8： ✅ 实现 MPC 规划器（L4 展开 + 搜索最优动作） ✅ 集成 L1-L5（完整的执行回路） ✅ 在仿真环境中端到端测试 月 9： ✅ Sim-to-Real 迁移（真实 Franka Panda） ✅ 优化延迟（目标：L1-L5 端到端 < 500ms） ✅ 撰写 Phase 3 报告 ✅ 发布完整系统演示视频 Phase 4（月 10-12）：评估 + 论文 目标：严谨评估 + 学术论文 月 10： ✅ 设计评估基准（10 类任务 × 20 次重复） ✅ 评估指标：任务成功率、延迟、泛化能力 ✅ 对比基线：RT-2、DreamerV3、GWM 月 11： ✅ 撰写论文（目标：CoRL 2027 / RSS 2027） ✅ 开源代码 + 预训练模型 ✅ 撰写技术博客 月 12： ✅ 提交会议论文 ✅ 回复审稿意见（如果有预印本） ✅ 项目总结报告 ``` ### 7.2 里程碑与交付物 | 里程碑 | 时间 | 交付物 | |--------|------|---------| | M1: L1-L2-L3 链路跑通 | 月 3 | 演示视频 + Phase 1 报告 | | M2: 3D-GDPM 验证 | 月 6 | 模型 checkpoint + Phase 2 报告 | | M3: 端到端系统集成 | 月 9 | 完整系统演示 + Phase 3 报告 | | M4: 论文投稿 | 月 12 | 会议论文 + 开源仓库 | --- ## 8. 关键模块设计 ### 8.1 L4 训练数据生成管线（Genesis → 3DWMT） ```python # scripts/data_generation/genesis_gen.py import genesis as gs from threedwmtp proto import ThreeDWorldModelToken, DeltaToken class GenesisDataGenerator: """ Genesis 数据生成器 功能： 1. 自动生成随机场景 2. 运行仿真，记录每一帧的 3D 状态 3. 转换为 3DWMT token 格式 4. 计算 delta（变化量） 5. 保存为 TFRecord / HDF5 """ def __init__(self, output_dir="./data/genesis_dataset", n_envs=1000): self.output_dir = output_dir self.n_envs = n_envs # 并行仿真环境数量 # 初始化 Genesis gs.init(backend=gs.gpu, precision="32", logging_level='warning') self.scene = gs.Scene( show_viewer=False, # 不显示 GUI（加速） viewer_camera_xyz=(3.0, 0.0, 2.0), ) def generate_dataset(self, n_episodes=100_000): """生成数据集""" for episode_id in range(n_episodes): # Step 1: 重置场景（随机生成） self._reset_random_scene() # Step 2: 生成随机动作序列 actions = self._generate_random_actions(T=100) # Step 3: 运行仿真，记录每一帧 episode_data = [] for t in range(100): # 执行动作 self.scene.step(actions[t]) # 记录当前帧的 3D 状态 tokens_t = self._scene_to_tokens() # 计算 delta（如果有前一帧） if t > 0: delta = self._compute_delta(tokens[t-1], tokens_t) else: delta = {} # 第一帧没有 delta episode_data.append({ 'tokens': tokens_t, 'actions': actions[t], 'delta': delta, 'timestamp': self.scene.sim_time }) # Step 4: 保存 episode self._save_episode(episode_id, episode_data) if episode_id % 1000 == 0: print(f"Generated {episode_id} / {n_episodes} episodes...") print(f"Dataset generation complete: {n_episodes} episodes") def _reset_random_scene(self): """重置场景（随机生成物体）""" self.scene.reset() # 随机添加 2-10 个物体 n_objects = random.randint(2, 10) self.object_entities = [] for obj_id in range(n_objects): # 随机物体形状 shape = random.choice(['box', 'sphere', 'cylinder']) size = ( random.uniform(0.02, 0.15), random.uniform(0.02, 0.15), random.uniform(0.02, 0.15) ) # 随机位置（不重叠） pos = self._random_non_overlapping_position() # 随机物理属性 mass = random.uniform(0.01, 2.0) friction = random.uniform(0.1, 1.0) # 添加物体到场景 if shape == 'box': entity = self.scene.add_entity( gs.morphs.Box(size=size), surface=gs.surfaces.Default(mass=mass, friction=friction), pos=pos ) elif shape == 'sphere': entity = self.scene.add_entity( gs.morphs.Sphere(radius=size[0]/2), surface=gs.surfaces.Default(mass=mass, friction=friction), pos=pos ) else: # cylinder entity = self.scene.add_entity( gs.morphs.Cylinder(radius=size[0]/2, height=size[2]), surface=gs.surfaces.Default(mass=mass, friction=friction), pos=pos ) self.object_entities.append(entity) # 添加机器人（Franka Panda） self.robot = self.scene.add_entity( gs.morphs.FrankaPanda(), pos=(0.5, 0.0, 0.0) ) def _scene_to_tokens(self) -> List[ThreeDWorldModelToken]: """将 Genesis 场景转换为 3DWMT tokens""" tokens = [] for idx, entity in enumerate(self.object_entities): token = ThreeDWorldModelToken() token.obj_id = f"obj_{idx}" # 位置 pos = entity.get_pos().numpy() token.pos_3d.x = pos[0] token.pos_3d.y = pos[1] token.pos_3d.z = pos[2] # 姿态（四元数） quat = entity.get_quat().numpy() token.orientation.x = quat[0] token.orientation.y = quat[1] token.orientation.z = quat[2] token.orientation.w = quat[3] # 速度 vel = entity.get_vel().numpy() token.velocity.vx = vel[0] token.velocity.vy = vel[1] token.velocity.vz = vel[2] # 角速度 ang_vel = entity.get_ang_vel().numpy() token.angular_velocity.wx = ang_vel[0] token.angular_velocity.wy = ang_vel[1] token.angular_velocity.wz = ang_vel[2] # 物体类别（从 morphs 推断） token.class_name = self._infer_class_name(entity) # 时间戳 token.timestamp = self.scene.sim_time tokens.append(token) return tokens def _compute_delta(self, tokens_t1, tokens_t2): """计算两帧之间的 delta""" delta = {} # 假设 obj_id 是对齐的（简化） for i, (tok1, tok2) in enumerate(zip(tokens_t1, tokens_t2)): # 计算位置变化 dx = tok2.pos_3d.x - tok1.pos_3d.x dy = tok2.pos_3d.y - tok1.pos_3d.y dz = tok2.pos_3d.z - tok1.pos_3d.z # 如果变化大于阈值 → 认为这个物体"变化了" if abs(dx) > 1e-4 or abs(dy) > 1e-4 or abs(dz) > 1e-4: d = DeltaToken() d.dx = dx d.dy = dy d.dz = dz # 速度变化 d.dvx = tok2.velocity.vx - tok1.velocity.vx d.dvy = tok2.velocity.vy - tok1.velocity.vy d.dvz = tok2.velocity.vz - tok1.velocity.vz delta[i] = d return delta def _save_episode(self, episode_id, episode_data): """保存 episode 到磁盘""" output_path = os.path.join(self.output_dir, f"episode_{episode_id:06d}.pkl") with open(output_path, 'wb') as f: pickle.dump(episode_data, f) # 运行 if __name__ == "__main__": generator = GenesisDataGenerator(n_envs=1000) generator.generate_dataset(n_episodes=100_000) ``` ### 8.2 L4 训练脚本 ```python # l4_world_model/training/train.py import torch from torch.utils.data import DataLoader from threedwmtp model import ThreeDGDPM from genesis_dataset import GenesisDataset def train(): # === 超参数 === batch_size = 32 learning_rate = 1e-4 n_epochs = 100 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') # === 数据加载 === train_dataset = GenesisDataset("./data/genesis_dataset/train") train_loader = DataLoader( train_dataset, batch_size=batch_size, shuffle=True, num_workers=8 ) # === 模型初始化 === model = ThreeDGDPM(d_model=256, n_layers=6) model.to(device) # === 优化器 === optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=n_epochs) # === 训练循环 === for epoch in range(n_epochs): model.train() total_loss = 0.0 for batch in train_loader: # 数据移到设备 s_t = batch['s_t'].to(device) # (B, N, ...) a_t = batch['a_t'].to(device) # (B, ...) s_next_gt = batch['s_next_gt'].to(device) # (B, N, ...) delta_gt = batch['delta_gt'] # List[Dict] # 前向传播 delta_pred, s_next_pred = model(s_t, a_t) # 计算损失 loss = model.compute_loss(s_t, a_t, s_next_gt, delta_gt) # 反向传播 optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() total_loss += loss.item() # === 验证 === val_loss = evaluate(model, val_loader, device) # === 日志 === avg_train_loss = total_loss / len(train_loader) print(f"Epoch {epoch+1}/{n_epochs}") print(f" Train Loss: {avg_train_loss:.4f}") print(f" Val Loss: {val_loss:.4f}") # === 保存 checkpoint === if (epoch + 1) % 10 == 0: checkpoint_path = f"./checkpoints/model_epoch_{epoch+1}.pt" torch.save({ 'epoch': epoch, 'model_state_dict': model.state_dict(), 'optimizer_state_dict': optimizer.state_dict(), 'train_loss': avg_train_loss, 'val_loss': val_loss, }, checkpoint_path) print(f" Checkpoint saved: {checkpoint_path}") scheduler.step() print("Training complete!") def evaluate(model, val_loader, device): """评估""" model.eval() total_loss = 0.0 with torch.no_grad(): for batch in val_loader: s_t = batch['s_t'].to(device) a_t = batch['a_t'].to(device) s_next_gt = batch['s_next_gt'].to(device) delta_gt = batch['delta_gt'] delta_pred, s_next_pred = model(s_t, a_t) loss = model.compute_loss(s_t, a_t, s_next_gt, delta_gt) total_loss += loss.item() return total_loss / len(val_loader) if __name__ == "__main__": train() ``` --- ## 9. 评估体系 ### 9.1 评估指标 ``` L2 感知层评估： - 3D IoU（3D 交并比）：预测 bbox vs ground truth bbox - 语义准确率：物体类别标注准确率 - 延迟：端到端感知延迟（毫秒） L4 世界模型评估： - 位置预测误差：||pos_pred - pos_gt||₂（米） - 稀疏准确率：变化的物体是否被准确预测？ - 物理合理性：有无穿透？动量是否守恒？ - 推理延迟：单次预测耗时（毫秒） L5 语言桥评估： - 指令理解准确率：正确解析的指令比例 - 空间消歧准确率：正确识别目标物体的比例 - 延迟：指令 → Command Token 耗时（毫秒） 端到端系统评估： - 任务成功率：成功完成任务的 episode 比例 - 执行延迟：指令 → 开始执行 耗时（毫秒） - 泛化能力：未见过的物体 / 场景 的成功率 ``` ### 9.2 评估基准设计 ``` 任务类别（10 类）： 1. 简单抓取（Simple Pick） - 场景：1 个物体放在桌子上 - 指令："拿起 X" - 成功标准：物体被抓起，不掉落 2. 精确放置（Precise Place） - 场景：1 个物体 + 目标位置标记 - 指令："把 X 放在 Y 位置" - 成功标准：物体放置位置误差 < 2cm 3. 多物体区分（Multi-Object Discrimination） - 场景：3-5 个物体，有相似属性 - 指令："拿起红色的杯子"（有多个红色物体） - 成功标准：正确选择指称的物体 4. 遮挡处理（Occlusion Handling） - 场景：目标物体部分被遮挡 - 指令："拿起 X" - 成功标准：成功抓取（需要处理遮挡） 5. 堆叠操作（Stacking） - 场景：2-3 个物体 - 指令："把 X 放在 Y 上面" - 成功标准：X 稳定堆叠在 Y 上（不倒） 6. 推操作（Pushing） - 场景：1 个物体（太重，无法抓取） - 指令："把 X 推到 Y 位置" - 成功标准：物体被推到目标位置（误差 < 5cm） 7. 铰接物体操作（Articulated Object） - 场景：抽屉 / 门 - 指令："打开抽屉" / "开门" - 成功标准：铰接关节角度 > 阈值 8. 多步骤任务（Multi-Step） - 场景：2-3 个物体，需要多步操作 - 指令："把 A 放进 B，然后放到桌子上" - 成功标准：所有子目标都完成 9. 泛化：新物体（Novel Object） - 场景：训练时未见过的新物体 - 指令："拿起 X"（X 是新物体） - 成功标准：成功抓取新物体 10. 泛化：新语言表述（Novel Language） - 场景：标准场景 - 指令："帮我拿那个能装水的东西"（训练时没见过这种表述） - 成功标准：正确推断"能装水的东西" = 杯子 每个任务：20 次重复 → 成功率 = 成功次数 / 20 ``` --- ## 10. 风险与缓解 ### 10.1 技术风险 | 风险 | 概率 | 影响 | 缓解措施 | |------|------|------|---------| | SE(3)-GNN 训练不稳定 | 中 | 高 | 先用标准 GNN 基线，逐步引入等变性；使用梯度裁剪 | | Genesis 仿真与真实差距大 | 高 | 中 | Domain Randomization；少量真实数据微调 | | 稀疏预测漏掉变化物体 | 中 | 高 | 变化检测器的置信度阈值动态调整；用 Oracle 评估上限 | | L2 感知延迟过高 | 高 | 中 | 模型量化（INT8）；异步感知（预测期间不感知） | | 计算资源不足（需要 A100） | 高 | 中 | 使用云 GPU（Lambda Labs / RunPod）；申请学术资助 | | 3DGS 对透明物体效果差 | 高 | 中 | 混合表示（透明物体用 mesh + SDF） | | VLM API 延迟高（GPT-4V） | 高 | 低 | 指令理解只在任务开始时做一次；换本地小模型（Qwen-VL） | ### 10.2 项目风险 | 风险 | 概率 | 影响 | 缓解措施 | |------|------|------|---------| | 关键成员离职 | 低 | 高 | 文档化所有设计决策；代码详细注释；定期知识分享 | | 硬件损坏（机器人） | 中 | 高 | 购买保险；先只用仿真（Genesis）开发，后期再上真实硬件 | | 预算超支 | 中 | 中 | 优先使用开源工具；云服务按小时计费（不用不收费） | | 论文被拒 | 高 | 中 | 多投几个会议；先发 arXiv 预印本；根据审稿意见持续改进 | --- ## 附录 A：术语表 ``` 3DWMT (3D World Model Token Protocol) 3D 世界模型令牌协议 —— 机器原生 3D 通信协议 3D-GDPM (3D Gaussian Delta Prediction Model) 3D 高斯增量预测模型 —— 本系统提出的世界模型架构 SE(3) 特殊欧几里得群（3D）— 包含所有 3D 刚体变换（旋转 + 平移）的群 等变性（Equivariance） 若输入变换 T，输出也变换 T → f(T·x) = T·f(x) 例：旋转输入场景，预测结果也自动旋转 维度税（Dimensionality Tax） 3D 信息 → 压缩为 1D → 再解压回 3D 的信息损失 稀疏预测（Sparse Prediction） 只预测发生变化的物体（不是全部物体）→ 计算效率提升 空间融合图（Spatial Fusion Graph） 将多视角检测结果合并为统一 3D 场景的图神经网络 Genesis 物理模拟器（2024 年底开源）—— 用于生成训练数据 Delta Token 表示物体状态变化量的 3DWMT token ``` --- ## 附录 B：参考文献 ``` [1] 3DWMT Protocol Specification v0.1. Solo, 2026. [2] SAM 2: Segment Anything in Images and Videos. Meta AI, 2024. [3] DUSt3R: Geometric 3D Vision Made Easy. CVPR 2024. [4] OWLv2: Open-World Object Detection with Vision Transformers. ICLR 2023. [5] SE(3)-Transformers for 3D Point Clouds. ICLR 2020. [6] DreamerV3: Mastering Diverse Domains through World Models. Nat Mach Intell 2023. [7] Genesis: A Universal Physics Engine. arXiv 2024. [8] Gaussian Splatting for Real-Time Radiance Field Rendering. SIGGRAPH 2023. [9] GWM: General World Model with 3D Gaussian Splatting. ICCV 2025. [10] V-JEPA 2: Self-Supervised Video Prediction. Meta AI, 2025. ``` --- ## 文档修订历史 | 版本 | 日期 | 作者 | 变更内容 | |------|------|------|---------| | v1.0 | 2026-06-14 | Solo | 初始版本，完整技术框架 | --- _文档结束_