深度学习算法中的强化学习（Reinforcement Learning）

pythonCopy codeimport gym
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.distributions import Categorical
# 创建一个强化学习智能体类
class Agent(nn.Module):
    def __init__(self, state_size, action_size):
        super(Agent, self).__init__()
        self.state_size = state_size
        self.action_size = action_size
        self.fc1 = nn.Linear(self.state_size, 128)
        self.fc2 = nn.Linear(128, self.action_size)
        self.optimizer = optim.Adam(self.parameters(), lr=0.001)
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.to(self.device)
    
    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.softmax(self.fc2(x), dim=-1)
        return x
    
    def select_action(self, state):
        state = torch.FloatTensor(state).to(self.device)
        action_probs = self.forward(state)
        action_dist = Categorical(action_probs)
        action = action_dist.sample()
        return action.item()
    
    def update_policy(self, rewards, log_probs):
        discounted_rewards = []
        cumulative_reward = 0
        for reward in rewards[::-1]:
            cumulative_reward = reward + 0.99 * cumulative_reward
            discounted_rewards.insert(0, cumulative_reward)
        discounted_rewards = torch.FloatTensor(discounted_rewards).to(self.device)
        discounted_rewards = (discounted_rewards - discounted_rewards.mean()) / (discounted_rewards.std() + 1e-8)
        policy_loss = []
        for log_prob, reward in zip(log_probs, discounted_rewards):
            policy_loss.append(-log_prob * reward)
        self.optimizer.zero_grad()
        policy_loss = torch.cat(policy_loss).sum()
        policy_loss.backward()
        self.optimizer.step()
# 创建一个游戏环境
env = gym.make('CartPole-v1')
state_size = env.observation_space.shape[0]
action_size = env.action_space.n
# 创建一个智能体
agent = Agent(state_size, action_size)
# 训练智能体
EPISODES = 1000
for episode in range(EPISODES):
    state = env.reset()
    done = False
    score = 0
    rewards = []
    log_probs = []
    while not done:
        action = agent.select_action(state)
        next_state, reward, done, _ = env.step(action)
        rewards.append(reward)
        log_probs.append(torch.log(agent.forward(torch.FloatTensor(state).to(agent.device)))[action])
        state = next_state
        score += reward
        if done:
            agent.update_policy(rewards, log_probs)
            print("episode: {}/{}, score: {}".format(episode + 1, EPISODES, score))
            break
# 使用训练好的智能体控制游戏
state = env.reset()
done = False
while not done:
    action = agent.select_action(state)
    state, _, done, _ = env.step(action)
    env.render()
# 保存训练好的模型
torch.save(agent.state_dict(), "model.pth")

pythonCopy codeimport gym
import numpy as np
from keras.models import Sequential
from keras.layers import Dense
from keras.optimizers import Adam
# 创建一个强化学习智能体类
class Agent:
    def __init__(self, state_size, action_size):
        self.state_size = state_size
        self.action_size = action_size
        self.memory = []
        self.gamma = 0.95  # 折扣因子，用于计算未来奖励的折扣值
        self.model = self.build_model()
    def build_model(self):
        model = Sequential()
        model.add(Dense(24, input_dim=self.state_size, activation='relu'))
        model.add(Dense(24, activation='relu'))
        model.add(Dense(self.action_size, activation='softmax'))
        model.compile(loss='categorical_crossentropy', optimizer=Adam(lr=0.001))
        return model
    def remember(self, state, action, reward, next_state, done):
        self.memory.append((state, action, reward, next_state, done))
    def act(self, state):
        state = np.reshape(state, [1, self.state_size])
        action_probs = self.model.predict(state)[0]
        return np.random.choice(self.action_size, p=action_probs)
    def replay(self, batch_size):
        minibatch = np.random.choice(len(self.memory), batch_size, replace=False)
        for index in minibatch:
            state, action, reward, next_state, done = self.memory[index]
            target = reward
            if not done:
                target = reward + self.gamma * np.amax(self.model.predict(np.reshape(next_state, [1, self.state_size]))[0])
            target_f = self.model.predict(np.reshape(state, [1, self.state_size]))
            target_f[0][action] = target
            self.model.fit(np.reshape(state, [1, self.state_size]), target_f, epochs=1, verbose=0)
    def load(self, name):
        self.model.load_weights(name)
    def save(self, name):
        self.model.save_weights(name)
# 创建一个游戏环境
env = gym.make('CartPole-v1')
state_size = env.observation_space.shape[0]
action_size = env.action_space.n
# 创建一个智能体
agent = Agent(state_size, action_size)
# 训练智能体
EPISODES = 1000
for episode in range(EPISODES):
    state = env.reset()
    done = False
    score = 0
    while not done:
        action = agent.act(state)
        next_state, reward, done, _ = env.step(action)
        agent.remember(state, action, reward, next_state, done)
        state = next_state
        score += reward
        if done:
            print("episode: {}/{}, score: {}".format(episode + 1, EPISODES, score))
            break
    if len(agent.memory) > 32:
        agent.replay(32)
# 保存训练好的模型
agent.save("model.h5")