离线强化学习实战如何用Python和TensorFlow从历史数据中训练智能体在人工智能的诸多分支中强化学习因其能够通过试错机制学习最优策略而备受关注。然而传统的在线强化学习需要智能体与环境持续交互这在许多实际场景中可能面临成本高昂或安全风险等问题。想象一下如果你正在开发一个医疗决策系统显然不能通过试错来获取实时反馈。这时离线强化学习Offline RL便展现出其独特价值——它允许我们仅利用历史数据就能训练出高效的决策模型。与在线强化学习相比离线强化学习有几个显著优势首先它完全避免了与环境交互的高昂成本其次它可以充分利用已有的海量历史数据最后它在安全性要求高的场景中具有不可替代的作用。金融交易策略优化、工业控制系统、推荐系统等领域的开发者尤其需要掌握这项技术。本文将手把手带你用Python和TensorFlow实现一个完整的离线强化学习流程从数据预处理到模型部署每个环节都会提供可运行的代码示例。1. 离线强化学习基础与环境准备1.1 核心概念与算法选择离线强化学习有时也被称为批量强化学习或历史数据驱动的强化学习其核心挑战在于如何从静态数据集中学习有效的策略而不进行任何新的探索。这与在线强化学习形成鲜明对比——后者可以不断尝试新动作并观察结果。这种差异带来了几个关键问题分布偏移训练数据中的状态-动作分布可能与学习策略产生的分布不同外推误差对数据集中未出现过的状态-动作对进行价值评估时可能出现误差策略约束需要限制学习策略不要偏离数据收集策略太远目前主流的离线RL算法可以分为三类算法类型代表算法核心思想适用场景策略约束型BCQ, BEAR限制学习策略接近数据收集策略数据覆盖有限不确定性感知型MOPO, MOReL对未知区域进行悲观估计安全关键场景价值正则化型CQL, BRAC通过正则化避免对OOD数据高估通用场景对于本教程我们将选择保守Q学习CQL作为基础算法因为它在各种基准测试中表现稳健且实现相对简单。1.2 开发环境配置在开始编码前我们需要准备Python环境。推荐使用conda创建隔离的环境conda create -n offline_rl python3.8 conda activate offline_rl pip install tensorflow2.7.0 pip install gym0.21.0 pip install pandas matplotlib tqdm提示如果使用GPU加速训练请安装对应版本的tensorflow-gpu和CUDA工具包我们将使用TensorFlow 2.x实现算法核心Gym提供标准环境接口Pandas用于数据处理。下面代码检查环境是否配置正确import tensorflow as tf print(TensorFlow版本:, tf.__version__) print(GPU可用:, tf.config.list_physical_devices(GPU))2. 数据准备与预处理2.1 数据集获取与理解离线强化学习的质量很大程度上取决于数据集的质量。我们可以从以下几个渠道获取合适的数据公开的强化学习基准数据集如D4RL从在线系统中收集的历史交互日志通过模拟器预先生成的数据为简化示例我们将使用Gym的Pendulum环境生成一个简单的数据集import gym import numpy as np def generate_dataset(env_namePendulum-v1, num_episodes1000): env gym.make(env_name) dataset { states: [], actions: [], rewards: [], next_states: [], dones: [] } for _ in range(num_episodes): state env.reset() done False while not done: action env.action_space.sample() # 随机策略生成数据 next_state, reward, done, _ env.step(action) dataset[states].append(state) dataset[actions].append(action) dataset[rewards].append(reward) dataset[next_states].append(next_state) dataset[dones].append(done) state next_state return {k: np.array(v) for k, v in dataset.items()} pendulum_data generate_dataset()2.2 数据预处理流程原始数据通常需要经过以下处理步骤才能用于训练标准化将状态和动作归一化到相近的范围分割划分为训练集和验证集转换组织成适合强化学习训练的形式下面是具体的实现代码from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split def preprocess_data(data, test_size0.2): # 状态标准化 state_scaler StandardScaler() states state_scaler.fit_transform(data[states]) # 动作标准化假设连续动作空间 action_scaler StandardScaler() actions action_scaler.fit_transform(data[actions]) # 构建完整数据集 X np.concatenate([states, actions], axis1) y { rewards: data[rewards], next_states: state_scaler.transform(data[next_states]), dones: data[dones].astype(np.float32) } # 数据集分割 X_train, X_val, y_train, y_val train_test_split( X, y, test_sizetest_size, random_state42) # 组织为训练需要的格式 train_data { states: X_train[:, :states.shape[1]], actions: X_train[:, states.shape[1]:], **{k: v for k, v in y_train.items()} } val_data { states: X_val[:, :states.shape[1]], actions: X_val[:, states.shape[1]:], **{k: v for k, v in y_val.items()} } return train_data, val_data, state_scaler, action_scaler train_data, val_data, state_scaler, action_scaler preprocess_data(pendulum_data)注意在实际应用中你可能还需要处理数据不平衡、缺失值等问题这里假设数据已经相对干净3. 模型构建与CQL算法实现3.1 神经网络结构设计我们将构建一个包含以下组件的深度强化学习系统Q网络评估状态-动作对的价值策略网络根据状态选择最优动作目标网络稳定训练过程首先定义Q网络的结构from tensorflow.keras.layers import Dense, Input, Concatenate from tensorflow.keras.models import Model def create_q_network(state_dim, action_dim, hidden_units(256, 256)): state_input Input(shape(state_dim,)) action_input Input(shape(action_dim,)) # 状态处理分支 x Dense(hidden_units[0], activationrelu)(state_input) x Dense(hidden_units[1], activationrelu)(x) # 动作处理分支 y Dense(hidden_units[0], activationrelu)(action_input) y Dense(hidden_units[1], activationrelu)(y) # 合并 concat Concatenate()([x, y]) q_output Dense(1)(concat) return Model(inputs[state_input, action_input], outputsq_output)接下来是策略网络对于连续动作空间def create_policy_network(state_dim, action_dim, hidden_units(256, 256)): state_input Input(shape(state_dim,)) x Dense(hidden_units[0], activationrelu)(state_input) x Dense(hidden_units[1], activationrelu)(x) action_output Dense(action_dim, activationtanh)(x) # 假设动作在[-1,1]范围 return Model(inputsstate_input, outputsaction_output)3.2 保守Q学习CQL实现CQL的核心思想是在标准Q学习目标上增加一个正则项最小化非数据分布动作的Q值同时最大化数据分布动作的Q值。下面是关键实现import tensorflow as tf class CQLAgent: def __init__(self, state_dim, action_dim, gamma0.99, tau0.005, cql_weight1.0): self.state_dim state_dim self.action_dim action_dim self.gamma gamma # 折扣因子 self.tau tau # 目标网络更新系数 self.cql_weight cql_weight # CQL正则项权重 # 创建网络 self.q_net1 create_q_network(state_dim, action_dim) self.q_net2 create_q_network(state_dim, action_dim) self.target_q_net1 create_q_network(state_dim, action_dim) self.target_q_net2 create_q_network(state_dim, action_dim) self.policy_net create_policy_network(state_dim, action_dim) self.target_policy_net create_policy_network(state_dim, action_dim) # 初始化目标网络权重 self.update_target_networks(tau1.0) # 优化器 self.q_optimizer tf.keras.optimizers.Adam(learning_rate3e-4) self.policy_optimizer tf.keras.optimizers.Adam(learning_rate3e-4) def update_target_networks(self, tau): # 更新目标Q网络 for src, dest in zip(self.q_net1.variables, self.target_q_net1.variables): dest.assign(tau * src (1 - tau) * dest) for src, dest in zip(self.q_net2.variables, self.target_q_net2.variables): dest.assign(tau * src (1 - tau) * dest) # 更新目标策略网络 for src, dest in zip(self.policy_net.variables, self.target_policy_net.variables): dest.assign(tau * src (1 - tau) * dest) tf.function def train_step(self, batch): states, actions, rewards, next_states, dones batch # 计算目标Q值 next_actions self.target_policy_net(next_states) target_q1 self.target_q_net1([next_states, next_actions]) target_q2 self.target_q_net2([next_states, next_actions]) target_q tf.minimum(target_q1, target_q2) y rewards (1 - dones) * self.gamma * target_q # 标准Q学习损失 with tf.GradientTape(persistentTrue) as tape: current_q1 self.q_net1([states, actions]) current_q2 self.q_net2([states, actions]) q1_loss tf.reduce_mean(tf.square(current_q1 - y)) q2_loss tf.reduce_mean(tf.square(current_q2 - y)) # CQL正则项 random_actions tf.random.uniform(actions.shape, -1, 1) q1_rand self.q_net1([states, random_actions]) q2_rand self.q_net2([states, random_actions]) cql_loss1 tf.reduce_mean(q1_rand - current_q1) cql_loss2 tf.reduce_mean(q2_rand - current_q2) total_q1_loss q1_loss self.cql_weight * cql_loss1 total_q2_loss q2_loss self.cql_weight * cql_loss2 # 更新Q网络 q1_grads tape.gradient(total_q1_loss, self.q_net1.trainable_variables) self.q_optimizer.apply_gradients(zip(q1_grads, self.q_net1.trainable_variables)) q2_grads tape.gradient(total_q2_loss, self.q_net2.trainable_variables) self.q_optimizer.apply_gradients(zip(q2_grads, self.q_net2.trainable_variables)) # 策略学习 with tf.GradientTape() as tape: new_actions self.policy_net(states) new_q self.q_net1([states, new_actions]) policy_loss -tf.reduce_mean(new_q) policy_grads tape.gradient(policy_loss, self.policy_net.trainable_variables) self.policy_optimizer.apply_gradients(zip(policy_grads, self.policy_net.trainable_variables)) # 更新目标网络 self.update_target_networks(self.tau) return { q1_loss: q1_loss.numpy(), q2_loss: q2_loss.numpy(), cql_loss1: cql_loss1.numpy(), cql_loss2: cql_loss2.numpy(), policy_loss: policy_loss.numpy() }4. 训练流程与性能评估4.1 训练循环实现有了核心算法后我们需要实现完整的训练流程import numpy as np from tqdm import trange def train_agent(agent, train_data, val_data, epochs100, batch_size256): # 准备数据集 dataset tf.data.Dataset.from_tensor_slices(( train_data[states], train_data[actions], train_data[rewards], train_data[next_states], train_data[dones] )).shuffle(100000).batch(batch_size) # 训练循环 history {train: [], val: []} for epoch in trange(epochs): # 训练阶段 epoch_losses [] for batch in dataset: losses agent.train_step(batch) epoch_losses.append(losses) # 计算平均损失 avg_losses {k: np.mean([l[k] for l in epoch_losses]) for k in epoch_losses[0]} history[train].append(avg_losses) # 验证阶段可选 if val_data: val_batch ( val_data[states], val_data[actions], val_data[rewards], val_data[next_states], val_data[dones] ) val_losses agent.train_step(val_batch) # 注意这里不更新参数 history[val].append(val_losses) return history # 初始化agent state_dim train_data[states].shape[1] action_dim train_data[actions].shape[1] agent CQLAgent(state_dim, action_dim) # 开始训练 history train_agent(agent, train_data, val_data, epochs50)4.2 性能评估与可视化训练完成后我们需要评估模型性能。对于Pendulum环境我们可以计算平均回报def evaluate_policy(agent, env, n_episodes10): total_rewards [] for _ in range(n_episodes): state env.reset() done False episode_reward 0 while not done: action agent.policy_net(state[np.newaxis, :]) next_state, reward, done, _ env.step(action[0]) episode_reward reward state next_state total_rewards.append(episode_reward) return np.mean(total_rewards) env gym.make(Pendulum-v1) mean_reward evaluate_policy(agent, env) print(f平均回报: {mean_reward:.2f})我们还可以可视化训练过程中的损失变化import matplotlib.pyplot as plt def plot_training_history(history): plt.figure(figsize(12, 8)) # Q损失 plt.subplot(2, 2, 1) plt.plot([h[q1_loss] for h in history[train]], labelQ1 Loss) plt.plot([h[q2_loss] for h in history[train]], labelQ2 Loss) plt.title(Q Network Loss) plt.legend() # CQL损失 plt.subplot(2, 2, 2) plt.plot([h[cql_loss1] for h in history[train]], labelCQL Loss1) plt.plot([h[cql_loss2] for h in history[train]], labelCQL Loss2) plt.title(CQL Regularization Loss) plt.legend() # 策略损失 plt.subplot(2, 2, 3) plt.plot([h[policy_loss] for h in history[train]], labelPolicy Loss) plt.title(Policy Loss) plt.legend() plt.tight_layout() plt.show() plot_training_history(history)5. 实际应用与优化技巧5.1 部署到生产环境将训练好的离线RL模型部署到生产环境需要考虑几个关键点模型序列化保存和加载模型权重推理优化优化预测速度监控跟踪模型性能衰减保存和加载模型的示例代码# 保存模型 def save_agent(agent, path): agent.q_net1.save_weights(f{path}/q_net1.h5) agent.q_net2.save_weights(f{path}/q_net2.h5) agent.policy_net.save_weights(f{path}/policy_net.h5) # 加载模型 def load_agent(agent, path): agent.q_net1.load_weights(f{path}/q_net1.h5) agent.q_net2.load_weights(f{path}/q_net2.h5) agent.policy_net.load_weights(f{path}/policy_net.h5) # 使用示例 save_agent(agent, saved_models) new_agent CQLAgent(state_dim, action_dim) load_agent(new_agent, saved_models)5.2 性能优化技巧根据实践经验以下是提升离线RL性能的几个有效方法数据质量确保数据集覆盖足够多的状态-动作空间超参数调优特别是CQL权重和学习率集成学习结合多个Q网络的预测结果状态表征使用自编码器学习更好的状态特征一个简单的集成Q学习实现class EnsembleQNetwork: def __init__(self, state_dim, action_dim, n_networks5): self.networks [create_q_network(state_dim, action_dim) for _ in range(n_networks)] def __call__(self, states, actions): q_values [net([states, actions]) for net in self.networks] return tf.reduce_mean(q_values, axis0)在实际项目中我发现调整CQL权重对最终性能影响最大。当数据集覆盖不够全面时需要增大CQL权重以避免对OODOut-of-Distribution数据的高估而当数据集质量很高时可以适当减小CQL权重让模型更积极地探索高回报区域。
离线强化学习实战：如何用Python和TensorFlow从历史数据中训练智能体（附完整代码）
发布时间：2026/5/19 3:51:17
离线强化学习实战如何用Python和TensorFlow从历史数据中训练智能体在人工智能的诸多分支中强化学习因其能够通过试错机制学习最优策略而备受关注。然而传统的在线强化学习需要智能体与环境持续交互这在许多实际场景中可能面临成本高昂或安全风险等问题。想象一下如果你正在开发一个医疗决策系统显然不能通过试错来获取实时反馈。这时离线强化学习Offline RL便展现出其独特价值——它允许我们仅利用历史数据就能训练出高效的决策模型。与在线强化学习相比离线强化学习有几个显著优势首先它完全避免了与环境交互的高昂成本其次它可以充分利用已有的海量历史数据最后它在安全性要求高的场景中具有不可替代的作用。金融交易策略优化、工业控制系统、推荐系统等领域的开发者尤其需要掌握这项技术。本文将手把手带你用Python和TensorFlow实现一个完整的离线强化学习流程从数据预处理到模型部署每个环节都会提供可运行的代码示例。1. 离线强化学习基础与环境准备1.1 核心概念与算法选择离线强化学习有时也被称为批量强化学习或历史数据驱动的强化学习其核心挑战在于如何从静态数据集中学习有效的策略而不进行任何新的探索。这与在线强化学习形成鲜明对比——后者可以不断尝试新动作并观察结果。这种差异带来了几个关键问题分布偏移训练数据中的状态-动作分布可能与学习策略产生的分布不同外推误差对数据集中未出现过的状态-动作对进行价值评估时可能出现误差策略约束需要限制学习策略不要偏离数据收集策略太远目前主流的离线RL算法可以分为三类算法类型代表算法核心思想适用场景策略约束型BCQ, BEAR限制学习策略接近数据收集策略数据覆盖有限不确定性感知型MOPO, MOReL对未知区域进行悲观估计安全关键场景价值正则化型CQL, BRAC通过正则化避免对OOD数据高估通用场景对于本教程我们将选择保守Q学习CQL作为基础算法因为它在各种基准测试中表现稳健且实现相对简单。1.2 开发环境配置在开始编码前我们需要准备Python环境。推荐使用conda创建隔离的环境conda create -n offline_rl python3.8 conda activate offline_rl pip install tensorflow2.7.0 pip install gym0.21.0 pip install pandas matplotlib tqdm提示如果使用GPU加速训练请安装对应版本的tensorflow-gpu和CUDA工具包我们将使用TensorFlow 2.x实现算法核心Gym提供标准环境接口Pandas用于数据处理。下面代码检查环境是否配置正确import tensorflow as tf print(TensorFlow版本:, tf.__version__) print(GPU可用:, tf.config.list_physical_devices(GPU))2. 数据准备与预处理2.1 数据集获取与理解离线强化学习的质量很大程度上取决于数据集的质量。我们可以从以下几个渠道获取合适的数据公开的强化学习基准数据集如D4RL从在线系统中收集的历史交互日志通过模拟器预先生成的数据为简化示例我们将使用Gym的Pendulum环境生成一个简单的数据集import gym import numpy as np def generate_dataset(env_namePendulum-v1, num_episodes1000): env gym.make(env_name) dataset { states: [], actions: [], rewards: [], next_states: [], dones: [] } for _ in range(num_episodes): state env.reset() done False while not done: action env.action_space.sample() # 随机策略生成数据 next_state, reward, done, _ env.step(action) dataset[states].append(state) dataset[actions].append(action) dataset[rewards].append(reward) dataset[next_states].append(next_state) dataset[dones].append(done) state next_state return {k: np.array(v) for k, v in dataset.items()} pendulum_data generate_dataset()2.2 数据预处理流程原始数据通常需要经过以下处理步骤才能用于训练标准化将状态和动作归一化到相近的范围分割划分为训练集和验证集转换组织成适合强化学习训练的形式下面是具体的实现代码from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split def preprocess_data(data, test_size0.2): # 状态标准化 state_scaler StandardScaler() states state_scaler.fit_transform(data[states]) # 动作标准化假设连续动作空间 action_scaler StandardScaler() actions action_scaler.fit_transform(data[actions]) # 构建完整数据集 X np.concatenate([states, actions], axis1) y { rewards: data[rewards], next_states: state_scaler.transform(data[next_states]), dones: data[dones].astype(np.float32) } # 数据集分割 X_train, X_val, y_train, y_val train_test_split( X, y, test_sizetest_size, random_state42) # 组织为训练需要的格式 train_data { states: X_train[:, :states.shape[1]], actions: X_train[:, states.shape[1]:], **{k: v for k, v in y_train.items()} } val_data { states: X_val[:, :states.shape[1]], actions: X_val[:, states.shape[1]:], **{k: v for k, v in y_val.items()} } return train_data, val_data, state_scaler, action_scaler train_data, val_data, state_scaler, action_scaler preprocess_data(pendulum_data)注意在实际应用中你可能还需要处理数据不平衡、缺失值等问题这里假设数据已经相对干净3. 模型构建与CQL算法实现3.1 神经网络结构设计我们将构建一个包含以下组件的深度强化学习系统Q网络评估状态-动作对的价值策略网络根据状态选择最优动作目标网络稳定训练过程首先定义Q网络的结构from tensorflow.keras.layers import Dense, Input, Concatenate from tensorflow.keras.models import Model def create_q_network(state_dim, action_dim, hidden_units(256, 256)): state_input Input(shape(state_dim,)) action_input Input(shape(action_dim,)) # 状态处理分支 x Dense(hidden_units[0], activationrelu)(state_input) x Dense(hidden_units[1], activationrelu)(x) # 动作处理分支 y Dense(hidden_units[0], activationrelu)(action_input) y Dense(hidden_units[1], activationrelu)(y) # 合并 concat Concatenate()([x, y]) q_output Dense(1)(concat) return Model(inputs[state_input, action_input], outputsq_output)接下来是策略网络对于连续动作空间def create_policy_network(state_dim, action_dim, hidden_units(256, 256)): state_input Input(shape(state_dim,)) x Dense(hidden_units[0], activationrelu)(state_input) x Dense(hidden_units[1], activationrelu)(x) action_output Dense(action_dim, activationtanh)(x) # 假设动作在[-1,1]范围 return Model(inputsstate_input, outputsaction_output)3.2 保守Q学习CQL实现CQL的核心思想是在标准Q学习目标上增加一个正则项最小化非数据分布动作的Q值同时最大化数据分布动作的Q值。下面是关键实现import tensorflow as tf class CQLAgent: def __init__(self, state_dim, action_dim, gamma0.99, tau0.005, cql_weight1.0): self.state_dim state_dim self.action_dim action_dim self.gamma gamma # 折扣因子 self.tau tau # 目标网络更新系数 self.cql_weight cql_weight # CQL正则项权重 # 创建网络 self.q_net1 create_q_network(state_dim, action_dim) self.q_net2 create_q_network(state_dim, action_dim) self.target_q_net1 create_q_network(state_dim, action_dim) self.target_q_net2 create_q_network(state_dim, action_dim) self.policy_net create_policy_network(state_dim, action_dim) self.target_policy_net create_policy_network(state_dim, action_dim) # 初始化目标网络权重 self.update_target_networks(tau1.0) # 优化器 self.q_optimizer tf.keras.optimizers.Adam(learning_rate3e-4) self.policy_optimizer tf.keras.optimizers.Adam(learning_rate3e-4) def update_target_networks(self, tau): # 更新目标Q网络 for src, dest in zip(self.q_net1.variables, self.target_q_net1.variables): dest.assign(tau * src (1 - tau) * dest) for src, dest in zip(self.q_net2.variables, self.target_q_net2.variables): dest.assign(tau * src (1 - tau) * dest) # 更新目标策略网络 for src, dest in zip(self.policy_net.variables, self.target_policy_net.variables): dest.assign(tau * src (1 - tau) * dest) tf.function def train_step(self, batch): states, actions, rewards, next_states, dones batch # 计算目标Q值 next_actions self.target_policy_net(next_states) target_q1 self.target_q_net1([next_states, next_actions]) target_q2 self.target_q_net2([next_states, next_actions]) target_q tf.minimum(target_q1, target_q2) y rewards (1 - dones) * self.gamma * target_q # 标准Q学习损失 with tf.GradientTape(persistentTrue) as tape: current_q1 self.q_net1([states, actions]) current_q2 self.q_net2([states, actions]) q1_loss tf.reduce_mean(tf.square(current_q1 - y)) q2_loss tf.reduce_mean(tf.square(current_q2 - y)) # CQL正则项 random_actions tf.random.uniform(actions.shape, -1, 1) q1_rand self.q_net1([states, random_actions]) q2_rand self.q_net2([states, random_actions]) cql_loss1 tf.reduce_mean(q1_rand - current_q1) cql_loss2 tf.reduce_mean(q2_rand - current_q2) total_q1_loss q1_loss self.cql_weight * cql_loss1 total_q2_loss q2_loss self.cql_weight * cql_loss2 # 更新Q网络 q1_grads tape.gradient(total_q1_loss, self.q_net1.trainable_variables) self.q_optimizer.apply_gradients(zip(q1_grads, self.q_net1.trainable_variables)) q2_grads tape.gradient(total_q2_loss, self.q_net2.trainable_variables) self.q_optimizer.apply_gradients(zip(q2_grads, self.q_net2.trainable_variables)) # 策略学习 with tf.GradientTape() as tape: new_actions self.policy_net(states) new_q self.q_net1([states, new_actions]) policy_loss -tf.reduce_mean(new_q) policy_grads tape.gradient(policy_loss, self.policy_net.trainable_variables) self.policy_optimizer.apply_gradients(zip(policy_grads, self.policy_net.trainable_variables)) # 更新目标网络 self.update_target_networks(self.tau) return { q1_loss: q1_loss.numpy(), q2_loss: q2_loss.numpy(), cql_loss1: cql_loss1.numpy(), cql_loss2: cql_loss2.numpy(), policy_loss: policy_loss.numpy() }4. 训练流程与性能评估4.1 训练循环实现有了核心算法后我们需要实现完整的训练流程import numpy as np from tqdm import trange def train_agent(agent, train_data, val_data, epochs100, batch_size256): # 准备数据集 dataset tf.data.Dataset.from_tensor_slices(( train_data[states], train_data[actions], train_data[rewards], train_data[next_states], train_data[dones] )).shuffle(100000).batch(batch_size) # 训练循环 history {train: [], val: []} for epoch in trange(epochs): # 训练阶段 epoch_losses [] for batch in dataset: losses agent.train_step(batch) epoch_losses.append(losses) # 计算平均损失 avg_losses {k: np.mean([l[k] for l in epoch_losses]) for k in epoch_losses[0]} history[train].append(avg_losses) # 验证阶段可选 if val_data: val_batch ( val_data[states], val_data[actions], val_data[rewards], val_data[next_states], val_data[dones] ) val_losses agent.train_step(val_batch) # 注意这里不更新参数 history[val].append(val_losses) return history # 初始化agent state_dim train_data[states].shape[1] action_dim train_data[actions].shape[1] agent CQLAgent(state_dim, action_dim) # 开始训练 history train_agent(agent, train_data, val_data, epochs50)4.2 性能评估与可视化训练完成后我们需要评估模型性能。对于Pendulum环境我们可以计算平均回报def evaluate_policy(agent, env, n_episodes10): total_rewards [] for _ in range(n_episodes): state env.reset() done False episode_reward 0 while not done: action agent.policy_net(state[np.newaxis, :]) next_state, reward, done, _ env.step(action[0]) episode_reward reward state next_state total_rewards.append(episode_reward) return np.mean(total_rewards) env gym.make(Pendulum-v1) mean_reward evaluate_policy(agent, env) print(f平均回报: {mean_reward:.2f})我们还可以可视化训练过程中的损失变化import matplotlib.pyplot as plt def plot_training_history(history): plt.figure(figsize(12, 8)) # Q损失 plt.subplot(2, 2, 1) plt.plot([h[q1_loss] for h in history[train]], labelQ1 Loss) plt.plot([h[q2_loss] for h in history[train]], labelQ2 Loss) plt.title(Q Network Loss) plt.legend() # CQL损失 plt.subplot(2, 2, 2) plt.plot([h[cql_loss1] for h in history[train]], labelCQL Loss1) plt.plot([h[cql_loss2] for h in history[train]], labelCQL Loss2) plt.title(CQL Regularization Loss) plt.legend() # 策略损失 plt.subplot(2, 2, 3) plt.plot([h[policy_loss] for h in history[train]], labelPolicy Loss) plt.title(Policy Loss) plt.legend() plt.tight_layout() plt.show() plot_training_history(history)5. 实际应用与优化技巧5.1 部署到生产环境将训练好的离线RL模型部署到生产环境需要考虑几个关键点模型序列化保存和加载模型权重推理优化优化预测速度监控跟踪模型性能衰减保存和加载模型的示例代码# 保存模型 def save_agent(agent, path): agent.q_net1.save_weights(f{path}/q_net1.h5) agent.q_net2.save_weights(f{path}/q_net2.h5) agent.policy_net.save_weights(f{path}/policy_net.h5) # 加载模型 def load_agent(agent, path): agent.q_net1.load_weights(f{path}/q_net1.h5) agent.q_net2.load_weights(f{path}/q_net2.h5) agent.policy_net.load_weights(f{path}/policy_net.h5) # 使用示例 save_agent(agent, saved_models) new_agent CQLAgent(state_dim, action_dim) load_agent(new_agent, saved_models)5.2 性能优化技巧根据实践经验以下是提升离线RL性能的几个有效方法数据质量确保数据集覆盖足够多的状态-动作空间超参数调优特别是CQL权重和学习率集成学习结合多个Q网络的预测结果状态表征使用自编码器学习更好的状态特征一个简单的集成Q学习实现class EnsembleQNetwork: def __init__(self, state_dim, action_dim, n_networks5): self.networks [create_q_network(state_dim, action_dim) for _ in range(n_networks)] def __call__(self, states, actions): q_values [net([states, actions]) for net in self.networks] return tf.reduce_mean(q_values, axis0)在实际项目中我发现调整CQL权重对最终性能影响最大。当数据集覆盖不够全面时需要增大CQL权重以避免对OODOut-of-Distribution数据的高估而当数据集质量很高时可以适当减小CQL权重让模型更积极地探索高回报区域。
