diff --git a/ding/bonus/dreamer/config.py b/ding/bonus/dreamer/config.py
new file mode 100644
index 0000000000..8f6f477810
--- /dev/null
+++ b/ding/bonus/dreamer/config.py
@@ -0,0 +1,115 @@
+from dataclasses import dataclass
+from typing import Any, Dict, Optional, Sequence
+
+import torch
+
+
+@dataclass
+class DreamerConfig:
+    """Dreamer 全局配置：涵盖模型结构、优化器参数与算法开关。"""
+    # --- 基础环境参数 ---
+    obs_dim: int
+    action_dim: int
+    
+    # --- RSSM 核心维度 ---
+    embed_dim: int = 64        # 观测编码后的维度
+    deter_dim: int = 128       # 确定性状态 h_t 的维度 (GRU)
+    stoch_dim: int = 32        # 随机状态 z_t 的维度
+    stoch_classes: int = 32    # (V2/V3) 离散潜在变量的类别数
+    hidden_dim: int = 128      # MLP 隐藏层维度
+    
+    # --- 训练超参数 ---
+    model_lr: float = 3e-4     # 世界模型学习率
+    actor_lr: float = 8e-5     # 策略网络学习率
+    critic_lr: float = 8e-5    # 价值网络学习率
+    entropy_scale: float = 1e-3
+    grad_clip: float = 100.0   # 梯度裁剪阈值
+    use_obs_norm: bool = True
+    normalize_advantage: bool = True
+    
+    # --- RL 算法参数 ---
+    discount: float = 0.99     # 折扣因子 gamma
+    lambda_: float = 0.95      # Lambda-return 平滑系数
+    horizon: int = 15          # 想象视界长度 H
+    
+    # --- Loss 权重 ---
+    free_nats: float = 1.0     # KL 散度的 Free bits 阈值
+    kl_scale: float = 1.0      # KL Loss 权重
+    discount_scale: float = 10.0
+    
+    # --- V2 特性开关 ---
+    kl_balance: float = 0.8    # KL Balancing 权重 (0.8 给先验)
+    
+    # --- V3 特性开关 ---
+    use_symlog: bool = False   # 是否启用 Symlog 数值压缩
+    target_tau: float = 0.01   # Critic Target 软更新系数
+    
+    # --- V3 离散回归配置 ---
+    reward_bins: int = 0       # 奖励离散化的桶数量 (0表示使用标量回归)
+    reward_min: float = -10.0
+    reward_max: float = 10.0
+    value_bins: int = 0        # 价值离散化的桶数量
+    value_min: float = -20.0
+    value_max: float = 20.0
+
+
+@dataclass
+class TrainConfig:
+    """训练流程相关配置。"""
+    env_ids: Sequence[str] = ("CartPole-v1",)
+    agent_versions: Sequence[str] = ("v1",)
+    seeds: Sequence[int] = (42, 2024)
+    total_steps: int = 30_000
+    seed_steps: int = 2_000
+    train_every: int = 1
+    train_steps: int = 1
+    batch_size: int = 32
+    seq_len: int = 8
+    horizon: int = 15
+    replay_size: int = 100_000
+    log_every: int = 1000
+    eval_every: int = 2_000
+    eval_episodes: int = 5
+    workdir: str = "runs/dreamer"
+    workdir_time_format: str = "%Y%m%d_%H%M%S"
+    timestamp_workdir_if_exists: bool = True
+    device: str = "cuda" if torch.cuda.is_available() else "cpu"
+    env_kwargs: Optional[Dict[str, Any]] = None
+
+    # Exploration schedule (epsilon-greedy after seed steps)
+    exploration_epsilon_start: float = 0.10
+    exploration_epsilon_end: float = 0.00
+    exploration_decay_steps: int = 20_000
+
+    # Actor regularization / reward scaling
+    entropy_scale: float = 1e-3
+    model_lr: float = 3e-4
+    actor_lr: float = 3e-4
+    critic_lr: float = 3e-4
+
+    # Model configuration
+    embed_dim: int = 64
+    deter_dim: int = 128
+    stoch_dim: int = 32
+    stoch_classes: int = 32
+    hidden_dim: int = 128
+    use_obs_norm: bool = True
+    normalize_advantage: bool = True
+    discount: float = 0.99
+    lambda_: float = 0.95
+    free_nats: float = 1.0
+    kl_scale: float = 1.0
+    discount_scale: float = 10.0
+    kl_balance: float = 0.8
+    target_tau: float = 0.01
+
+    # 是否每次训练前清空 metrics 文件，避免重复运行时混入历史点
+    overwrite_metrics: bool = True
+
+    # Plotting
+    plot_path: str = "runs/dreamer/return_curve.png"
+
+    def __post_init__(self) -> None:
+        """填充默认环境参数。"""
+        if self.env_kwargs is None:
+            self.env_kwargs = {}
diff --git a/ding/bonus/dreamer/dreamer/__init__.py b/ding/bonus/dreamer/dreamer/__init__.py
new file mode 100644
index 0000000000..5b250305da
--- /dev/null
+++ b/ding/bonus/dreamer/dreamer/__init__.py
@@ -0,0 +1,5 @@
+"""Dreamer components for CartPole."""
+
+from .agent import DreamerV1Agent, DreamerV2Agent, DreamerV3Agent, make_agent
+
+__all__ = ["DreamerV1Agent", "DreamerV2Agent", "DreamerV3Agent", "make_agent"]
diff --git a/ding/bonus/dreamer/dreamer/agent.py b/ding/bonus/dreamer/dreamer/agent.py
new file mode 100644
index 0000000000..207a9002ae
--- /dev/null
+++ b/ding/bonus/dreamer/dreamer/agent.py
@@ -0,0 +1,520 @@
+from contextlib import contextmanager
+from dataclasses import dataclass
+from typing import Dict, Optional, Tuple
+
+import numpy as np
+import torch
+from torch import nn
+import torch.nn.functional as F
+
+from config import DreamerConfig
+from .models import Actor, Critic, Decoder, DiscountModel, Encoder, RewardModel
+from .rssm import RSSM, RSSMState
+
+
+def symlog(x: torch.Tensor) -> torch.Tensor:
+    """对称对数变换，压缩数值范围。"""
+    return torch.sign(x) * torch.log1p(torch.abs(x))
+
+
+def symexp(x: torch.Tensor) -> torch.Tensor:
+    """symlog 的逆变换。"""
+    return torch.sign(x) * (torch.exp(torch.abs(x)) - 1.0)
+
+
+def two_hot(x: torch.Tensor, bins: int, min_value: float, max_value: float) -> torch.Tensor:
+    """将标量转成 two-hot 分布表示。"""
+    x = torch.clamp(x, min=min_value, max=max_value)
+    scale = (bins - 1) / (max_value - min_value)
+    pos = (x - min_value) * scale
+    idx_low = torch.floor(pos).long()
+    idx_high = torch.clamp(idx_low + 1, max=bins - 1)
+    weight_high = (pos - idx_low.float()).clamp(0.0, 1.0)
+    weight_low = 1.0 - weight_high
+    shape = x.shape + (bins,)
+    out = torch.zeros(shape, device=x.device, dtype=x.dtype)
+    out.scatter_(-1, idx_low.unsqueeze(-1), weight_low.unsqueeze(-1))
+    out.scatter_add_(-1, idx_high.unsqueeze(-1), weight_high.unsqueeze(-1))
+    return out
+
+
+def logits_to_mean(logits: torch.Tensor, min_value: float, max_value: float) -> torch.Tensor:
+    """将分布 logits 转成对应的期望值。"""
+    probs = torch.softmax(logits, dim=-1)
+    bins = torch.linspace(min_value, max_value, logits.shape[-1], device=logits.device, dtype=logits.dtype)
+    return torch.sum(probs * bins, dim=-1)
+
+
+@contextmanager
+def freeze_module_params(*modules: Optional[nn.Module]):
+    """临时冻结模块参数，但保留对输入的梯度传播。"""
+    params = []
+    requires_grad = []
+    for module in modules:
+        if module is None:
+            continue
+        for param in module.parameters():
+            params.append(param)
+            requires_grad.append(param.requires_grad)
+            param.requires_grad_(False)
+    try:
+        yield
+    finally:
+        for param, flag in zip(params, requires_grad):
+            param.requires_grad_(flag)
+
+
+class RunningMeanStd:
+    """在线估计均值与方差，用于归一化。"""
+
+    def __init__(self, epsilon: float = 1e-4, shape=()):
+        """初始化统计量。"""
+        self.mean = np.zeros(shape, "float64")
+        self.var = np.ones(shape, "float64")
+        self.count = epsilon
+
+    def update(self, x: np.ndarray) -> None:
+        """用新样本更新均值与方差。"""
+        x = np.asarray(x, dtype="float64")
+        batch_mean = np.mean(x, axis=0)
+        batch_var = np.var(x, axis=0)
+        batch_count = x.shape[0]
+        self._update_from_moments(batch_mean, batch_var, batch_count)
+
+    def _update_from_moments(self, batch_mean, batch_var, batch_count) -> None:
+        """从批次统计量更新。"""
+        delta = batch_mean - self.mean
+        total_count = self.count + batch_count
+        new_mean = self.mean + delta * batch_count / total_count
+        m_a = self.var * self.count
+        m_b = batch_var * batch_count
+        m2 = m_a + m_b + delta**2 * self.count * batch_count / total_count
+        new_var = m2 / total_count
+        self.mean = new_mean
+        self.var = new_var
+        self.count = total_count
+
+
+@dataclass
+class TrainingMetrics:
+    """记录一次训练更新中的关键指标。"""
+
+    model_loss: float
+    obs_loss: float
+    reward_loss: float
+    discount_loss: float
+    kl_loss: float
+    actor_loss: float
+    critic_loss: float
+
+
+class WorldModel(nn.Module):
+    """世界模型：编码器 + RSSM + 重建/奖励/折扣预测。"""
+    def __init__(self, config: DreamerConfig,
+        kl_balance: Optional[float] = None, *,
+        discrete_latent: bool = False,
+    ):
+        """初始化世界模型的子模块与配置。"""
+        super().__init__()
+        self.encoder = Encoder(config.obs_dim, config.embed_dim, config.hidden_dim)
+        self.rssm = RSSM(
+            action_dim=config.action_dim,
+            embed_dim=config.embed_dim,
+            deter_dim=config.deter_dim,
+            stoch_dim=config.stoch_dim,
+            hidden_dim=config.hidden_dim,
+            discrete=discrete_latent,
+            stoch_classes=config.stoch_classes,
+        )
+        stoch_feat_dim = config.stoch_dim * config.stoch_classes if discrete_latent else config.stoch_dim
+        feat_dim = config.deter_dim + stoch_feat_dim
+        self.decoder = Decoder(feat_dim, config.obs_dim, config.hidden_dim)
+        reward_out_dim = config.reward_bins if config.reward_bins and config.reward_bins > 1 else 1
+        self.reward_model = RewardModel(feat_dim, config.hidden_dim, out_dim=reward_out_dim)
+        self.discount_model = DiscountModel(feat_dim, config.hidden_dim)
+        self.config = config
+        self.kl_balance = kl_balance
+        self.reward_bins = reward_out_dim
+        self.reward_min = config.reward_min
+        self.reward_max = config.reward_max
+        self.use_symlog = config.use_symlog
+
+    def loss(self, obs: torch.Tensor, actions: torch.Tensor, rewards: torch.Tensor, dones: torch.Tensor, init_state: RSSMState) -> Tuple[torch.Tensor, Dict[str, float], RSSMState]:
+        """计算世界模型损失，并返回后验状态序列。"""
+        embeds = self.encoder(obs)
+        priors, posts = self.rssm.observe(embeds, actions, init_state)
+        feat = self.rssm.get_feat(posts)
+        obs_pred = self.decoder(feat)
+        reward_pred = self.reward_model(feat)
+        discount_logits = self.discount_model(feat).squeeze(-1)
+
+        obs_loss = F.mse_loss(obs_pred, obs)
+        reward_loss = self.reward_loss(reward_pred, rewards)
+        discount_target = 1.0 - dones
+        discount_loss = F.binary_cross_entropy_with_logits(discount_logits, discount_target)
+
+        kl_post = self.rssm.kl_divergence(posts, priors)
+        if self.kl_balance is None:
+            kl = kl_post
+        else:
+            # DreamerV2/V3 的 balanced KL 使用同方向 KL，并分别停止 posterior / prior 的梯度。
+            post_sg = self.rssm.detach_state(posts)
+            prior_sg = self.rssm.detach_state(priors)
+            kl_prior = self.rssm.kl_divergence(post_sg, priors)
+            kl_post = self.rssm.kl_divergence(posts, prior_sg)
+            kl = self.kl_balance * kl_prior + (1.0 - self.kl_balance) * kl_post
+        kl_loss = torch.mean(torch.clamp(kl, min=self.config.free_nats))
+
+        model_loss = obs_loss + reward_loss + self.config.discount_scale * discount_loss + self.config.kl_scale * kl_loss
+        metrics = {
+            "model_loss": model_loss.item(),
+            "obs_loss": obs_loss.item(),
+            "reward_loss": reward_loss.item(),
+            "discount_loss": discount_loss.item(),
+            "kl_loss": kl_loss.item(),
+        }
+        return model_loss, metrics, posts
+
+    def reward_loss(self, reward_pred: torch.Tensor, rewards: torch.Tensor) -> torch.Tensor:
+        """奖励预测损失（可选 two-hot + symlog）。"""
+        if self.reward_bins == 1:
+            return F.mse_loss(reward_pred.squeeze(-1), rewards)
+        target = symlog(rewards) if self.use_symlog else rewards
+        target = two_hot(target, self.reward_bins, self.reward_min, self.reward_max)
+        log_probs = F.log_softmax(reward_pred, dim=-1)
+        return -(target * log_probs).sum(-1).mean()
+
+    def predict_reward(self, feat: torch.Tensor) -> torch.Tensor:
+        """将奖励头输出转为标量奖励。"""
+        pred = self.reward_model(feat)
+        if self.reward_bins == 1:
+            return pred.squeeze(-1)
+        mean = logits_to_mean(pred, self.reward_min, self.reward_max)
+        return symexp(mean) if self.use_symlog else mean
+
+
+def lambda_return(rewards: torch.Tensor, values: torch.Tensor, discounts: torch.Tensor, bootstrap: torch.Tensor, lambda_: float) -> torch.Tensor:
+    """计算 lambda-return。"""
+    horizon = rewards.shape[0]
+    returns = []
+    last = bootstrap
+    for t in reversed(range(horizon)):
+        next_value = values[t + 1] if t + 1 < horizon else bootstrap
+        last = rewards[t] + discounts[t] * ((1 - lambda_) * next_value + lambda_ * last)
+        returns.append(last)
+    returns.reverse()
+    return torch.stack(returns, dim=0)
+
+
+def discount_cumprod(discounts: torch.Tensor) -> torch.Tensor:
+    """计算折扣累计乘积，用于重要性权重。"""
+    horizon = discounts.shape[0]
+    weights = []
+    acc = torch.ones_like(discounts[0])
+    for t in range(horizon):
+        weights.append(acc)
+        acc = acc * discounts[t]
+    return torch.stack(weights, dim=0)
+
+
+class DreamerV1Agent:
+    """Simplified DreamerV1 baseline."""
+
+    def __init__(self, config: DreamerConfig, device: torch.device):
+        """初始化 DreamerV1 的模型与优化器。"""
+        self.config = config
+        self.device = device
+        self.world_model = self._build_world_model().to(device)
+        stoch_feat_dim = config.stoch_dim * config.stoch_classes if self._discrete_latent() else config.stoch_dim
+        feat_dim = config.deter_dim + stoch_feat_dim
+        self.actor = Actor(feat_dim, config.action_dim, config.hidden_dim).to(device)
+        self.critic = Critic(feat_dim, config.hidden_dim, out_dim=self._critic_out_dim()).to(device)
+        self.critic_target = self._build_target_critic(feat_dim, self._critic_out_dim())
+
+        self.model_opt = torch.optim.Adam(self.world_model.parameters(), lr=config.model_lr)
+        self.actor_opt = torch.optim.Adam(self.actor.parameters(), lr=config.actor_lr)
+        self.critic_opt = torch.optim.Adam(self.critic.parameters(), lr=config.critic_lr)
+
+    def init_state(self, batch_size: int) -> RSSMState:
+        """初始化 RSSM 隐状态。"""
+        return self.world_model.rssm.init_state(batch_size, self.device)
+
+    def observe(self, obs: np.ndarray, state: RSSMState, prev_action: torch.Tensor) -> RSSMState:
+        """使用观测更新后验状态。"""
+        obs_t = torch.as_tensor(obs, device=self.device, dtype=torch.float32).unsqueeze(0)
+        with torch.no_grad():
+            embed = self.world_model.encoder(obs_t)
+            prior = self.world_model.rssm.img_step(state, prev_action)
+            post = self.world_model.rssm.obs_step(prior.deter, embed)
+        return post
+
+    def policy(self, state: RSSMState, eval_mode: bool = False) -> int:
+        """根据当前状态采样/选择动作。"""
+        feat = self.world_model.rssm.get_feat(state)
+        dist = self.actor(feat)
+        if eval_mode:
+            action = torch.argmax(dist.probs, dim=-1)
+        else:
+            action = dist.sample()
+        return int(action.item())
+
+    def action_to_onehot(self, action: int) -> torch.Tensor:
+        """将离散动作转换为 one-hot。"""
+        return F.one_hot(torch.tensor([action], device=self.device), num_classes=self.config.action_dim).float()
+
+    def train_step(self, batch: Dict[str, np.ndarray]) -> TrainingMetrics:
+        """执行一次训练更新（世界模型 + actor/critic）。"""
+        obs = torch.as_tensor(batch["obs"], device=self.device, dtype=torch.float32)
+        actions = torch.as_tensor(batch["actions"], device=self.device, dtype=torch.int64)
+        rewards = torch.as_tensor(batch["rewards"], device=self.device, dtype=torch.float32)
+        dones = torch.as_tensor(batch["dones"], device=self.device, dtype=torch.float32)
+
+        actions_oh = F.one_hot(actions, num_classes=self.config.action_dim).float()
+
+        init_state = self.init_state(obs.shape[0])
+        model_loss, model_metrics, posts = self.world_model.loss(obs, actions_oh, rewards, dones, init_state)
+
+        self.model_opt.zero_grad()
+        model_loss.backward()
+        torch.nn.utils.clip_grad_norm_(self.world_model.parameters(), self.config.grad_clip)
+        self.model_opt.step()
+
+        actor_loss, critic_loss = self._actor_critic_update(posts, dones)
+
+        return TrainingMetrics(
+            model_loss=model_metrics["model_loss"],
+            obs_loss=model_metrics["obs_loss"],
+            reward_loss=model_metrics["reward_loss"],
+            discount_loss=model_metrics["discount_loss"],
+            kl_loss=model_metrics["kl_loss"],
+            actor_loss=actor_loss,
+            critic_loss=critic_loss,
+        )
+
+    def _actor_critic_update(self, posts: RSSMState, dones: torch.Tensor) -> Tuple[float, float]:
+        """在想象轨迹上更新策略与价值网络。"""
+        start = self.world_model.rssm.detach_state(posts)
+        if start.deter.shape[1] <= 1:
+            return 0.0, 0.0
+        # chunk 的第一个 posterior 缺少前文上下文，跳过它能减轻 RSSM 冷启动偏差。
+        start = RSSMState(
+            deter=start.deter[:, 1:],
+            stoch=start.stoch[:, 1:],
+            stats=start.stats[:, 1:],
+        )
+        dones = dones[:, 1:]
+        # flatten batch and time
+        b, t, _ = start.deter.shape
+        nonterminal = (dones.reshape(b * t) < 0.5)
+        if not torch.any(nonterminal):
+            return 0.0, 0.0
+        start = RSSMState(
+            deter=start.deter.reshape(b * t, -1)[nonterminal],
+            stoch=start.stoch.reshape(b * t, -1)[nonterminal],
+            stats=start.stats.reshape(b * t, *start.stats.shape[2:])[nonterminal],
+        )
+
+        with freeze_module_params(self.world_model, self.critic, self.critic_target):
+            feats, entropies, rewards, discounts, last_state = self._imagine(start)
+            values_raw = self._critic_raw(feats)
+            values = self._critic_value(values_raw)
+            bootstrap_raw = self._critic_raw(self.world_model.rssm.get_feat(last_state))
+            bootstrap = self._critic_value(bootstrap_raw)
+            returns = lambda_return(rewards, values, discounts, bootstrap, self.config.lambda_)
+            weights = discount_cumprod(discounts).detach()
+            actor_loss = -(weights * returns).mean() - self.config.entropy_scale * (weights * entropies).mean()
+
+        self.actor_opt.zero_grad()
+        actor_loss.backward()
+        torch.nn.utils.clip_grad_norm_(self.actor.parameters(), self.config.grad_clip)
+        self.actor_opt.step()
+
+        feats_detached = feats.detach()
+        with torch.no_grad():
+            target_raw = self._target_critic_raw(feats_detached)
+            target_values = self._critic_value(target_raw)
+            last_feat = self.world_model.rssm.get_feat(last_state).detach()
+            bootstrap_raw = self._target_critic_raw(last_feat)
+            bootstrap = self._critic_value(bootstrap_raw)
+            returns = lambda_return(rewards.detach(), target_values, discounts.detach(), bootstrap, self.config.lambda_)
+            weights = discount_cumprod(discounts.detach())
+
+        values_raw = self._critic_raw(feats_detached)
+        critic_loss = self._critic_loss(values_raw, returns, weights)
+        self.critic_opt.zero_grad()
+        critic_loss.backward()
+        torch.nn.utils.clip_grad_norm_(self.critic.parameters(), self.config.grad_clip)
+        self.critic_opt.step()
+        self._update_target()
+
+        return actor_loss.item(), critic_loss.item()
+
+    def _imagine(self, start: RSSMState):
+        """在世界模型中进行想象 rollout。"""
+        feats = []
+        entropies = []
+        rewards = []
+        discounts = []
+        state = start
+        for _ in range(self.config.horizon):
+            feat = self.world_model.rssm.get_feat(state)
+            dist = self.actor(feat)
+            action = dist.sample()
+            entropy = dist.entropy()
+            action_oh = F.one_hot(action, num_classes=self.config.action_dim).float()
+            action_oh = action_oh + dist.probs - dist.probs.detach()
+
+            state = self.world_model.rssm.img_step(state, action_oh)
+            next_feat = self.world_model.rssm.get_feat(state)
+            reward = self.world_model.predict_reward(next_feat)
+            discount = torch.sigmoid(self.world_model.discount_model(next_feat).squeeze(-1)) * self.config.discount
+
+            feats.append(feat)
+            entropies.append(entropy)
+            rewards.append(reward)
+            discounts.append(discount)
+
+        feats = torch.stack(feats, dim=0)
+        entropies = torch.stack(entropies, dim=0)
+        rewards = torch.stack(rewards, dim=0)
+        discounts = torch.stack(discounts, dim=0)
+
+        return feats, entropies, rewards, discounts, state
+
+    def _update_target(self) -> None:
+        """Polyak 平滑更新目标 critic。"""
+        if self.critic_target is None:
+            return
+        tau = self.config.target_tau
+        if tau <= 0.0:
+            return
+        with torch.no_grad():
+            for p, tp in zip(self.critic.parameters(), self.critic_target.parameters()):
+                tp.data.mul_(1.0 - tau)
+                tp.data.add_(tau * p.data)
+
+    def _build_world_model(self) -> WorldModel:
+        """构建世界模型（V1 为连续潜在）。"""
+        return WorldModel(self.config, kl_balance=None, discrete_latent=self._discrete_latent())
+
+    def _build_target_critic(self, feat_dim: int, out_dim: int):
+        """构建目标 critic，用于稳定 bootstrap。"""
+        critic_target = Critic(feat_dim, self.config.hidden_dim, out_dim=out_dim).to(self.device)
+        critic_target.load_state_dict(self.critic.state_dict())
+        return critic_target
+
+    def _critic_out_dim(self) -> int:
+        """critic 输出维度（V1 为标量）。"""
+        return 1
+
+    def _critic_raw(self, feats: torch.Tensor) -> torch.Tensor:
+        """critic 原始输出。"""
+        return self.critic(feats)
+
+    def _critic_value(self, raw: torch.Tensor) -> torch.Tensor:
+        """将 critic 输出转换为标量价值。"""
+        return raw.squeeze(-1)
+
+    def _target_critic_raw(self, feats: torch.Tensor) -> torch.Tensor:
+        """目标 critic 的原始输出。"""
+        if self.critic_target is None:
+            return self._critic_raw(feats)
+        return self.critic_target(feats)
+
+    def _compute_advantages(self, returns: torch.Tensor, values: torch.Tensor) -> torch.Tensor:
+        """计算优势函数。"""
+        return returns - values
+
+    def _critic_loss(self, raw_values: torch.Tensor, returns: torch.Tensor, weights: torch.Tensor) -> torch.Tensor:
+        """critic 的回归损失。"""
+        values = self._critic_value(raw_values)
+        return ((values - returns.detach()) ** 2 * weights).mean()
+
+    def _discrete_latent(self) -> bool:
+        """是否使用离散潜在（V1 为 False）。"""
+        return False
+    
+    
+class DreamerV2Agent(DreamerV1Agent):
+    """Simplified DreamerV2: KL balancing."""
+
+    def _build_world_model(self) -> WorldModel:
+        """V2 使用 KL 平衡的世界模型。"""
+        return WorldModel(self.config, kl_balance=self.config.kl_balance, discrete_latent=self._discrete_latent())
+
+    def _discrete_latent(self) -> bool:
+        """V2 使用离散潜在表示。"""
+        return True
+
+
+class DreamerV3Agent(DreamerV2Agent):
+    """Simplified DreamerV3: symlog + two-hot value/reward + target critic."""
+
+    def __init__(self, config: DreamerConfig, device: torch.device):
+        """启用 symlog、two-hot 回归与目标 critic。"""
+        if config.reward_bins <= 1:
+            config.reward_bins = 255
+        if config.value_bins <= 1:
+            config.value_bins = 255
+        config.use_symlog = True
+        super().__init__(config, device)
+        self.return_rms = RunningMeanStd(shape=())
+
+    def _critic_out_dim(self) -> int:
+        """V3 critic 输出为分布 logits。"""
+        return self.config.value_bins
+
+    def _build_world_model(self) -> WorldModel:
+        """V3 仍使用离散潜在 + KL 平衡。"""
+        return WorldModel(self.config, kl_balance=self.config.kl_balance, discrete_latent=self._discrete_latent())
+
+    def _build_target_critic(self, feat_dim: int, out_dim: int):
+        """构建目标 critic，用于稳定训练。"""
+        critic_target = Critic(feat_dim, self.config.hidden_dim, out_dim=out_dim).to(self.device)
+        critic_target.load_state_dict(self.critic.state_dict())
+        return critic_target
+
+    def _critic_value(self, raw: torch.Tensor) -> torch.Tensor:
+        """将 two-hot logits 转为标量价值。"""
+        symlog_value = logits_to_mean(raw, self.config.value_min, self.config.value_max)
+        return symexp(symlog_value)
+
+    def _target_critic_raw(self, feats: torch.Tensor) -> torch.Tensor:
+        """目标 critic 原始输出。"""
+        return self.critic_target(feats)
+
+    def _compute_advantages(self, returns: torch.Tensor, values: torch.Tensor) -> torch.Tensor:
+        """对回报做归一化后计算优势。"""
+        if not self.config.normalize_advantage:
+            return returns - values
+        returns_np = returns.detach().cpu().numpy().reshape(-1)
+        self.return_rms.update(returns_np)
+        mean = torch.tensor(self.return_rms.mean, device=returns.device, dtype=returns.dtype)
+        std = torch.tensor(np.sqrt(self.return_rms.var) + 1e-8, device=returns.device, dtype=returns.dtype)
+        returns_norm = (returns - mean) / std
+        values_norm = (values - mean) / std
+        return returns_norm - values_norm
+
+    def _critic_loss(self, raw_values: torch.Tensor, returns: torch.Tensor, weights: torch.Tensor) -> torch.Tensor:
+        """V3 的 two-hot 价值回归损失。"""
+        target = symlog(returns)
+        target = two_hot(target, self.config.value_bins, self.config.value_min, self.config.value_max)
+        log_probs = F.log_softmax(raw_values, dim=-1)
+        loss = -(target * log_probs).sum(-1)
+        return (loss * weights).mean()
+
+    def _update_target(self) -> None:
+        """Polyak 平滑更新目标 critic。"""
+        super()._update_target()
+
+
+def make_agent(version: str, config: DreamerConfig, device: torch.device):
+    """根据版本字符串构造对应的 Agent。"""
+    version = version.lower()
+    if version == "v1":
+        return DreamerV1Agent(config, device)
+    if version == "v2":
+        return DreamerV2Agent(config, device)
+    if version == "v3":
+        return DreamerV3Agent(config, device)
+    raise ValueError(f"Unknown agent version: {version}")
diff --git a/ding/bonus/dreamer/dreamer/models.py b/ding/bonus/dreamer/dreamer/models.py
new file mode 100644
index 0000000000..4dee0f7ddc
--- /dev/null
+++ b/ding/bonus/dreamer/dreamer/models.py
@@ -0,0 +1,99 @@
+from typing import Iterable, Optional
+
+import torch
+from torch import nn
+from torch.distributions import Categorical
+
+
+class MLP(nn.Module):
+    """多层感知机模块。"""
+    def __init__(self, in_dim: int, hidden_dims: Iterable[int], out_dim: int, act=nn.ELU, out_act: Optional[nn.Module] = None):
+        """构建 MLP。"""
+        super().__init__()
+        dims = [in_dim] + list(hidden_dims)
+        layers = []
+        for i in range(len(dims) - 1):
+            layers.append(nn.Linear(dims[i], dims[i + 1]))
+            layers.append(act())
+        layers.append(nn.Linear(dims[-1], out_dim))
+        if out_act is not None:
+            layers.append(out_act)
+        self.net = nn.Sequential(*layers)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        """前向传播。"""
+        return self.net(x)
+
+
+class Encoder(nn.Module):
+    """观测编码器。"""
+    def __init__(self, obs_dim: int, embed_dim: int, hidden: int):
+        """构建编码器。"""
+        super().__init__()
+        self.net = MLP(obs_dim, [hidden, hidden], embed_dim)
+
+    def forward(self, obs: torch.Tensor) -> torch.Tensor:
+        """将观测编码为嵌入向量。"""
+        return self.net(obs)
+
+
+class Decoder(nn.Module):
+    """观测解码器。"""
+    def __init__(self, feat_dim: int, obs_dim: int, hidden: int):
+        """构建解码器。"""
+        super().__init__()
+        self.net = MLP(feat_dim, [hidden, hidden], obs_dim)
+
+    def forward(self, feat: torch.Tensor) -> torch.Tensor:
+        """从特征重建观测。"""
+        return self.net(feat)
+
+
+class RewardModel(nn.Module):
+    """奖励预测模型。"""
+    def __init__(self, feat_dim: int, hidden: int, out_dim: int = 1):
+        """构建奖励模型。"""
+        super().__init__()
+        self.net = MLP(feat_dim, [hidden, hidden], out_dim)
+
+    def forward(self, feat: torch.Tensor) -> torch.Tensor:
+        """预测奖励（或分布 logits）。"""
+        return self.net(feat)
+
+
+class DiscountModel(nn.Module):
+    """折扣（终止）预测模型。"""
+    def __init__(self, feat_dim: int, hidden: int):
+        """构建折扣模型。"""
+        super().__init__()
+        self.net = MLP(feat_dim, [hidden, hidden], 1)
+
+    def forward(self, feat: torch.Tensor) -> torch.Tensor:
+        """预测折扣 logits。"""
+        return self.net(feat)
+
+
+class Actor(nn.Module):
+    """策略网络，输出动作分布。"""
+    def __init__(self, feat_dim: int, num_actions: int, hidden: int):
+        """构建策略网络。"""
+        super().__init__()
+        self.net = MLP(feat_dim, [hidden, hidden], num_actions)
+        self.num_actions = num_actions
+
+    def forward(self, feat: torch.Tensor) -> Categorical:
+        """输出动作分布。"""
+        logits = self.net(feat)
+        return Categorical(logits=logits)
+
+
+class Critic(nn.Module):
+    """价值网络。"""
+    def __init__(self, feat_dim: int, hidden: int, out_dim: int = 1):
+        """构建价值网络。"""
+        super().__init__()
+        self.net = MLP(feat_dim, [hidden, hidden], out_dim)
+
+    def forward(self, feat: torch.Tensor) -> torch.Tensor:
+        """输出价值（或分布 logits）。"""
+        return self.net(feat)
diff --git a/ding/bonus/dreamer/dreamer/replay.py b/ding/bonus/dreamer/dreamer/replay.py
new file mode 100644
index 0000000000..a91ed5bc7c
--- /dev/null
+++ b/ding/bonus/dreamer/dreamer/replay.py
@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import random
+from typing import Dict, List
+
+import numpy as np
+
+class ReplayBuffer:
+    """按 Episode 存储，支持采样固定长度的时间序列片段。"""
+    def __init__(self, capacity_steps: int):
+        self.capacity = capacity_steps
+        self.episodes = [] # 存储完整轨迹的列表
+        self.total_steps = 0
+
+    def __len__(self) -> int:
+        """返回回放中累计的环境步数。"""
+        return self.total_steps
+
+    def add_episode(self, obs, actions, rewards, dones):
+        """存入一条完整的 Episode 轨迹。"""
+        if not (len(obs) == len(actions) == len(rewards) == len(dones)):
+            raise ValueError("Episode fields must have the same length.")
+        episode = {
+            "obs": np.array(obs, dtype=np.float32),
+            "actions": np.array(actions, dtype=np.int64),
+            "rewards": np.array(rewards, dtype=np.float32),
+            "dones": np.array(dones, dtype=np.float32),
+        }
+        self.episodes.append(episode)
+        self.total_steps += len(obs)
+        # 移除旧数据以维持容量限制
+        while self.total_steps > self.capacity:
+            removed = self.episodes.pop(0)
+            self.total_steps -= len(removed["obs"])
+
+    def sample(self, batch_size: int, seq_len: int):
+        """核心功能：采样 [B, T, ...] 格式的序列片段。"""
+        # 1. 筛选出长度足够的轨迹
+        candidates = [ep for ep in self.episodes if len(ep["obs"]) >= seq_len]
+        if not candidates:
+            raise ValueError("Not enough data to sample sequences.")
+        
+        batch = {"obs": [], "actions": [], "rewards": [], "dones": []}
+        for _ in range(batch_size):
+            ep = random.choice(candidates)
+            # 2. 在轨迹中随机选择起始点
+            start = random.randint(0, len(ep["obs"]) - seq_len)
+            end = start + seq_len
+            
+            # 3. 切片提取
+            for k in batch:
+                batch[k].append(ep[k][start:end])
+        
+        # 4. 堆叠为 Tensor: [Batch, Time, Dim]
+        return {k: np.stack(v) for k, v in batch.items()}
diff --git a/ding/bonus/dreamer/dreamer/rssm.py b/ding/bonus/dreamer/dreamer/rssm.py
new file mode 100644
index 0000000000..c8a6b8915c
--- /dev/null
+++ b/ding/bonus/dreamer/dreamer/rssm.py
@@ -0,0 +1,151 @@
+from dataclasses import dataclass
+from typing import List
+
+import torch
+from torch import nn
+from torch.distributions import Normal, OneHotCategorical
+import torch.nn.functional as F
+
+
+@dataclass
+class RSSMState:
+    """RSSM 隐状态容器：同时包含确定性部分(h)与随机部分(z)。"""
+    deter: torch.Tensor       # h_t (GRU state)
+    stoch: torch.Tensor       # z_t (Sampled latent)
+    stats: torch.Tensor       # Mean/Std (V1) 或 Logits (V2/V3)
+
+class RSSM(nn.Module):
+    """Recurrent State Space Model，支持连续/离散潜在。"""
+    def __init__(self, action_dim: int, embed_dim: int, deter_dim: int, 
+    stoch_dim: int, hidden_dim: int, min_std: float = 0.1,
+    discrete: bool = False, stoch_classes: int = 0,
+    ):
+        """初始化 RSSM 结构与参数。"""
+        super().__init__()
+        self.action_dim = action_dim
+        self.embed_dim = embed_dim
+        self.deter_dim = deter_dim
+        self.stoch_dim = stoch_dim
+        self.hidden_dim = hidden_dim
+        self.min_std = min_std
+        self.discrete = discrete
+        self.stoch_classes = stoch_classes
+        if self.discrete and self.stoch_classes <= 1:
+            raise ValueError("Discrete RSSM requires stoch_classes > 1.")
+
+        stoch_input = stoch_dim * stoch_classes if self.discrete else stoch_dim
+        self.input_layer = nn.Linear(stoch_input + action_dim, hidden_dim)
+        self.rnn = nn.GRUCell(hidden_dim, deter_dim)
+        if self.discrete:
+            self.prior_layer = nn.Linear(deter_dim, stoch_dim * stoch_classes)
+            self.post_layer = nn.Linear(deter_dim + embed_dim, stoch_dim * stoch_classes)
+        else:
+            self.prior_layer = nn.Linear(deter_dim, 2 * stoch_dim)
+            self.post_layer = nn.Linear(deter_dim + embed_dim, 2 * stoch_dim)
+
+    def init_state(self, batch_size: int, device: torch.device) -> RSSMState:
+        """创建初始隐状态。"""
+        deter = torch.zeros(batch_size, self.deter_dim, device=device)
+        if self.discrete:
+            stoch = torch.zeros(batch_size, self.stoch_dim * self.stoch_classes, device=device)
+            stats = torch.zeros(batch_size, self.stoch_dim, self.stoch_classes, device=device)
+        else:
+            stoch = torch.zeros(batch_size, self.stoch_dim, device=device)
+            mean = torch.zeros(batch_size, self.stoch_dim, device=device)
+            std = torch.ones(batch_size, self.stoch_dim, device=device)
+            stats = torch.cat([mean, std], dim=-1)
+        return RSSMState(deter=deter, stoch=stoch, stats=stats)
+
+    def get_feat(self, state: RSSMState) -> torch.Tensor:
+        """拼接确定性与随机性状态作为特征。"""
+        return torch.cat([state.deter, state.stoch], dim=-1)
+
+    def _stats(self, x: torch.Tensor) -> (torch.Tensor, torch.Tensor):
+        """将线性输出拆分为 mean/std。"""
+        mean, std = torch.chunk(x, 2, dim=-1)
+        std = F.softplus(std) + self.min_std
+        return mean, std
+
+    def _sample(self, mean: torch.Tensor, std: torch.Tensor) -> torch.Tensor:
+        """连续潜在采样。"""
+        return mean + std * torch.randn_like(mean)
+
+    def _sample_discrete(self, logits: torch.Tensor) -> torch.Tensor:
+        """离散潜在采样（straight-through）。"""
+        dist = OneHotCategorical(logits=logits)
+        sample = dist.sample()
+        probs = dist.probs
+        return sample + probs - probs.detach()
+
+    def img_step(self, prev_state: RSSMState, action: torch.Tensor) -> RSSMState:
+        """先验步：仅依赖前一状态与动作。"""
+        x = torch.cat([prev_state.stoch, action], dim=-1)
+        x = F.elu(self.input_layer(x))
+        deter = self.rnn(x, prev_state.deter)
+        if self.discrete:
+            logits = self.prior_layer(deter).view(-1, self.stoch_dim, self.stoch_classes)
+            stoch = self._sample_discrete(logits).reshape(-1, self.stoch_dim * self.stoch_classes)
+            return RSSMState(deter=deter, stoch=stoch, stats=logits)
+        mean, std = self._stats(self.prior_layer(deter))
+        stoch = self._sample(mean, std)
+        stats = torch.cat([mean, std], dim=-1)
+        return RSSMState(deter=deter, stoch=stoch, stats=stats)
+
+    def obs_step(self, deter: torch.Tensor, embed: torch.Tensor) -> RSSMState:
+        """后验步：融合观测嵌入。"""
+        x = torch.cat([deter, embed], dim=-1)
+        if self.discrete:
+            logits = self.post_layer(x).view(-1, self.stoch_dim, self.stoch_classes)
+            stoch = self._sample_discrete(logits).reshape(-1, self.stoch_dim * self.stoch_classes)
+            return RSSMState(deter=deter, stoch=stoch, stats=logits)
+        mean, std = self._stats(self.post_layer(x))
+        stoch = self._sample(mean, std)
+        stats = torch.cat([mean, std], dim=-1)
+        return RSSMState(deter=deter, stoch=stoch, stats=stats)
+
+    def observe(self, embeds: torch.Tensor, actions: torch.Tensor, init_state: RSSMState) -> (RSSMState, RSSMState):
+        """按时间序列滚动，返回 prior 与 posterior 序列。"""
+        priors: List[RSSMState] = []
+        posts: List[RSSMState] = []
+        state = init_state
+        time_steps = embeds.shape[1]
+        for t in range(time_steps):
+            prior = self.img_step(state, actions[:, t])
+            post = self.obs_step(prior.deter, embeds[:, t])
+            state = post
+            priors.append(prior)
+            posts.append(post)
+        return stack_states(priors), stack_states(posts)
+
+    def detach_state(self, state: RSSMState) -> RSSMState:
+        """从计算图中分离隐状态。"""
+        return RSSMState(
+            deter=state.deter.detach(),
+            stoch=state.stoch.detach(),
+            stats=state.stats.detach(),
+        )
+
+    def kl_divergence(self, post: RSSMState, prior: RSSMState) -> torch.Tensor:
+        """计算 posterior 与 prior 的 KL。"""
+        if self.discrete:
+            post_logits = post.stats
+            prior_logits = prior.stats
+            post_dist = OneHotCategorical(logits=post_logits)
+            prior_dist = OneHotCategorical(logits=prior_logits)
+            kl = torch.distributions.kl.kl_divergence(post_dist, prior_dist)
+            return torch.sum(kl, dim=-1)
+        post_mean, post_std = torch.chunk(post.stats, 2, dim=-1)
+        prior_mean, prior_std = torch.chunk(prior.stats, 2, dim=-1)
+        post_dist = Normal(post_mean, post_std)
+        prior_dist = Normal(prior_mean, prior_std)
+        kl = torch.distributions.kl.kl_divergence(post_dist, prior_dist)
+        return torch.sum(kl, dim=-1)
+
+
+def stack_states(states: List[RSSMState], dim: int = 1) -> RSSMState:
+    """按时间维堆叠状态列表。"""
+    return RSSMState(
+        deter=torch.stack([s.deter for s in states], dim=dim),
+        stoch=torch.stack([s.stoch for s in states], dim=dim),
+        stats=torch.stack([s.stats for s in states], dim=dim),
+    )
diff --git a/ding/bonus/dreamer/plot_returns.py b/ding/bonus/dreamer/plot_returns.py
new file mode 100644
index 0000000000..20f9239dbf
--- /dev/null
+++ b/ding/bonus/dreamer/plot_returns.py
@@ -0,0 +1,119 @@
+import csv
+import os
+from collections import defaultdict
+from typing import Dict, List, Optional, Sequence, Tuple
+
+import matplotlib.pyplot as plt
+import numpy as np
+
+from config import TrainConfig
+
+
+def load_csv(path: str) -> Tuple[np.ndarray, np.ndarray]:
+    """读取评估 CSV 文件。"""
+    steps: List[int] = []
+    returns: List[float] = []
+    with open(path, "r", newline="") as f:
+        reader = csv.DictReader(f)
+        for row in reader:
+            steps.append(int(row["env_steps"]))
+            returns.append(float(row["avg_return"]))
+    return np.asarray(steps, dtype=np.float32), np.asarray(returns, dtype=np.float32)
+
+
+def collect_runs(workdir: str) -> Dict[Tuple[str, str], List[Tuple[np.ndarray, np.ndarray]]]:
+    """收集所有实验的评估曲线。"""
+    runs: Dict[Tuple[str, str], List[Tuple[np.ndarray, np.ndarray]]] = defaultdict(list)
+    for root, _, files in os.walk(workdir):
+        if "eval_returns.csv" not in files:
+            continue
+        csv_path = os.path.join(root, "eval_returns.csv")
+        rel = os.path.relpath(root, workdir)
+        parts = rel.split(os.sep)
+        if len(parts) < 3:
+            continue
+        env_id, version = parts[0], parts[1]
+        steps, returns = load_csv(csv_path)
+        if len(steps) == 0:
+            continue
+        runs[(env_id, version)].append((steps, returns))
+    return runs
+
+
+def aggregate_group(series: List[Tuple[np.ndarray, np.ndarray]]) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
+    """对同一组实验按步数聚合均值与方差。"""
+    step_to_values: Dict[int, List[float]] = defaultdict(list)
+    for steps, returns in series:
+        for s, r in zip(steps, returns):
+            step_to_values[int(s)].append(float(r))
+    all_steps = np.array(sorted(step_to_values.keys()), dtype=np.float32)
+    means = []
+    stds = []
+    for s in all_steps:
+        vals = np.asarray(step_to_values[int(s)], dtype=np.float32)
+        means.append(np.mean(vals))
+        stds.append(np.std(vals))
+    return all_steps, np.asarray(means), np.asarray(stds)
+
+
+def env_plot_path(base_path: str, env_id: str) -> str:
+    """根据环境名生成单独图片路径。"""
+    root, ext = os.path.splitext(base_path)
+    safe_env = env_id.replace("/", "_")
+    if not ext:
+        ext = ".png"
+    return f"{root}_{safe_env}{ext}"
+
+
+def plot_returns(
+    workdir: Optional[str] = None,
+    plot_path: Optional[str] = None,
+    agent_versions: Optional[Sequence[str]] = None,
+) -> None:
+    """按环境分别绘制学习曲线。"""
+    cfg = TrainConfig()
+    resolved_workdir = workdir if workdir is not None else cfg.workdir
+    resolved_plot_path = plot_path if plot_path is not None else cfg.plot_path
+    resolved_versions = list(agent_versions) if agent_versions is not None else list(cfg.agent_versions)
+
+    runs = collect_runs(resolved_workdir)
+    if not runs:
+        print("No metrics found. Run training first.")
+        return
+
+    env_ids = sorted({env_id for env_id, _ in runs.keys()})
+    plot_dir = os.path.dirname(resolved_plot_path) or "."
+    os.makedirs(plot_dir, exist_ok=True)
+    for env_id in env_ids:
+        plt.figure(figsize=(10, 6))
+        found = False
+        for version in resolved_versions:
+            series = runs.get((env_id, version), [])
+            if not series:
+                continue
+            steps, mean, std = aggregate_group(series)
+            if len(steps) == 0:
+                continue
+            found = True
+            x = steps / 1000.0
+            label = version.upper()
+            plt.plot(x, mean, label=label)
+            if len(series) > 1:
+                plt.fill_between(x, mean - std, mean + std, alpha=0.2)
+        if not found:
+            plt.close()
+            continue
+        plt.xlabel("Env Steps (k)")
+        plt.ylabel("Return")
+        plt.title(f"Dreamer V1/V2/V3 on {env_id}")
+        plt.grid(True, alpha=0.3)
+        plt.legend()
+        plt.tight_layout()
+        out_path = env_plot_path(resolved_plot_path, env_id)
+        plt.savefig(out_path, dpi=200)
+        plt.close()
+        print(f"Saved plot to {out_path}")
+
+
+if __name__ == "__main__":
+    plot_returns()
diff --git a/ding/bonus/dreamer/train.py b/ding/bonus/dreamer/train.py
new file mode 100644
index 0000000000..c4f4325016
--- /dev/null
+++ b/ding/bonus/dreamer/train.py
@@ -0,0 +1,344 @@
+import csv
+import os
+import time
+from collections import deque
+from typing import Optional
+
+import numpy as np
+import torch
+from tqdm import trange
+
+try:
+    import gymnasium as gym
+except ImportError:  # fallback
+    import gym
+
+from config import DreamerConfig, TrainConfig
+from dreamer.agent import make_agent
+from dreamer.replay import ReplayBuffer
+
+
+def seed_everything(seed: int) -> None:
+    """设置随机种子以提升可复现性。"""
+    import random
+
+    random.seed(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    torch.cuda.manual_seed_all(seed)
+    if hasattr(torch.backends, "cudnn"):
+        torch.backends.cudnn.deterministic = True
+        torch.backends.cudnn.benchmark = False
+    try:
+        torch.use_deterministic_algorithms(True, warn_only=True)
+    except Exception:
+        pass
+
+
+class RunningMeanStd:
+    """在线估计均值与方差，用于归一化。"""
+
+    def __init__(self, epsilon: float = 1e-4, shape=()):
+        """初始化统计量。"""
+        self.mean = np.zeros(shape, "float64")
+        self.var = np.ones(shape, "float64")
+        self.count = epsilon
+
+    def update(self, x: np.ndarray) -> None:
+        """用新样本更新均值与方差。"""
+        x = np.asarray(x, dtype="float64")
+        batch_mean = np.mean(x, axis=0)
+        batch_var = np.var(x, axis=0)
+        batch_count = x.shape[0]
+        self._update_from_moments(batch_mean, batch_var, batch_count)
+
+    def _update_from_moments(self, batch_mean, batch_var, batch_count) -> None:
+        """从批次统计量更新。"""
+        delta = batch_mean - self.mean
+        total_count = self.count + batch_count
+        new_mean = self.mean + delta * batch_count / total_count
+        m_a = self.var * self.count
+        m_b = batch_var * batch_count
+        m2 = m_a + m_b + delta**2 * self.count * batch_count / total_count
+        new_var = m2 / total_count
+        self.mean = new_mean
+        self.var = new_var
+        self.count = total_count
+
+
+class ObsNormalizer:
+    """观测归一化器。"""
+
+    def __init__(self, shape, clip: float = 5.0):
+        """初始化观测归一化器。"""
+        self.rms = RunningMeanStd(shape=shape)
+        self.clip = clip
+
+    def update(self, obs: np.ndarray) -> None:
+        """更新观测统计量。"""
+        self.rms.update(obs)
+
+    def __call__(self, obs: np.ndarray) -> np.ndarray:
+        """对观测进行归一化并裁剪。"""
+        obs = (obs - self.rms.mean) / (np.sqrt(self.rms.var) + 1e-8)
+        return np.clip(obs, -self.clip, self.clip)
+
+
+def reset_env(env, seed=None):
+    """重置环境并兼容 gym/gymnasium 返回格式。"""
+    out = env.reset(seed=seed)
+    if isinstance(out, tuple):
+        obs = out[0]
+    else:
+        obs = out
+    return obs
+
+
+def step_env(env, action):
+    """执行一步交互并统一返回格式。"""
+    out = env.step(action)
+    if len(out) == 5:
+        obs, reward, terminated, truncated, info = out
+        done = terminated or truncated
+        terminal = terminated
+    else:
+        obs, reward, done, info = out
+        terminal = done
+    return obs, reward, done, terminal, info
+
+
+def epsilon_by_step(step: int, cfg: TrainConfig) -> float:
+    """按训练步数计算 epsilon 值。"""
+    if cfg.exploration_decay_steps <= 0:
+        return cfg.exploration_epsilon_end
+    progress = min(1.0, step / float(cfg.exploration_decay_steps))
+    return cfg.exploration_epsilon_start + progress * (cfg.exploration_epsilon_end - cfg.exploration_epsilon_start)
+
+
+def evaluate(agent, env_id: str, episodes: int, seed: int, normalizer: Optional[ObsNormalizer], env_kwargs=None) -> float:
+    """评估当前策略的平均回报。"""
+    env = gym.make(env_id, **(env_kwargs or {}))
+    scores = []
+    for ep in range(episodes):
+        obs = reset_env(env, seed=seed + 1000 + ep)
+        state = agent.init_state(1)
+        prev_action = torch.zeros(1, agent.config.action_dim, device=agent.device)
+        done = False
+        total = 0.0
+        while not done:
+            obs_in = normalizer(obs) if normalizer is not None else obs
+            state = agent.observe(obs_in, state, prev_action)
+            action = agent.policy(state, eval_mode=True)
+            prev_action = agent.action_to_onehot(action)
+            obs, reward, done, _, _ = step_env(env, action)
+            total += reward
+        scores.append(total)
+    env.close()
+    return float(np.mean(scores))
+
+
+def init_metrics_csv(path: str, overwrite: bool = False) -> None:
+    """初始化评估指标 CSV 文件。"""
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    if os.path.exists(path) and not overwrite:
+        return
+    with open(path, "w", newline="") as f:
+        writer = csv.writer(f)
+        writer.writerow(["env_steps", "avg_return", "seed"])
+
+
+def append_metrics_csv(path: str, env_steps: int, avg_return: float, seed: int) -> None:
+    """追加一条评估记录。"""
+    with open(path, "a", newline="") as f:
+        writer = csv.writer(f)
+        writer.writerow([env_steps, f"{avg_return:.4f}", seed])
+
+
+def reset_episode(env, agent, action_dim: int, seed: Optional[int] = None):
+    """重置环境并初始化状态。"""
+    obs = reset_env(env, seed=seed)
+    state = agent.init_state(1)
+    prev_action = torch.zeros(1, action_dim, device=agent.device)
+    return obs, state, prev_action
+
+
+def normalize_obs(obs: np.ndarray, normalizer: Optional[ObsNormalizer]) -> np.ndarray:
+    """按需更新并归一化观测。"""
+    if normalizer is None:
+        return obs
+    normalizer.update(np.asarray([obs]))
+    return normalizer(obs)
+
+
+def resolve_run_workdir(base_workdir: str, time_format: str, timestamp_if_exists: bool) -> str:
+    """若基础目录已存在，则追加时间戳形成新的实验目录。"""
+    if (not timestamp_if_exists) or (not os.path.exists(base_workdir)):
+        return base_workdir
+    timestamp = time.strftime(time_format, time.localtime())
+    candidate = f"{base_workdir}_{timestamp}"
+    if not os.path.exists(candidate):
+        return candidate
+    suffix = 1
+    while True:
+        candidate_with_suffix = f"{candidate}_{suffix}"
+        if not os.path.exists(candidate_with_suffix):
+            return candidate_with_suffix
+        suffix += 1
+
+
+def train(cfg: TrainConfig, env_id: str, version: str, seed: int) -> None:
+    """训练单个环境/版本/种子组合。"""
+    seed_everything(seed)
+
+    # 1. 初始化实验目录
+    seed_dir = os.path.join(cfg.workdir, env_id, version, f"seed_{seed}")
+    metrics_csv = os.path.join(seed_dir, "metrics", "eval_returns.csv")
+    os.makedirs(seed_dir, exist_ok=True)
+    init_metrics_csv(metrics_csv, overwrite=cfg.overwrite_metrics)
+
+    # 2. 初始化环境与基础信息
+    env = gym.make(env_id, **(cfg.env_kwargs or {}))
+    obs = reset_env(env, seed=seed)
+    if not hasattr(env.action_space, "n"):
+        raise ValueError(f"{env_id} action space is not discrete; this implementation only supports Discrete.")
+    obs_dim = int(np.prod(env.observation_space.shape))
+    action_dim = env.action_space.n
+
+    # 3. 初始化 Agent / Replay / Normalizer
+    config = DreamerConfig(
+        obs_dim=obs_dim,
+        action_dim=action_dim,
+        embed_dim=cfg.embed_dim,
+        deter_dim=cfg.deter_dim,
+        stoch_dim=cfg.stoch_dim,
+        stoch_classes=cfg.stoch_classes,
+        hidden_dim=cfg.hidden_dim,
+        horizon=cfg.horizon,
+        model_lr=cfg.model_lr,
+        actor_lr=cfg.actor_lr,
+        critic_lr=cfg.critic_lr,
+        entropy_scale=cfg.entropy_scale,
+        use_obs_norm=cfg.use_obs_norm,
+        normalize_advantage=cfg.normalize_advantage,
+        discount=cfg.discount,
+        lambda_=cfg.lambda_,
+        free_nats=cfg.free_nats,
+        kl_scale=cfg.kl_scale,
+        discount_scale=cfg.discount_scale,
+        kl_balance=cfg.kl_balance,
+        target_tau=cfg.target_tau,
+    )
+    device = torch.device(cfg.device)
+    agent = make_agent(version, config, device)
+
+    replay = ReplayBuffer(cfg.replay_size)
+    normalizer = ObsNormalizer(shape=obs.shape) if config.use_obs_norm else None
+
+    episode_obs, episode_actions, episode_rewards, episode_dones = [], [], [], []
+    episode_return = 0.0
+    recent_returns = deque(maxlen=10)
+
+    obs, state, prev_action = reset_episode(env, agent, action_dim, seed=seed)
+
+    pbar = trange(cfg.total_steps, desc=f"Training {version} {env_id} seed {seed}")
+    for step in pbar:
+        # 4. 观测预处理
+        obs_raw = obs
+        obs_norm = normalize_obs(obs_raw, normalizer)
+
+        # 5. 更新后验状态
+        state = agent.observe(obs_norm, state, prev_action)
+
+        # 6. 选择动作（探索 or 策略）
+        if step < cfg.seed_steps:
+            action = env.action_space.sample()
+        else:
+            eps = epsilon_by_step(step - cfg.seed_steps, cfg)
+            if np.random.rand() < eps:
+                action = env.action_space.sample()
+            else:
+                action = agent.policy(state, eval_mode=False)
+
+        # 7. 与环境交互
+        prev_action = agent.action_to_onehot(action)
+        next_obs, reward, done, _, _ = step_env(env, action)
+
+        # 8. 记录 episode 数据
+        # 这里保存动作后的观测，使 obs / action / reward / done 在时间上严格对齐。
+        episode_obs.append(next_obs)
+        episode_actions.append(action)
+        episode_rewards.append(reward)
+        episode_dones.append(done)
+        episode_return += reward
+
+        obs = next_obs
+
+        # 9. Episode 结束处理
+        if done:
+            replay.add_episode(episode_obs, episode_actions, episode_rewards, episode_dones)
+            recent_returns.append(episode_return)
+            obs, state, prev_action = reset_episode(env, agent, action_dim, seed=seed + step + 1)
+            episode_obs, episode_actions, episode_rewards, episode_dones = [], [], [], []
+            episode_return = 0.0
+
+        # 10. 训练更新
+        if len(replay) >= cfg.batch_size * cfg.seq_len and step >= cfg.seed_steps:
+            if step % cfg.train_every == 0:
+                metrics = None
+                for _ in range(cfg.train_steps):
+                    batch = replay.sample(cfg.batch_size, cfg.seq_len)
+                    if normalizer is not None:
+                        batch["obs"] = normalizer(batch["obs"])
+                    metrics = agent.train_step(batch)
+                if metrics is not None:
+                    pbar.set_postfix(
+                        {
+                            "model": f"{metrics.model_loss:.3f}",
+                            "actor": f"{metrics.actor_loss:.3f}",
+                            "critic": f"{metrics.critic_loss:.3f}",
+                            "return": f"{np.mean(recent_returns) if recent_returns else 0.0:.1f}",
+                        }
+                    )
+
+        # 11. 日志打印
+        if (step + 1) % cfg.log_every == 0:
+            avg_return = float(np.mean(recent_returns)) if recent_returns else 0.0
+            print(f"Step {step + 1} | avg_return={avg_return:.1f}")
+
+        # 12. 定期评估 + 记录 CSV
+        if (step + 1) % cfg.eval_every == 0:
+            eval_return = evaluate(agent, env_id, cfg.eval_episodes, seed, normalizer, env_kwargs=cfg.env_kwargs)
+            print(f"Eval at step {step + 1}: avg_return={eval_return:.1f}")
+            append_metrics_csv(metrics_csv, step + 1, eval_return, seed)
+
+    env.close()
+
+
+def main() -> None:
+    """入口函数，依次训练所有组合并绘图。"""
+    cfg = TrainConfig()
+    cfg.workdir = resolve_run_workdir(
+        cfg.workdir,
+        time_format=cfg.workdir_time_format,
+        timestamp_if_exists=cfg.timestamp_workdir_if_exists,
+    )
+    cfg.plot_path = os.path.join(cfg.workdir, "return_curve.png")
+    os.makedirs(cfg.workdir, exist_ok=True)
+    print(f"Run workdir: {cfg.workdir}")
+
+    for env_id in cfg.env_ids:
+        for version in cfg.agent_versions:
+            for seed in cfg.seeds:
+                print(f"\n=== Training {version} on {env_id} with seed {seed} ===")
+                train(cfg, env_id, version, seed)
+
+    try:
+        from plot_returns import plot_returns
+
+        plot_returns(workdir=cfg.workdir, plot_path=cfg.plot_path, agent_versions=cfg.agent_versions)
+    except Exception as exc:
+        print(f"Plotting skipped: {exc}")
+
+
+if __name__ == "__main__":
+    main()