第 15 課：強化學習生產 — 部署與監控強化學習代理

強化學習：從基礎到高級

第 4 部分：RLHF、LLM 調整和製作

亞洲開發網

簡介

將強化學習代理部署到生產中與監督式機器學習有很大不同——需要處理安全約束、線上學習、獎勵監控和策略版本控制。

---

1. 模型匯出與服務

ONNX 匯出

import torch
from stable_baselines3 import PPO

model = PPO.load("best_model")

# Export policy to ONNX
dummy_input = torch.randn(1, model.observation_space.shape[0])
torch.onnx.export(
    model.policy, dummy_input, "policy.onnx",
    input_names=["observation"],
    output_names=["action"],
    dynamic_axes={"observation": {0: "batch"}, "action": {0: "batch"}}
)

FastAPI 服務

from fastapi import FastAPI
import onnxruntime as ort
import numpy as np

app = FastAPI()
session = ort.InferenceSession("policy.onnx")

@app.post("/predict")
def predict(observation: list[float]):
    obs = np.array([observation], dtype=np.float32)
    result = session.run(None, {"observation": obs})
    action = int(np.argmax(result[0]))
    return {"action": action}

---

2. 安全限制

class SafeRLPolicy:
    def __init__(self, model, constraints):
        self.model = model
        self.constraints = constraints
    
    def predict(self, observation):
        action, _ = self.model.predict(observation, deterministic=True)
        
        # Check safety constraints
        if self.constraints.is_unsafe(observation, action):
            action = self.constraints.safe_fallback(observation)
            self.log_safety_override(observation, action)
        
        return action
    
    def log_safety_override(self, obs, action):
        # Track safety overrides for monitoring
        pass

---

3. 離線盧比

根據記錄的資料進行訓練－無環境互動：

# Conservative Q-Learning (CQL)
from d3rlpy.algos import CQLConfig

cql = CQLConfig().create(device="cuda")
cql.fit(
    offline_dataset,
    n_steps=100_000,
    evaluators={"environment": gym_evaluator}
)

---

4. 監控與 A/B 測試

class RLMonitor:
    def __init__(self):
        self.rewards = []
        self.actions = []
    
    def log_step(self, obs, action, reward):
        self.rewards.append(reward)
        self.actions.append(action)
    
    def check_drift(self, window=1000):
        recent = self.rewards[-window:]
        historical = self.rewards[-2*window:-window]
        # Statistical test for reward drift
        from scipy.stats import ks_2samp
        statistic, p_value = ks_2samp(recent, historical)
        if p_value < 0.05:
            alert("Reward distribution drift detected!")

---

5. 生產清單

方面	方法
模型格式	ONNX 或 TorchScript
服務	FastAPI + 非同步推理
安全	硬約束+後備政策
監控	獎勵追蹤、漂移檢測
版本控制	政策版本登記
回滾	即時切換到先前的政策
A/B 測試	金絲雀部署
記錄	完整的狀態-動作-獎勵軌跡

---

總結

主題	要點
出口	用於跨平台部署的 ONNX
安全	總是有後備政策
監控	追蹤一段時間內的獎勵分配
離線強化學習	無法在線上時從日誌進行訓練