Python脚本数据强化学习Gym环境如何用

wen 实用脚本 2026-07-18 2

本文目录导读：

基础环境设置
完整DQN强化学习示例
数据采集与处理
自定义环境
高级功能示例
常用Gym环境列表

我来详细解释如何在Python中使用强化学习Gym环境进行数据强化学习。

基础环境设置

安装必要的库

# 安装命令
# pip install gym
# pip install numpy
# pip install matplotlib
# pip install gym[classic_control]  # 包含经典控制环境

基本Gym环境使用框架

import gym
import numpy as np
import matplotlib.pyplot as plt
# 创建环境
env = gym.make('CartPole-v1')
# 重置环境
observation = env.reset()
print(f"初始观测: {observation}")
# 基本交互循环
for episode in range(5):
    observation = env.reset()
    total_reward = 0
    done = False
    while not done:
        # 可视化（可选）
        # env.render()
        # 随机选择动作
        action = env.action_space.sample()
        # 执行动作
        observation, reward, done, info = env.step(action)
        total_reward += reward
        print(f"Episode: {episode}, Step: {env._elapsed_steps}, "
              f"Action: {action}, Reward: {reward}, Done: {done}")
    print(f"Episode {episode} 总奖励: {total_reward}\n")
env.close()

完整DQN强化学习示例

import gym
import numpy as np
import tensorflow as tf
from collections import deque
import random
class DQNAgent:
    def __init__(self, state_size, action_size):
        self.state_size = state_size
        self.action_size = action_size
        self.memory = deque(maxlen=2000)
        self.gamma = 0.95
        self.epsilon = 1.0
        self.epsilon_min = 0.01
        self.epsilon_decay = 0.995
        self.learning_rate = 0.001
        self.model = self._build_model()
    def _build_model(self):
        model = tf.keras.Sequential([
            tf.keras.layers.Dense(24, input_dim=self.state_size, activation='relu'),
            tf.keras.layers.Dense(24, activation='relu'),
            tf.keras.layers.Dense(self.action_size, activation='linear')
        ])
        model.compile(loss='mse', optimizer=tf.keras.optimizers.Adam(lr=self.learning_rate))
        return model
    def remember(self, state, action, reward, next_state, done):
        self.memory.append((state, action, reward, next_state, done))
    def act(self, state):
        if np.random.rand() <= self.epsilon:
            return random.randrange(self.action_size)
        act_values = self.model.predict(state)
        return np.argmax(act_values[0])
    def replay(self, batch_size):
        minibatch = random.sample(self.memory, batch_size)
        for state, action, reward, next_state, done in minibatch:
            target = reward
            if not done:
                target = reward + self.gamma * np.amax(self.model.predict(next_state)[0])
            target_f = self.model.predict(state)
            target_f[0][action] = target
            self.model.fit(state, target_f, epochs=1, verbose=0)
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay
# 训练主函数
def train_dqn():
    env = gym.make('CartPole-v1')
    state_size = env.observation_space.shape[0]
    action_size = env.action_space.n
    agent = DQNAgent(state_size, action_size)
    episodes = 100
    batch_size = 32
    scores = []
    for e in range(episodes):
        state = env.reset()
        state = np.reshape(state, [1, state_size])
        total_reward = 0
        for time in range(500):
            # env.render()  # 如需可视化取消注释
            action = agent.act(state)
            next_state, reward, done, _ = env.step(action)
            next_state = np.reshape(next_state, [1, state_size])
            agent.remember(state, action, reward, next_state, done)
            state = next_state
            total_reward += reward
            if done:
                print(f"Episode: {e+1}, Score: {time}, Epsilon: {agent.epsilon:.2f}")
                scores.append(total_reward)
                break
            if len(agent.memory) > batch_size:
                agent.replay(batch_size)
    env.close()
    return scores
# 运行训练
scores = train_dqn()
# 绘制结果
plt.plot(scores)
plt.xlabel('Episode')
plt.ylabel('Score')'DQN Training Progress')
plt.show()

数据采集与处理

import gym
import numpy as np
import pandas as pd
class DataCollector:
    def __init__(self, env_name='CartPole-v1'):
        self.env = gym.make(env_name)
        self.data = []
    def collect_random_data(self, episodes=10):
        for episode in range(episodes):
            state = self.env.reset()
            done = False
            episode_data = []
            while not done:
                # 随机策略
                action = self.env.action_space.sample()
                next_state, reward, done, info = self.env.step(action)
                # 记录数据
                episode_data.append({
                    'episode': episode,
                    'state': state.tolist(),
                    'action': action,
                    'reward': reward,
                    'next_state': next_state.tolist(),
                    'done': done
                })
                state = next_state
            self.data.extend(episode_data)
        self.env.close()
        return pd.DataFrame(self.data)
    def save_to_csv(self, filename='rl_data.csv'):
        df = pd.DataFrame(self.data)
        df.to_csv(filename, index=False)
        print(f"数据已保存到 {filename}")
    def analyze_data(self):
        df = pd.DataFrame(self.data)
        print("数据统计:")
        print(f"总步数: {len(df)}")
        print(f"平均奖励: {df['reward'].mean():.2f}")
        print(f"最大奖励: {df['reward'].max()}")
        print(f"最小奖励: {df['reward'].min()}")
# 使用数据收集器
collector = DataCollector('CartPole-v1')
df = collector.collect_random_data(episodes=5)
collector.analyze_data()
collector.save_to_csv('cartpole_data.csv')

自定义环境

import gym
from gym import spaces
import numpy as np
class CustomEnv(gym.Env):
    def __init__(self):
        super(CustomEnv, self).__init__()
        # 定义动作空间和观测空间
        self.action_space = spaces.Discrete(2)  # 0 或 1
        self.observation_space = spaces.Box(
            low=-np.inf, 
            high=np.inf, 
            shape=(4,), 
            dtype=np.float32
        )
        self.state = None
        self.step_count = 0
    def reset(self):
        # 重置环境状态
        self.state = np.random.uniform(-1, 1, 4)
        self.step_count = 0
        return self.state
    def step(self, action):
        # 执行动作
        self.step_count += 1
        # 简单的动态系统
        if action == 0:
            self.state += np.random.normal(0, 0.1, 4)
        else:
            self.state += np.random.normal(0, 0.2, 4)
        # 计算奖励
        reward = -np.sum(self.state ** 2)
        # 判断是否结束
        done = self.step_count >= 200 or abs(reward) > 10
        info = {}
        return self.state, reward, done, info
    def render(self, mode='human'):
        print(f"State: {self.state}, Step: {self.step_count}")
# 测试自定义环境
env = CustomEnv()
obs = env.reset()
print(f"初始状态: {obs}")
for _ in range(10):
    action = env.action_space.sample()
    obs, reward, done, info = env.step(action)
    print(f"状态: {obs}, 奖励: {reward:.2f}, 完成: {done}")
    if done:
        break
env.close()

高级功能示例

import gym
import numpy as np
from gym.wrappers import Monitor
# 环境包装器（Wrappers）
def create_wrapped_env():
    env = gym.make('CartPole-v1')
    # 记录视频
    env = Monitor(env, './video', force=True)
    # 自定义包装
    class RewardScaler(gym.Wrapper):
        def step(self, action):
            obs, reward, done, info = self.env.step(action)
            # 缩放奖励
            scaled_reward = reward * 0.1
            return obs, scaled_reward, done, info
    env = RewardScaler(env)
    return env
# 多环境并行
def create_multiple_envs(num_envs=4):
    from concurrent.futures import ThreadPoolExecutor
    def run_env(env_id):
        env = gym.make('CartPole-v1')
        total_reward = 0
        observation = env.reset()
        for _ in range(100):
            action = env.action_space.sample()
            observation, reward, done, info = env.step(action)
            total_reward += reward
            if done:
                break
        env.close()
        return total_reward
    with ThreadPoolExecutor(max_workers=num_envs) as executor:
        results = list(executor.map(run_env, range(num_envs)))
    return results
# 运行示例
if __name__ == "__main__":
    # 1. 基础环境测试
    print("=== 基础环境测试 ===")
    env = gym.make('CartPole-v1')
    print(f"动作空间: {env.action_space}")
    print(f"观测空间: {env.observation_space}")
    # 2. 多环境并行
    print("\n=== 多环境并行测试 ===")
    results = create_multiple_envs(num_envs=4)
    print(f"4个环境的奖励: {results}")
    # 3. 包装环境
    print("\n=== 包装环境测试 ===")
    wrapped_env = create_wrapped_env()
    obs = wrapped_env.reset()
    print(f"包装后的初始观测: {obs}")
    wrapped_env.close()

常用Gym环境列表

# 经典控制问题
envs = [
    'CartPole-v1',        # 平衡杆
    'MountainCar-v0',     # 爬山车
    'Acrobot-v1',         # 双摆
    'Pendulum-v0',        # 摆锤
]
# Atari游戏
atari_envs = [
    'SpaceInvaders-v0',   # 太空侵略者
    'Pong-v0',            # 乒乓球
    'Breakout-v0',        # 打砖块
]
# 连续控制
continuous_envs = [
    'LunarLanderContinuous-v2',  # 月球着陆器(连续)
    'BipedalWalker-v3',          # 双足行走
]

环境交互: reset() → step() 循环
数据收集: 记录(state, action, reward, next_state, done)
经验回放: 存储经验样本用于训练
探索策略: ε-greedy探索
环境包装: 使用Wrapper修改环境行为

这个框架可以扩展到任何Gym兼容的环境,适用于各种强化学习算法。