Lesson 06 · 支柱 B · 控制

无模型 DRL 设定值优化（SAC）

预测已就位，现在让它"动起来"——用深度强化学习优化 HVAC 设定值，在能耗与舒适间找最优。

一、为什么是无模型，为什么是 SAC

先澄清一个常见混淆（第 1 课讲过，这里再敲一遍）：

无模型 RL 里，连续控制（设定值是连续数）的事实首选是 SAC（Soft Actor-Critic）。三个理由：

• off-policy：历史交互存进 replay buffer 反复用，样本高效——这对每一步都是昂贵 EnergyPlus 仿真的 HVAC 至关重要。
• 熵正则：自带探索，不易陷局部最优；对奖励形状不敏感。
• 连续动作 SOTA：SB3 文档明确把 SAC 列为单进程连续控制首选。

二、奖励函数：复刻 BrainBox 的多目标

BrainBox 专利反复出现"user objective indicators"，奖励是能耗 + 舒适度 + 设备磨损的加权。这是整课最需要"调"的部分——权重直接决定策略行为。

def hvac_reward(obs, action, prev_action, pwr_max=1e6):
    # obs 里取出关键量（列号按你的 env 调整）
    T_zone  = obs[0]          # 区域温度 °C
    co2     = obs[1]          # CO₂ ppm
    fan_pwr = obs[2]          # 风机功率 W
    chiller = obs[3]          # 冷机功率 W

    # ① 能耗（归一化）
    e_energy = (fan_pwr + chiller) / pwr_max

    # ② 舒适违规：超出 ASHRAE-55 的 18–26°C 死区
    over_hot  = max(0.0, T_zone - 26.0)
    over_cold = max(0.0, 18.0 - T_zone)
    comfort   = 2.0 * (over_hot + over_cold)        # K/K 斜率

    # ③ IAQ：CO₂ 超 1000ppm 惩罚
    co2_pen = max(0.0, co2 - 1000.0) / 1000.0

    # ④ 设备磨损：动作抖动（设定点反复横跳伤压缩机）
    wear = np.sum(np.abs(action - prev_action))

    return -(0.4*e_energy + 0.4*comfort + 0.1*co2_pen + 0.1*wear)

三、动作空间：设定值 + 安全限幅

BrainBox 控的是设定值（setpoint），不是阀门。SAC 输出归一化的 [-1,1]，你要映射回真实物理范围并限幅 + 平滑。

SAFE = {"sat": (12.0, 18.0),     # 送风温度设定点 °C
        "chw": (5.0, 10.0),      # 冷冻水供水温度设定点 °C
        "airflow": (0.2, 1.0)}   # 风量占比

def to_safe_action(a, prev):
    # a ∈ [-1,1]^3 → 真实设定值
    sat = (a[0]+1)/2 * (SAFE["sat"][1]-SAFE["sat"][0]) + SAFE["sat"][0]
    chw = (a[1]+1)/2 * (SAFE["chw"][1]-SAFE["chw"][0]) + SAFE["chw"][0]
    af  = (a[2]+1)/2 * (SAFE["airflow"][1]-SAFE["airflow"][0]) + SAFE["airflow"][0]
    raw = np.clip([sat, chw, af],
                  [SAFE[k][0] for k in SAFE], [SAFE[k][1] for k in SAFE])
    # EMA 平滑，防策略"乱抖"伤设备
    smoothed = 0.7*prev + 0.3*raw
    return smoothed

限幅保证 SAC 永远发不出"冻管"或"失流"指令；平滑压制高频抖动。这两步是第 7 课安全门控的雏形。

四、动手做：SAC 训练完整闭环

import gymnasium as gym, sinergym
from stable_baselines3 import SAC
from stable_baselines3.common.vec_env import DummyVecEnv
from stable_baselines3.common.callbacks import EvalCallback
from gymnasium.wrappers import NormalizeObservation
from sinergym.utils.wrappers import NormalizeAction, LoggerWrapper

ENV_ID = "Eplus-5zone-hot-continuous-stochastic-v1"

def make_env():
    env = gym.make(ENV_ID)
    env = NormalizeObservation(env)        # 观测归一化（必做！）
    env = NormalizeAction(env)             # 动作映射到 [-1,1]
    env = LoggerWrapper(env)               # 记录能耗/舒适
    return env

train_env = DummyVecEnv([make_env])
eval_env  = DummyVecEnv([make_env])        # 独立评估环境

model = SAC("MlpPolicy", train_env,
            learning_rate=3e-4, buffer_size=100_000,
            batch_size=256, gamma=0.99, verbose=1)

# 评估回调：每 N 步在 eval_env 上跑，存最优模型
cb = EvalCallback(eval_env, best_model_save_path="./sac_best",
                  log_path="./sac_logs", eval_freq=5_000,
                  deterministic=True)

model.learn(total_timesteps=50_000, callback=cb)   # 先小步快跑通
model.save("sac_hvac")
print("训练完成，最优模型在 ./sac_best")

import gymnasium as gym, sinergym, numpy as np
from stable_baselines3 import SAC

def run_policy(model, env_id, steps=3504):   # 一年
    env = gym.make(env_id)
    obs, _ = env.reset()
    energy, comfort_viol = 0.0, 0
    for _ in range(steps):
        action, _ = model.predict(obs, deterministic=True)
        obs, r, term, trunc, info = env.step(action)
        energy += info.get("total_power", 0.0)
        if obs[0] > 26 or obs[0] < 18: comfort_viol += 1
        if term or trunc: obs, _ = env.reset()
    env.close()
    return energy, comfort_viol

sac = SAC.load("./sac_best/best_model")
e_sac, c_sac = run_policy(sac, "Eplus-5zone-hot-continuous-stochastic-v1")
# e_pid, c_pid = run_policy(pid_model, ...)   # 第 3 课的 PID 包成同样接口
# saving = (e_pid - e_sac) / e_pid * 100
print(f"SAC 能耗={e_sac:.0f}  违规步数={c_sac}")

五、提速与冷启动（进阶，但必须知道）

无模型 RL 在 HVAC 上的两大现实痛点：训练慢、真实楼不能随便试错。

训练慢 → 并行 + 离线预训练

import d3rlpy
# 从 BMS 历史日志构造 (obs, action, reward, next_obs, terminal)
dataset = d3rlpy.dataset.MDPDataset(observations, actions, rewards, terminals)
iql = d3rlpy.algos.IQLConfig().create(device="cuda:0")
iql.build_with_dataset(dataset)
iql.fit(dataset.episodes, n_steps=100_000)
# 再把 IQL 权重灌进 SAC 做在线微调，省掉大量危险探索