深度学习学习履历——强化学习（rl） - ToB企服应用市场:ToB评测及商务社交产业平台

import numpy as np
import random
import torch
class SnakeGame:
def __init__(self, grid_size=10):
self.grid_size = grid_size
self.reset()
def reset(self):
self.snake = [(self.grid_size // 2, self.grid_size // 2)]
self.direction = (0, 1) # Start by moving right
self.food = self._place_food()
self.done = False
return self._get_observation()
def _place_food(self):
while True:
food = (random.randint(0, self.grid_size - 1), random.randint(0, self.grid_size - 1))
if food not in self.snake:
return food
def _get_observation(self):
obs = np.zeros((self.grid_size, self.grid_size))
for x, y in self.snake:
obs[x, y] = 1
obs[self.food[0], self.food[1]] = 2
return obs
def step(self, action):
if action == 0: # Up
self.direction = (-1, 0)
elif action == 1: # Down
self.direction = (1, 0)
elif action == 2: # Left
self.direction = (0, -1)
elif action == 3: # Right
self.direction = (0, 1)
head = (self.snake[0][0] + self.direction[0], self.snake[0][1] + self.direction[1])
if (head in self.snake) or (head[0] < 0 or head[0] >= self.grid_size) or (head[1] < 0 or head[1] >= self.grid_size):
self.done = True
return self._get_observation(), -10, self.done # Collision: negative reward
self.snake = [head] + self.snake[:-1]
if head == self.food:
self.snake.append(self.snake[-1]) # Grow
self.food = self._place_food()
return self._get_observation(), 10, self.done # Food eaten: positive reward
return self._get_observation(), 0, self.done # Neutral step

复制代码

import torch.nn as nn
import torch.optim as optim
class DQN(nn.Module):
def __init__(self, grid_size, num_actions):
super(DQN, self).__init__()
self.fc1 = nn.Linear(grid_size * grid_size, 128) # 第一层全连接层，输入大小为网格大小平方
self.fc2 = nn.Linear(128, 64) # 第二层全连接层
self.fc3 = nn.Linear(64, num_actions) # 输出层，输出动作的Q值
def forward(self, x):
batch_size = x.size(0) # 获取批次大小
x = x.view(batch_size, -1) # 将输入展平为 (batch_size, grid_size * grid_size)
x = torch.relu(self.fc1(x)) # 经过第一层全连接并ReLU激活
x = torch.relu(self.fc2(x)) # 经过第二层全连接并ReLU激活
return self.fc3(x) # 输出动作的Q值
# 计算输入张量的展平后的特征数目
def num_flat_features(self, x):
size = x.size()[1:] # 除去批次维度，计算特征维度的大小
num_features = 1
for s in size:
num_features *= s
return num_features
# Initialize DQN
grid_size = 10
num_actions = 4 # Up, Down, Left, Right
dqn = DQN(grid_size, num_actions)
optimizer = optim.Adam(dqn.parameters(), lr=0.001)
loss_fn = nn.MSELoss()

复制代码

def train(game, dqn, optimizer, loss_fn, episodes=1000, gamma=0.9):
for episode in range(episodes):
state = torch.tensor(game.reset(), dtype=torch.float32).unsqueeze(0) # 添加批次维度
total_reward = 0
while True:
q_values = dqn(state) # 预测当前状态的Q值
action = torch.argmax(q_values).item() # 选择Q值最大的动作
next_state, reward, done = game.step(action) # 执行动作并获得下一状态和奖励
next_state = torch.tensor(next_state, dtype=torch.float32).unsqueeze(0) # 将下一状态转换为张量并添加批次维度
with torch.no_grad():
max_next_q_values = torch.max(dqn(next_state)) # 计算下一状态的最大Q值
target = reward + gamma * max_next_q_values * (1 - int(done)) # 计算目标Q值
loss = loss_fn(q_values[0, action], target) # 计算损失
optimizer.zero_grad() # 梯度清零
loss.backward() # 反向传播
optimizer.step() # 更新网络参数
state = next_state # 更新当前状态
total_reward += reward
if done: # 如果游戏结束，退出循环
break
print(f"Episode {episode + 1}/{episodes} - Total Reward: {total_reward}") # 输出当前回合的总奖励
# Train the agent
game = SnakeGame(grid_size)
train(game, dqn, optimizer, loss_fn, episodes=1000)

复制代码

import matplotlib.pyplot as plt
def evaluate(game, dqn, episodes=10):
for episode in range(episodes):
state = torch.tensor(game.reset(), dtype=torch.float32).unsqueeze(0) # 添加批次维度
total_reward = 0
steps = 0
plt.figure() # 创建新的图像窗口
while True:
q_values = dqn(state) # 预测当前状态的Q值
action = torch.argmax(q_values).item() # 选择Q值最大的动作
next_state, reward, done = game.step(action) # 执行动作并获得下一状态和奖励
state = torch.tensor(next_state, dtype=torch.float32).unsqueeze(0) # 将下一状态转换为张量并添加批次维度
total_reward += reward
steps += 1
# Visualization (Optional)
plt.imshow(state.squeeze(0).numpy()) # 去掉批次维度进行显示
plt.axis('off') # 隐藏坐标轴
plt.pause(0.1) # 短暂停留以更新视图
if done:
print(f"Episode {episode + 1}/{episodes} - Total Reward: {total_reward} - Steps: {steps}")
break
plt.close() # 关闭图像窗口

复制代码

import numpy as np
import random
import torch
import torch.nn as nn
import torch.optim as optim
import matplotlib.pyplot as plt
# 定义贪吃蛇游戏环境
class SnakeGame:
def __init__(self, grid_size=10):
self.grid_size = grid_size
self.reset()
# 重置游戏状态，初始化蛇的位置和食物的位置
def reset(self):
self.snake = [(self.grid_size // 2, self.grid_size // 2)] # 蛇的初始位置在网格的中央
self.direction = (0, 1) # 蛇的初始移动方向为向右
self.food = self._place_food() # 随机放置食物
self.done = False
return self._get_observation()
# 随机生成一个食物位置，确保不与蛇的身体重叠
def _place_food(self):
while True:
food = (random.randint(0, self.grid_size - 1), random.randint(0, self.grid_size - 1))
if food not in self.snake:
return food
# 获取当前游戏状态，返回一个网格表示的观察值
def _get_observation(self):
obs = np.zeros((self.grid_size, self.grid_size)) # 初始化一个全0的网格
for x, y in self.snake:
obs[x, y] = 1 # 用1表示蛇的身体
obs[self.food[0], self.food[1]] = 2 # 用2表示食物的位置
return obs
# 游戏的主要逻辑：根据动作更新蛇的状态
def step(self, action):
if action == 0: # 上
self.direction = (-1, 0)
elif action == 1: # 下
self.direction = (1, 0)
elif action == 2: # 左
self.direction = (0, -1)
elif action == 3: # 右
self.direction = (0, 1)
# 更新蛇头的位置
head = (self.snake[0][0] + self.direction[0], self.snake[0][1] + self.direction[1])
# 如果蛇撞到自己或墙壁，游戏结束
if (head in self.snake) or (head[0] < 0 or head[0] >= self.grid_size) or (head[1] < 0 or head[1] >= self.grid_size):
self.done = True
return self._get_observation(), -10, self.done # 碰撞：给予负奖励
# 更新蛇的身体
self.snake = [head] + self.snake[:-1]
# 如果蛇吃到食物
if head == self.food:
self.snake.append(self.snake[-1]) # 蛇的身体增长
self.food = self._place_food() # 重新放置食物
return self._get_observation(), 10, self.done # 吃到食物：给予正奖励
return self._get_observation(), 0, self.done # 普通移动：奖励为0
# 定义DQN（深度Q网络）模型
class DQN(nn.Module):
def __init__(self, grid_size, num_actions):
super(DQN, self).__init__()
self.fc1 = nn.Linear(grid_size * grid_size, 128) # 第一层全连接层，输入大小为网格大小平方
self.fc2 = nn.Linear(128, 64) # 第二层全连接层
self.fc3 = nn.Linear(64, num_actions) # 输出层，输出动作的Q值
def forward(self, x):
batch_size = x.size(0) # 获取批次大小
x = x.view(batch_size, -1) # 将输入展平为 (batch_size, grid_size * grid_size)
x = torch.relu(self.fc1(x)) # 经过第一层全连接并ReLU激活
x = torch.relu(self.fc2(x)) # 经过第二层全连接并ReLU激活
return self.fc3(x) # 输出动作的Q值
# 计算输入张量的展平后的特征数目
def num_flat_features(self, x):
size = x.size()[1:] # 除去批次维度，计算特征维度的大小
num_features = 1
for s in size:
num_features *= s
return num_features
# 初始化DQN
grid_size = 10
num_actions = 4 # 上、下、左、右四个动作
dqn = DQN(grid_size, num_actions)
optimizer = optim.Adam(dqn.parameters(), lr=0.00001) # 使用Adam优化器
loss_fn = nn.MSELoss() # 使用均方误差损失函数
# 训练智能体
def train(game, dqn, optimizer, loss_fn, episodes=1000, gamma=0.9):
for episode in range(episodes):
state = torch.tensor(game.reset(), dtype=torch.float32).unsqueeze(0) # 添加批次维度
total_reward = 0
while True:
q_values = dqn(state) # 预测当前状态的Q值
action = torch.argmax(q_values).item() # 选择Q值最大的动作
next_state, reward, done = game.step(action) # 执行动作并获得下一状态和奖励
next_state = torch.tensor(next_state, dtype=torch.float32).unsqueeze(0) # 将下一状态转换为张量并添加批次维度
with torch.no_grad():
max_next_q_values = torch.max(dqn(next_state)) # 计算下一状态的最大Q值
target = reward + gamma * max_next_q_values * (1 - int(done)) # 计算目标Q值
loss = loss_fn(q_values[0, action], target) # 计算损失
optimizer.zero_grad() # 梯度清零
loss.backward() # 反向传播
optimizer.step() # 更新网络参数
state = next_state # 更新当前状态
total_reward += reward
if done: # 如果游戏结束，退出循环
break
print(f"Episode {episode + 1}/{episodes} - Total Reward: {total_reward}") # 输出当前回合的总奖励
# 训练智能体
game = SnakeGame(grid_size)
train(game, dqn, optimizer, loss_fn, episodes=10000) # 开始训练
# Step 5: Evaluate the Agent
def evaluate(game, dqn, episodes=10):
for episode in range(episodes):
state = torch.tensor(game.reset(), dtype=torch.float32).unsqueeze(0) # 添加批次维度
total_reward = 0
steps = 0
plt.figure() # 创建新的图像窗口
while True:
q_values = dqn(state) # 预测当前状态的Q值
action = torch.argmax(q_values).item() # 选择Q值最大的动作
next_state, reward, done = game.step(action) # 执行动作并获得下一状态和奖励
state = torch.tensor(next_state, dtype=torch.float32).unsqueeze(0) # 将下一状态转换为张量并添加批次维度
total_reward += reward
steps += 1
# Visualization (Optional)
plt.imshow(state.squeeze(0).numpy()) # 去掉批次维度进行显示
plt.axis('off') # 隐藏坐标轴
plt.pause(0.1) # 短暂停留以更新视图
if done:
print(f"Episode {episode + 1}/{episodes} - Total Reward: {total_reward} - Steps: {steps}")
break
plt.close() # 关闭图像窗口
# Evaluate the trained agent
evaluate(game, dqn)

复制代码

optimizer = optim.Adam(dqn.parameters(), lr=0.0005) # 例如，减小学习率

复制代码

train(game, dqn, optimizer, loss_fn, episodes=1000, gamma=0.95) # 例如，稍微降低 γ

复制代码

epsilon = 0.9 # 初始探索率
decay_rate = 0.995 # 衰减率
for episode in range(episodes):
epsilon = max(epsilon * decay_rate, 0.01) # 确保 epsilon 最小为 0.01
# 其余代码保持不变

复制代码

from collections import deque
replay_memory = deque(maxlen=10000) # 扩展经验回放池的容量
def train(game, dqn, optimizer, loss_fn, episodes=1000, gamma=0.99, batch_size=64):
for episode in range(episodes):
state = torch.tensor(game.reset(), dtype=torch.float32)
total_reward = 0
while True:
if random.random() < epsilon:
action = random.randint(0, num_actions - 1)
else:
q_values = dqn(state)
action = torch.argmax(q_values).item()
next_state, reward, done = game.step(action)
next_state = torch.tensor(next_state, dtype=torch.float32)
replay_memory.append((state, action, reward, next_state, done))
# 开始经验回放训练
if len(replay_memory) > batch_size:
minibatch = random.sample(replay_memory, batch_size)
for s, a, r, ns, d in minibatch:
q_values = dqn(s)
with torch.no_grad():
max_next_q_values = torch.max(dqn(ns))
target = r + gamma * max_next_q_values * (1 - int(d))
loss = loss_fn(q_values[a], target)
optimizer.zero_grad()
loss.backward()
optimizer.step()
state = next_state
total_reward += reward
if done:
break
print(f"Episode {episode + 1}/{episodes} - Total Reward: {total_reward}")

复制代码

train(game, dqn, optimizer, loss_fn, episodes=5000) # 增加训练轮数

复制代码

def step(self, action):
if action == 0: # Up
self.direction = (-1, 0)
elif action == 1: # Down
self.direction = (1, 0)
elif action == 2: # Left
self.direction = (0, -1)
elif action == 3: # Right
self.direction = (0, 1)
head = (self.snake[0][0] + self.direction[0], self.snake[0][1] + self.direction[1])
if (head in self.snake) or (head[0] < 0 or head[0] >= self.grid_size) or (head[1] < 0 or head[1] >= self.grid_size):
self.done = True
return self._get_observation(), -10, self.done # Collision: negative reward
self.snake = [head] + self.snake[:-1]
if head == self.food:
self.snake.append(self.snake[-1]) # Grow
self.food = self._place_food()
return self._get_observation(), 10, self.done # Food eaten: positive reward
# 距离食物近一步，给出微小奖励
dist = abs(head[0] - self.food[0]) + abs(head[1] - self.food[1])
reward = 1.0 / (dist + 1) # 距离越近，奖励越大
return self._get_observation(), reward, self.done # Neutral step

复制代码