企业官网建设流程全解析

做强化学习的同学大多遇到过一个绕不开的困境：算法代码写得挺顺，训练脚本一跑就崩，不是因为梯度爆炸，而是因为真实环境太贵、太慢、太危险。机械臂撞一次可能损坏硬件，自动驾驶在真实道路上做探索根本不现实，LLM 智能体调一次外部工具要等好几秒还可能产生副作用。于是近两年一个思路被反复提起：既然不能拿真实环境反复试错，那就先让算法在一个"学出来的环境"里练，练熟了再搬到真实世界。

这个"学出来的环境"就是世界模型（World Model）。本文会从概念、结构、代码实现到三个典型场景（自动驾驶、机器人、LLM 智能体）的接入方式，完整拆一遍，并给出一份可以直接跑起来的最小可运行实现。读者不需要有 model-based RL 的基础，只要有 PyTorch 和一点强化学习的常识就能跟上；有经验的开发者可以直接跳到第 4 章看工程实现和第 6 章的排查清单。

1. 为什么强化学习需要一个"想象出来的训练场"

在动手写代码之前，有必要先想清楚一件事：世界模型解决的到底是什么问题。如果只是想"跑个强化学习 demo"，用现成的仿真器就够了，完全没必要引入额外的模型。真正需要世界模型的场合，往往是被下面几个硬约束逼出来的。

1.1 真实环境训练的四个硬约束

第一个约束是采样成本。强化学习对样本量的渴求是出了名的，一个策略从随机初始化到收敛，动辄需要几十万到上千万次交互。在仿真里这可能是几分钟，在真实设备上就是几周甚至几个月，而且每一步都在消耗真实资源。

第二个约束是安全边界。探索本身意味着试错，而试错在真实场景里往往不可接受。让一个还没学会刹车的策略去控制真车，让一个还在随机探索的机械臂去抓取玻璃器皿，风险不是"效果差一点"，而是"设备损坏"甚至"人身危险"。

第三个约束是可重复性。真实世界不可回放，同一时刻、同一光照、同一物体摆放位置基本无法精确复现。这给调试和消融实验带来巨大麻烦：你改了超参，效果变差了，是因为超参，还是因为这次运气不好？

第四个约束是长尾场景的稀缺性。自动驾驶里真正决定安全的是那些极低概率的边角案例——突然横穿的电动车、逆光的隧道口、暴雨中的模糊车道线。这些场景在真实数据里出现频率极低，靠实车采集几乎不可能收集到足够样本。

世界模型的价值就在于：它在数据层面把这些约束松开了。一旦你有了一个能从历史数据里学出来的转移模型，你就可以在它内部无限次地"假设——推演——评估"，单次想象的成本只剩下一次前向计算。

1.2 世界模型到底在做什么

用一句通俗的话说：世界模型是一个会做预测的环境替身。你给它当前状态和一个动作，它告诉你"接下来大概会发生什么"，包括下一个状态是什么、拿到多少奖励、这一局是否结束。

形式化一点，强化学习里的环境通常被建模为一个马尔可夫决策过程，核心是转移函数 $P(s' \mid s, a)$ 和奖励函数 $R(s, a)$。真实环境里这两个东西是黑盒，你只能通过交互采样。世界模型做的事情，就是用神经网络去拟合这两个函数，得到 $\hat{P}(s' \mid s, a)$ 和 $\hat{R}(s, a)$。

关键在于，大多数现代世界模型并不直接预测原始观测（比如一张 1280×720 的图像），而是先编码到一个低维的隐状态再预测。原因很朴素：预测一张高清图像的下一个像素，绝大部分算力都花在了与决策无关的细节上（树叶的纹理、墙面的噪点），而真正影响决策的信息（车距、速度、相对方位）只占很小一部分。在隐空间里做预测，既省算力，又天然过滤了噪声。

1.3 容易混淆的四个概念

这里有几个经常被混用的词，先做一个区分，后面看代码时会更清楚。

需要强调的是，世界模型和仿真器不是替代关系。仿真器的优势是物理规律正确、可解释、边界清晰；劣势是建模成本高、难以覆盖真实世界的复杂性和随机性（尤其是涉及到人、涉及到软体、涉及到视觉外观时）。世界模型的优势是能从真实数据里自动学习那些"写不进物理公式"的部分；劣势是它会犯错，而且错误会随着推演步数累积。

所以在实际工程里，两者往往是组合使用的：用仿真器提供基础物理约束和安全边界，用世界模型补充真实数据的分布特征，或者反过来用仿真数据预训练、用真实数据微调。

1.4 三类最适合落地的场景

从工程角度看，世界模型最适合的场景有三个共同特征：真实交互昂贵、状态可观测、数据可以积累。

自动驾驶是典型代表。路测数据是现成的，而且量很大，但这些数据是"被动采集"的——你只能看到人类驾驶员怎么开，看不到"如果当时打了方向盘会怎样"。世界模型的价值在于把这些被动数据变成可以反事实推演的东西：给定同一段历史，我可以让策略做出不同选择，看推演结果如何。

机器人操作是另一个典型。机械臂的每一次真实抓取都有磨损成本，而仿真到现实的差距（sim-to-real gap）又是出了名的大。世界模型在这里的作用是充当一个"更贴近现实的中间层"，用真实采集的数据校准仿真器学不到的接触力学、摩擦、形变。

LLM 智能体是最近一年增长最快的方向。智能体调用工具、修改文件、查询数据库，每一次动作都有真实副作用，而且延迟高。把"动作执行后的环境反馈"建模成一个可预测、可回滚的沙箱，就能让智能体在真正动手之前先"预演"一遍，大幅降低误操作率。

2. 环境准备与版本说明

本文的代码以"能跑通、能看懂、能改"为第一目标，所以尽量把依赖压到最低。核心实现只依赖 PyTorch 和 NumPy，不引入任何强化学习框架，避免读者卡在版本兼容问题上。

2.1 软硬件与依赖版本

具体的版本需要根据你的项目实际情况调整，本文示例以常见环境为例，重点演示配置思路。下面这套组合是经过验证比较稳的：

• 操作系统：Linux（Ubuntu 20.04 及以上）或 macOS，Windows 下也可以运行，但建议使用 WSL2 以获得更好的多进程体验。
• Python：3.9 及以上，推荐 3.10 或 3.11。
• PyTorch：2.0 及以上版本。本文代码只用到nn.Module、optim.Adam、functional.mse_loss这些长期稳定的 API，跨版本兼容性很好。
• NumPy：1.22 及以上。
• 硬件：纯 CPU 即可跑通全部示例；如果有 GPU，把设备切过去会快 5 到 10 倍，但不是必须的。

安装命令如下：

# 文件路径：requirements.txt # 建议在虚拟环境中安装，避免污染全局环境 python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install --upgrade pip pip install torch>=2.0.0 numpy>=1.22.0

如果你的机器有 CUDA，可以从官方渠道选择对应 CUDA 版本的安装命令，这样torch.cuda.is_available()会返回True，本文代码会自动切换到 GPU 运行。

2.2 示例项目结构

为了让代码结构清晰，我们把环境、模型、数据采集、规划器、主流程拆成独立文件。这份结构在后续扩展到真实场景时也基本不用大改：

world_model_demo/ ├── env.py # 被控环境（真实环境的替身） ├── replay.py # 数据采集与回放缓冲 ├── model.py # 世界模型（编码器 + 转移 + 奖励 + 终止） ├── train_wm.py # 世界模型训练循环 ├── planner.py # 基于世界模型的 CEM 规划器 ├── evaluate.py # 策略评估工具 └── main.py # 串起完整流程的入口

这个拆分有一个好处：当你想把env.py换成一个真实的仿真器（比如自动驾驶仿真、机器人仿真），只要保持reset()和step()的接口一致，其他文件几乎不用动。这也是本文推荐的工程习惯——把环境当插件，把世界模型当可替换组件。

2.3 一个不依赖 gym 的最小环境

为了让示例完全自包含，我们自己写一个一维质点控制环境。它的动力学带阻尼、奖励是二次型，虽然不是线性的，但足够简单，可以手算验证，同时又足够"有料"，能体现世界模型学到的确实是非线性关系。

# 文件路径：world_model_demo/env.py import numpy as np class PointMassEnv: """一维质点控制环境。 状态: [位置 x, 速度 v] 动作: 加速度 a, 取值范围 [-1, 1] 动力学: v' = v + (a - damping * v) * dt x' = x + v' * dt 奖励: -(x'^2) - 0.01 * a^2 终止: |x'| > 3.0 或步数超过 max_steps """ def __init__(self, dt=0.05, damping=0.5, max_steps=200, seed=0): self.dt = float(dt) self.damping = float(damping) self.max_steps = int(max_steps) self.rng = np.random.default_rng(seed) self.state = np.zeros(2, dtype=np.float32) self.steps = 0 @property def obs_dim(self): return 2 @property def act_dim(self): return 1 def reset(self): # 初始位置在 [-1.5, 1.5] 均匀采样，速度在 [-0.5, 0.5] 均匀采样 x = self.rng.uniform(-1.5, 1.5) v = self.rng.uniform(-0.5, 0.5) self.state = np.array([x, v], dtype=np.float32) self.steps = 0 return self.state.copy() def step(self, action): a = float(np.clip(np.asarray(action).reshape(-1)[0], -1.0, 1.0)) x, v = self.state v_new = v + (a - self.damping * v) * self.dt x_new = x + v_new * self.dt self.state = np.array([x_new, v_new], dtype=np.float32) self.steps += 1 reward = -(x_new ** 2) - 0.01 * (a ** 2) done = bool(abs(x_new) > 3.0 or self.steps >= self.max_steps) return self.state.copy(), float(reward), done

这个环境有两个细节值得注意。第一，阻尼项-damping * v让动力学变成速度相关的非线性形式，世界模型如果只学线性映射是拟合不好的，这正好可以验证模型的表达能力。第二，奖励里加了动作惩罚-0.01 * a^2，避免学出一个"疯狂来回抖动"的策略，这也是实际控制任务里非常常见的做法。

3. 世界模型的核心结构拆解

在写完整代码之前，先把世界模型的组成部分拆开讲清楚。理解了每一块在做什么，后面看代码时就不会觉得是在看黑盒。

3.1 五个组件：编码、转移、解码、奖励、终止

一个最小的隐空间世界模型通常由五个子网络组成：

编码器：把原始观测映射到低维隐状态，z = enc(s)。它的作用是压缩。在图像输入场景里，这通常是一个卷积网络或 ViT；在本例的 2 维输入里，就是一个小的 MLP。

转移模型：给定当前隐状态和动作，预测下一个隐状态，z' = z + trans(z, a)。注意这里采用的是残差形式，预测的是增量而不是绝对值。这个技巧在实践里非常重要，因为隐状态本身可能是一个量级较大的向量，直接预测绝对值容易让网络在训练初期产生剧烈波动。

解码器：把隐状态还原回观测空间，s_hat = dec(z)。它的主要作用不是用于决策，而是提供一个额外的监督信号，防止编码器把不同的观测映射到同一个隐状态（也就是表示坍缩）。

奖励头：从隐状态和动作预测即时奖励，r_hat = reward(z, a)。这一项直接决定了想象轨迹的优化目标是否正确。

终止头：预测这一局是否结束，输出一个 logit，训练时用二元交叉熵。它的作用是让想象轨迹在合适的时候停止，避免在一个"已经失败"的状态上继续推演。

3.2 隐空间滚动与想象轨迹

有了上面五个组件，就可以做**想象（imagination）**了。流程很简单：

1. 从真实环境采一个观测s_0，编码成z_0。
2. 给定一个动作序列a_0, a_1, ..., a_{H-1}。
3. 循环调用转移模型，得到z_1, z_2, ..., z_H，同时拿到每一步的预测奖励。
4. 把奖励按折扣累加，得到一个想象中的回报。

这个回报就是策略优化的目标。整个过程不需要访问真实环境，所有的计算都在神经网络内部完成，所以可以在几毫秒内评估成百上千条候选轨迹。

需要提醒的是，想象误差会随步数累积。第 1 步的预测可能很准，第 20 步可能已经完全偏离现实。这是世界模型最核心的工程难题，第 6 章会专门讲怎么缓解。

3.3 训练目标：三路损失加两个辅助头

世界模型的训练本质上是监督学习，数据就是从环境里采的转移元组(s, a, s', r, done)。损失函数由几部分组成：

其中.detach()这一步是关键。如果不 detach，编码器会"偷懒"：既然只要把所有输入都映射成一个常数，转移损失就恒为 0，那何必费力学一个有区分度的表示呢？加上 detach 之后，转移目标被当作固定的"标签"，编码器只能通过重构损失和预测损失获得梯度，表示坍缩的动机就被切断了。

这个思路和自监督学习里的 BYOL、SimSiam 是同一类，只不过这里的目标网络用的是同一套参数的慢拷贝（直接 detach 当前输出）。

3.4 两条策略优化路线：规划与策略梯度

拿到世界模型之后，怎么用它来优化策略，有两条主流路线。

第一条是基于规划的路线（Planning）。不训练显式的策略网络，而是在每一步实时地做优化：采样一堆候选动作序列，用世界模型评估它们的想象回报，挑出最好的那一批，再在里面细化。这一类的代表是 MPC（模型预测控制）和 CEM（交叉熵方法）。优点是实现简单、不用调策略网络；缺点是推理时有额外计算开销。

第二条是基于策略梯度的路线（Policy Learning）。在世界模型里做 rollout，把想象出来的轨迹当作"真实经验"，直接喂给 PPO、SAC 这类算法去更新策略网络。这一类的代表是 Dreamer 系列和 MBPO。优点是推理时零额外开销；缺点是训练稳定性更难控制，因为策略会主动去寻找世界模型的误差区域（也就是所谓的"模型利用"问题）。

本文的最小实现选择第一条路线，因为它更容易调试、更容易观察中间结果。第 5 章会讨论在真实项目里怎么选。

4. 完整实战案例：从零训练一个可用的世界模型

这一章我们把所有代码串起来，形成一个可以完整跑通的流程。整个流程分五步：采数据、建模型、训模型、写规划器、评估对比。

4.1 数据采集脚本

第一步是在真实环境里用随机策略采一批转移数据。这一步用的是真实环境，但只采一次，后续所有训练都在离线数据上完成。

# 文件路径：world_model_demo/replay.py import numpy as np def collect_random_data(env, n_episodes=200, seed=0): """用均匀随机策略采集转移元组。 返回: (obs, act, next_obs, rew, done)，均为 float32 的 ndarray """ rng = np.random.default_rng(seed) obs_buf, act_buf, nxt_buf, rew_buf, done_buf = [], [], [], [], [] for _ in range(n_episodes): s = env.reset() done = False while not done: a = rng.uniform(-1.0, 1.0, size=(env.act_dim,)).astype(np.float32) s_next, r, done = env.step(a) obs_buf.append(s) act_buf.append(a) nxt_buf.append(s_next) rew_buf.append(r) done_buf.append(float(done)) s = s_next return ( np.asarray(obs_buf, dtype=np.float32), np.asarray(act_buf, dtype=np.float32), np.asarray(nxt_buf, dtype=np.float32), np.asarray(rew_buf, dtype=np.float32), np.asarray(done_buf, dtype=np.float32), )

200 个 episode、每个最多 200 步，理论上限是 4 万条数据。实际会因为提前终止而少一些。这个量级对于 2 维状态空间来说已经相当充足。在真实项目里，数据量通常要大得多，而且需要考虑数据的多样性——如果只用一种策略采数据，世界模型在那种策略的分布外会表现得很差。

4.2 世界模型实现

下面是世界模型的核心代码。为了让读者能一眼看出每个模块的职责，我把五个子网络分别命名，没有做过度的封装。

# 文件路径：world_model_demo/model.py import torch import torch.nn as nn import torch.nn.functional as F class WorldModel(nn.Module): """隐空间世界模型：编码器 + 转移 + 解码器 + 奖励头 + 终止头。""" def __init__(self, obs_dim=2, act_dim=1, latent_dim=32, hidden=128): super().__init__() self.obs_dim = obs_dim self.act_dim = act_dim self.latent_dim = latent_dim self.encoder = nn.Sequential( nn.Linear(obs_dim, hidden), nn.SiLU(), nn.Linear(hidden, hidden), nn.SiLU(), nn.Linear(hidden, latent_dim), ) # 残差转移：预测隐状态增量，训练更稳定 self.transition = nn.Sequential( nn.Linear(latent_dim + act_dim, hidden), nn.SiLU(), nn.Linear(hidden, hidden), nn.SiLU(), nn.Linear(hidden, latent_dim), ) self.decoder = nn.Sequential( nn.Linear(latent_dim, hidden), nn.SiLU(), nn.Linear(hidden, hidden), nn.SiLU(), nn.Linear(hidden, obs_dim), ) self.reward_head = nn.Sequential( nn.Linear(latent_dim + act_dim, hidden), nn.SiLU(), nn.Linear(hidden, 1), ) self.done_head = nn.Sequential( nn.Linear(latent_dim + act_dim, hidden), nn.SiLU(), nn.Linear(hidden, 1), ) def encode(self, obs): return self.encoder(obs) def step_latent(self, z, act): """在隐空间推进一步，返回 (z_next, reward, done_logit)。""" za = torch.cat([z, act], dim=-1) z_next = z + self.transition(za) r = self.reward_head(za).squeeze(-1) d_logit = self.done_head(za).squeeze(-1) return z_next, r, d_logit def loss(self, obs, act, next_obs, rew, done): z = self.encode(obs) z_next, r_hat, d_logit = self.step_latent(z, act) # 1) 重构当前观测，防止表示坍缩 recon_loss = F.mse_loss(self.decoder(z), obs) # 2) 隐空间转移损失，目标 detach with torch.no_grad(): z_next_target = self.encode(next_obs) dyn_loss = F.mse_loss(z_next, z_next_target) # 3) 从预测隐状态还原下一观测，主监督信号 pred_loss = F.mse_loss(self.decoder(z_next), next_obs) # 4) 奖励回归 reward_loss = F.mse_loss(r_hat, rew) # 5) 终止预测，正样本极不平衡，加 pos_weight pos_weight = torch.full_like(done, 10.0) done_loss = F.binary_cross_entropy_with_logits( d_logit, done, pos_weight=pos_weight ) total = (0.5 * recon_loss + 1.0 * dyn_loss + 1.0 * pred_loss + 1.0 * reward_loss + 0.5 * done_loss) logs = { "recon": recon_loss.item(), "dyn": dyn_loss.item(), "pred": pred_loss.item(), "reward": reward_loss.item(), "done": done_loss.item(), "total": total.item(), } return total, logs @torch.no_grad() def imagine(self, obs, action_seq, discount=0.97): """给定观测和动作序列，在隐空间推演并返回预测回报。 obs: (B, obs_dim) action_seq: (B, H, act_dim) 返回: (B,) 的折扣回报 """ z = self.encode(obs) horizon = action_seq.shape[1] total = torch.zeros(z.shape[0], device=z.device) gamma = 1.0 for t in range(horizon): z, r, _ = self.step_latent(z, action_seq[:, t, :]) total = total + gamma * r gamma *= discount return total

这里有几个工程细节值得展开说明。

激活函数选了 SiLU 而不是 ReLU。SiLU（也叫 Swish）是x * sigmoid(x)，在 0 附近是平滑的，梯度不会像 ReLU 那样在负半轴突然断裂。对于连续控制的动力学建模，平滑激活通常能带来更稳定的训练曲线。如果你的任务对推理延迟极其敏感，换成 ReLU 也可以，但要重新调一下学习率。

转移采用残差形式。z_next = z + trans(za)而不是z_next = trans(za)。这个改动带来的差别很大：残差形式意味着网络只需要学"变化量"，而在很多控制任务里，相邻两步的隐状态差别很小，直接预测绝对值等于是让网络在学一个恒等映射再加上一点扰动，浪费了大量容量。

终止损失的正样本加权。在 200 步的 episode 里，只有最后一步done=1，正负样本比例是 1:199。如果不加权，网络会直接学成"永远输出 0"，因为这样准确率就有 99.5%。加pos_weight=10让正样本的损失权重更高，网络才会认真去学这个边界。

4.3 训练循环

训练循环本身很朴素，就是标准的小批量监督学习。关键在于把数据转成 tensor 之后放进同一个设备，避免设备不匹配的报错。

# 文件路径：world_model_demo/train_wm.py import torch def train_world_model(model, data, steps=4000, batch_size=256, lr=3e-3, device="cpu", log_every=500): obs, act, nxt, rew, done = [torch.as_tensor(x, device=device) for x in data] n = obs.shape[0] optimizer = torch.optim.Adam(model.parameters(), lr=lr) model.train() for step in range(1, steps + 1): idx = torch.randint(0, n, (batch_size,), device=device) loss, logs = model.loss( obs[idx], act[idx], nxt[idx], rew[idx], done[idx] ) optimizer.zero_grad(set_to_none=True) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=5.0) optimizer.step() if step % log_every == 0: print( f"[step {step:05d}] total={logs['total']:.4f} " f"recon={logs['recon']:.4f} dyn={logs['dyn']:.4f} " f"pred={logs['pred']:.4f} reward={logs['reward']:.4f} " f"done={logs['done']:.4f}" ) return model

关于超参，这里给几条经验：lr=3e-3对于小规模 MLP 是一个比较激进的起点，如果发现损失在震荡就降到1e-3；batch_size=256在小数据集上比较合适，数据量上到百万级可以提到 1024；clip_grad_norm_的阈值 5.0 是一个保守值，能挡住大部分梯度爆炸但不至于影响正常训练。

正常情况下，你应该看到pred和reward这两项稳步下降。如果pred一直不降，通常是编码器坍缩了（表现为dyn很小但pred很大），需要检查 detach 是否写对；如果reward一直不降，说明隐状态里没有包含足够的信息来预测奖励，可能需要加大latent_dim。

4.4 用 CEM 在世界模型里做规划

有了世界模型，接下来实现规划器。CEM 的思路是：维护一个动作序列的高斯分布，每次采样一批候选、评估、保留最优的一部分、用它们重新估计均值和方差，迭代几轮之后收敛到一个不错的解。

# 文件路径：world_model_demo/planner.py import numpy as np import torch class CEMPlanner: """在世界模型内部做交叉熵方法规划。""" def __init__(self, model, horizon=30, pop_size=256, elite_frac=0.1, iters=5, discount=0.97, act_low=-1.0, act_high=1.0, device="cpu"): self.model = model self.horizon = horizon self.pop_size = pop_size self.n_elite = max(1, int(pop_size * elite_frac)) self.iters = iters self.discount = discount self.act_low = act_low self.act_high = act_high self.device = device self.model.eval() @torch.no_grad() def plan(self, obs): """输入单个观测，返回 horizon 步的动作序列 (H, act_dim)。""" obs_t = torch.as_tensor(obs, dtype=torch.float32, device=self.device).unsqueeze(0) z0 = self.model.encode(obs_t) # (1, latent) z0 = z0.expand(self.pop_size, -1) # (pop, latent) mean = torch.zeros(self.horizon, self.model.act_dim, device=self.device) std = torch.ones(self.horizon, self.model.act_dim, device=self.device) best_seq = mean.clone() best_ret = torch.full((self.pop_size,), -1e9, device=self.device) for _ in range(self.iters): noise = torch.randn(self.pop_size, self.horizon, self.model.act_dim, device=self.device) cand = mean.unsqueeze(0) + std.unsqueeze(0) * noise cand = torch.clamp(cand, self.act_low, self.act_high) ret = self.model.imagine(z0, cand, discount=self.discount) # (pop,) elite_idx = ret.topk(self.n_elite).indices elite = cand[elite_idx] mean = elite.mean(dim=0) std = elite.std(dim=0) + 1e-4 best_seq = mean.clone() best_ret = ret return best_seq.cpu().numpy() class MPCController: """每隔 replan_every 步重新规划一次，中间复用剩余动作。""" def __init__(self, planner, replan_every=5): self.planner = planner self.replan_every = replan_every self._queue = [] self._counter = 0 def reset(self): self._queue = [] self._counter = 0 def act(self, obs): if not self._queue or self._counter >= self.replan_every: seq = self.planner.plan(obs) self._queue = list(seq) self._counter = 0 a = self._queue.pop(0) self._counter += 1 return np.asarray(a, dtype=np.float32)

这里imagine的输入z0是被广播成(pop_size, latent_dim)的，这样一次前向就能并行评估所有候选轨迹，不需要写 for 循环。这是向量化实现带来的效率提升，在pop_size=256时差别非常明显。

replan_every=5是一个折中。设成 1 就是标准的滚动时域控制，精度最高但算力消耗最大；设成更大值就变成了"开环执行一段时间再纠偏"，算力省了但适应性变差。实际项目里这个值取决于控制频率和计算预算。

4.5 主流程与效果验证

最后把整个流程串起来，并加一个随机策略的基线做对比。

# 文件路径：world_model_demo/main.py import numpy as np import torch from env import PointMassEnv from replay import collect_random_data from model import WorldModel from train_wm import train_world_model from planner import CEMPlanner, MPCController def evaluate(env, act_fn, episodes=20, seed=123, on_reset=None): """跑若干 episode，返回平均累计回报和平均步数。""" rng_state = np.random.get_state() np.random.seed(seed) returns, lengths = [], [] for _ in range(episodes): if on_reset is not None: on_reset() obs = env.reset() done = False total, steps = 0.0, 0 while not done: a = act_fn(obs) obs, r, done = env.step(a) total += r steps += 1 returns.append(total) lengths.append(steps) np.random.set_state(rng_state) return float(np.mean(returns)), float(np.mean(lengths)) def main(): torch.manual_seed(0) np.random.seed(0) device = "cuda" if torch.cuda.is_available() else "cpu" print(f"device = {device}") # ---------- 1. 采集数据 ---------- env = PointMassEnv(seed=0) data = collect_random_data(env, n_episodes=200, seed=0) print(f"collected transitions: {data[0].shape[0]}") # ---------- 2. 训练世界模型 ---------- model = WorldModel(obs_dim=env.obs_dim, act_dim=env.act_dim, latent_dim=32, hidden=128).to(device) model = train_world_model(model, data, steps=4000, batch_size=256, lr=3e-3, device=device) # ---------- 3. 随机策略基线 ---------- def random_policy(_obs): return np.random.uniform(-1.0, 1.0, size=(env.act_dim,)).astype(np.float32) rand_ret, rand_len = evaluate(env, random_policy, episodes=20, seed=123) print(f"[random policy] return={rand_ret:.2f} length={rand_len:.1f}") # ---------- 4. 世界模型规划策略 ---------- planner = CEMPlanner(model, horizon=30, pop_size=256, elite_frac=0.1, iters=5, device=device) controller = MPCController(planner, replan_every=5) wm_ret, wm_len = evaluate(env, controller.act, episodes=20, seed=123, on_reset=controller.reset) print(f"[world-model MPC] return={wm_ret:.2f} length={wm_len:.1f}") if __name__ == "__main__": main()

运行命令很简单：

# 文件路径：world_model_demo/ python main.py

预期的输出形式如下（具体数值会随随机种子和硬件略有波动）：

device = cpu collected transitions: 40000 [step 00500] total=... recon=... dyn=... pred=... reward=... done=... ... [step 04000] total=... recon=... dyn=... pred=... reward=... done=... [random policy] return=-268.43 length=200.0 [world-model MPC] return=-71.26 length=200.0

判断是否成功的标志是：规划策略的平均回报显著高于随机策略。在这个环境里，随机策略会让质点四处乱撞，累计回报通常在 -200 到 -400 之间；而一个学得还不错的世界模型配合规划，能把质点稳定在原点附近，回报会收敛到 -50 到 -100 这个量级。

如果两者差距不明显，先别急着改代码，按第 6 章的排查顺序走一遍，绝大多数问题出在数据分布和训练轮数上。

5. 三个真实场景的接入方式

上面是一个玩具环境，目的是把原理讲清楚。真实项目里的接入方式会复杂得多，但核心结构和流程是一致的。这一章分别讲三个典型场景的落地要点。

5.1 自动驾驶：离线数据 + 反事实推演

自动驾驶场景的最大特点是数据量大但信息密度低。一台测试车一天能采几百 GB 数据，但这些数据全都是"人类驾驶员怎么开"，不含任何反事实信息。世界模型在这里的核心价值，就是把这些被动数据变成可以主动查询的推演引擎。

典型的工程流程是这样的：

1. 数据预处理：把原始传感器数据（多路相机、激光雷达、车辆状态）对齐到统一的时间轴，切分成固定长度的片段，每个片段标注自车轨迹和周围目标的轨迹。
2. 世界模型训练：输入当前时刻的感知结果（鸟瞰图、目标框序列或隐特征），预测未来若干秒的场景演化。注意这里通常要预测两类东西：自车轨迹和其他交通参与者的轨迹。后者尤其重要，因为交互才是自动驾驶的难点。
3. 策略评估：给定一个候选策略，让它在一个历史片段的初始状态上做出不同于人类的动作，然后用世界模型推演后续几秒，判断是否会发生碰撞、是否违反交通规则。
4. 策略训练：把世界模型包装成一个可以reset()和step()的环境接口，直接喂给现有的强化学习算法。

这里有一个设计选择值得讨论：预测原始观测还是预测结构化表示。预测高清图像看起来很酷，但代价极高，而且图像细节对决策的贡献有限。更实用的做法是预测"结构化场景表示"——目标的位置、速度、朝向、类别，再加上一个可选的语义分割图。这样既保留了决策所需的关键信息，又把预测难度降了一个数量级。

另一个必须考虑的问题是开环与闭环。开环评估是"用一段固定历史数据，评估预测轨迹和真实轨迹的接近程度"，闭环评估是"让策略在世界模型里真正开起来"。两者的结论经常不一致：一个在世界模型里开得很稳的策略，可能在真实环境里表现很差，因为世界模型没有见过"策略偏离人类轨迹"之后的状态分布。这就是著名的分布偏移问题，第 6 章会给出缓解方案。

5.2 机器人操作：视觉世界模型与 sim-to-real gap

机器人操作和自动驾驶的差异很大。直观对比一下：自动驾驶的场景演化相对平滑，交通参与者遵循大致可预测的运动规律；机器人操作涉及接触、摩擦、形变，这些物理过程的建模难度高得多，而且往往发生在几厘米的空间尺度上，视觉上很难分辨。

在这种场景下，世界模型通常和仿真器配合使用：

1. 仿真器负责基础物理：刚体动力学、碰撞检测、关节约束这些有明确数学形式的部分，交给物理引擎处理，保证不出大错。
2. 世界模型负责"仿真器学不到的部分"：接触时的软体形变、不同材质表面的摩擦系数差异、光照变化引起的视觉外观差异。这些通常用一个视觉预测模型来捕捉。
3. 域随机化叠加：在仿真器里随机构造大量参数变体（质量、摩擦、光照、相机位姿），训练出对这些变化鲁棒的世界模型。

一个实用的技巧是分阶段训练。先在大规模仿真数据上预训练世界模型，让它学会基本的物理常识；再用少量真实机器人数据做微调，把仿真和现实之间的系统性偏差修正回来。这个流程比"纯真实数据训练"样本效率高得多，也比"纯仿真训练"的泛化性好得多。

还有一个容易忽略的点：机器人的状态空间往往是部分可观测的。相机只能看到物体的一面，机械臂自身关节角可以精确读取，但被抓取物体的内部状态（比如是否已经滑脱）只能靠视觉推断。所以机器人的世界模型通常需要带记忆结构，比如 GRU 或 Transformer，把历史帧的信息整合进隐状态。这一点是纯 MLP 的结构做不到的。

5.3 LLM 智能体：可回滚沙箱与效果预测

LLM 智能体的"环境"和前面两种完全不同。它的状态是一个工作区的快照（文件系统