Unity + Pytorch utilizando técnicas de Deep Q-Learning para la creación de agentes inteligentes en videojuegos.
Jose Luis Aguilera Luzanía, Jesus Armando Báez Camacho, Francisco Javier Castro Márquez.
Universidad de Sonora, Licenciatura en ciencias de la computación, Hermosillo, Sonora
1. Introducción
Aprender interactuando con el entorno es probablemente lo primero que se nos viene a la mente cuando pensamos acerca de la naturaleza del aprendizaje. Por ejemplo, los niños tienden a tener comportamientos erráticos que van desapareciendo a medida que crecen, pero son por medio de estos que los niños interactúan con el entorno, aprenden que comportamiento puede ser beneficioso, acerca de las consecuencias de las acciones y que deben hacer para conseguir ciertas metas. Este paradigma en Machine Learning es conocido como aprendizaje por refuerzo (Reinforcement Learning o RL) y es el tema central de este trabajo, en el que utilizaremos un motor de videojuegos para crear una IA que aprenda por medio de algoritmos de Reinforcement Learning a moverse en un circuito de carreras.
¿Qué es Reinforcement Learning? y ¿Cómo aplicarlo?
Reinforcement Learning es aprender qué se debe hacer interactuando con el entorno para maximizar una recompensa numérica. El sujeto que quiere maximizar la recompensa no sabe que acciones debe tomar, pero sabe que acciones puede tomar y debe descubrir cuáles acciones tienen como consecuencia una mayor recompensa por medio de prueba y error. También es importante reconocer que no todas las acciones solo tienen una recompensa inmediata, algunas también tienen recompensas a largo plazo.
Para poder resolver problemas de Rinforcement Learning vamos a ver el concepto de Markov Decision Processes o MDP. En un MDP tenemos a un tomador de decisiones al que llamaremos agente (agent), este interactúa con el entorno en el que se encuentra (environment o env). En cada paso (step), el agente obtiene una representación del estado del entorno (state). Con esta representación, el agente tomará alguna acción (action). El entorno entonces cambiará a un nuevo estado y el agente adquiere una recompensa como consecuencia de sus acciones previas. La ciclo de recibir el estado del entorno, tomar una acción y obtener una recompensa es también llamado trayectoria (trajectory). Si esta explicación de lo que son los MDP parece familiar, es porque es una perfecta descripción de lo que queremos hacer al usar Reinforcement Learning.
Componentes de un MDP:
Agente.
Entorno.
Estado.
Recompensa.
Notación de un MDP:
En un MDP tenemos un conjunto de estados, un conjunto de acciones, y un conjunto de recompensas. Se asume que todos los conjuntos son finitos.
En cada paso de tiempo el agente recibe una representación el estado del entorno. Basado en este estado, el agente selecciona una acción. Esto nos da una pareja de estado-acción (St,At).
Entonces el tiempo incrementa al siguiente paso de tiempo t + 1 y el entorno cambia al nuevo estado St+1 ∈ S. Para este tiempo, el agente recibe una recompensa Rt+1 ∈ R para la acción At tomada en el estado St.
Podemos pensar en este proceso de recibir una recompensa como una función arbitraría que mapea las parejas estado-acción a recompensas. Entonces para cada tiempo t, tenemos:
f(St,At) = Rt+1
También podemos representar la trayectoria como:
S0,A0,R1, S1,A1,R2,⋯
Este diagrama muestra claramente la idea detrás de un MDP:
Vamos a describir el proceso que se muestra en el diagrama:
1. Para el tiempo el entorno se encuentra en el estado St.
2. El agente observa el estado del entorno y selecciona una acción At.
3. El entorno cambia al estado St+1 y el agente adquiere una recompensa Rt+1.
4. Este proceso se repite de nuevo para un paso en tiempo t + 1.
Nota: t + 1 ya no es el paso en el siguiente tiempo, ahora es el presente. Cuando cruzamos la línea punteada abajo a la izquierda, el diagrama muestra como t + 1 se transforma en el paso actual t, entonces St+1 Rt+1 ahora se muestran como St y Rt respectivamente.
¿Qué es Deep QLearning?
Para hablar de Deep QLearning primero debemos saber un poco sobre QLearning. QLearning es un algoritmo de Reinforcement Learning que busca optimizar una función para la toma de decisiones similar al ejemplo de un niño aprendiendo a interactuar con el entorno, al inicio el algoritmo opta por la exploración pero a medida que avanza y aprende sobre el entorno y las acciones que escoge comienza a ser más “codicioso” y tiende a explorar menos, intentando maximizar la recompensas obtenidas con todo lo previamente aprendido sin arriesgarse tanto al explorar. Sin embargo, esta técnica no suele adaptarse bien a entornos complejos, pero conocemos una herramienta que si lo hace, las redes neuronales. De ahí su nombre Deep QLearning, donde utilizaremos redes neuronales para aproximar los valores de la función .
¿Qué vamos a hacer?
Como el título sugiere, vamos a utilizar Unity y ML-Agents para crear una IA de videojuegos, inspirados en los juegos de carreras 3D, decidimos que el objetivo de este proyecto es crear una IA que pueda manejar por un circuito de carreras sin ser explícitamente programada para ello, solo se programaran sus acciones y recompensas. Por ejemplo, seguir en el camino aumentará las recompensas obtenidas, mientras que salir del circuito terminará inmediatamente con la simulación, pero es la IA quien tendrá que descubrir esto.
2. Primeros pasos con Unity
En esta sección se crea la estructura desde la vista de Unity.
- Assets utilizados.
- Plugins utilizados de Unity.
- Estructura del proyecto en Unity.
- Estructura del environment.
- Componentes.
- Estructura del Agente.
- Componentes.
- Comportamiento.
Assets utilizados:
Para la simulación del entorno usaremos los assets de la página de Kenney, en específico el Racing Kit.
Paquetes del proyecto.
Los paquetes del proyecto pueden ser encontrados a través del Package Manager.
Estructura del proyecto en Unity.
Al momento de abrir el proyecto en Unity debemos ir a la ventana del proyecto y buscar entre los assets la carpeta que se llama Scenes y abrir hacer doble clic en el archivo que se llama TrainingScene.
Esto cargará una nueva escena en Unity.
Editor de Unity:
Jerarquía con la estructura de la escena:
Estructura del enviroment.
Como se observa al cargar la escena TrainingScene en Unity el entorno es el siguiente:
Estructura del Agente.
Es hora de ver a nuestro agente, en este caso nuestro agente es un carro de carreras azul.
Los objetos anidados y sus componentes son:
Los componentes del agente en Unity son:
CarAgent(Script):
using UnityEngine;
using Unity.MLAgents;
using Unity.MLAgents.Actuators;
public class CarAgent : Agent
{
/// <summary>
/// Steer force.
/// </summary>
[Header("Movement")] [SerializeField] private float steerForce;
/// <summary>
/// Motor force.
/// </summary>
[SerializeField] private float motorForce;
/// <summary>
/// The starting position where the agent will start in each episode.
/// </summary>
[Header("Start position")] [SerializeField] private Transform initialPosition;
/// <summary>
/// WheelCollider component for the Wheel Front Left.
/// </summary>
[Header("Wheels")] [SerializeField] private WheelCollider wheelFrontLeft;
/// <summary>
/// WheelCollider component for the Wheel Front Right.
/// </summary>
[SerializeField] private WheelCollider wheelFrontRight;
/// <summary>
/// WheelCollider component for the Wheel Back Left.
/// </summary>
[SerializeField] private WheelCollider wheelBackLeft;
/// <summary>
/// WheelCollider component for the Wheel Back Right.
/// </summary>
[SerializeField] private WheelCollider wheelBackRight;
/// <summary>
/// Reference to this transform component.
/// </summary>
private Transform _transform;
/// <summary>
/// Function that is called when starting a new episode and
/// is responsible for restarting the agent.
/// </summary>
public override void OnEpisodeBegin()
{
_transform = transform;
_transform.position = initialPosition.position;
_transform.eulerAngles = new Vector3(0f, 90f, 0f);
}
/// <summary>
/// Function that is in charge of determining the behavior depending on the actions it receives.
/// Actions: 0: go straight | 1: Steer to the right | 2: Steer to the left.
/// </summary>
/// <param name="actions"></param>
public override void OnActionReceived(ActionBuffers actions)
{
int action = actions.DiscreteActions[0];
float h, v = -motorForce;
if (action <= 1) h = action;
else h = -1;
wheelBackLeft.motorTorque = v;
wheelBackRight.motorTorque = v;
wheelFrontLeft.steerAngle = h * steerForce;
wheelFrontRight.steerAngle = h * steerForce;
}
/// <summary>
/// Function that is invoked when the physics engine detects a collision
/// between two or more objects.
/// </summary>
/// <param name="other">An object that contains all the information about the collision.</param>
private void OnTriggerEnter(Collider other)
{
if (other.gameObject.CompareTag("Target"))
SetReward(+1f);
if (other.gameObject.CompareTag("Wall"))
{
SetReward(-5f);
EndEpisode();
}
}
}Vista del código desde Unity Editor:
3. Dependencias de Python
Se cargan las dependencias necesarias para las operaciones de la red neuronal, y la comunicación entre Python y Unity.
- Librerias a utilizar.
- Importar las librerías.
import matplotlib.pyplot as plt
import numpy as np
import sys
import torch
import random
from mlagents_envs.environment import UnityEnvironment
from mlagents_envs.environment import ActionTuple, BaseEnv
from typing import Tuple
from typing import NamedTuple, List
from typing import Dict
from math import floor4. Primeros pasos con ML-Agents
Entrenamiento de nuestro agente para ganar algún premio por hacer algo innovador, y conexión entre Python y Unity.
- ¿Cómo conectar ML-Agents con Unity a través de Python?
- ¿Cómo obtener información de los Agentes?
- La primera observación visual del Agente.
- Probar el seguimiento de un Agente.
El código para conectar nuestro entorno (que llamaremos env) es el siguiente:
env = UnityEnvironment(file_name = None, base_port=5004)
env.reset() # <--- Reinicia la simulación.Obtener información de los Agentes:
# Nombre del primer comportamiento.
behavior_name = list(env.behavior_specs)[0]
print(f"Behavior name: {behavior_name} \n")
# Específicaciones del comportamiento.
spec = env.behavior_specs[behavior_name]
print(spec.observation_specs)
# Número de observaciones.
print(f"\nNúmer of observations: {len(spec.observation_specs)}")
# ¿Hay una observación visual?
vis_obs = any(len(spec.shape) == 3 for spec in spec.observation_specs)
print(f"\nVisual observation: {vis_obs}")
# ¿La acción es continua o multi discreta?
if spec.action_spec.continuous_size > 0:
print(f"\nThere are {spec.action_spec.continuous_size} continuous actions.")
if spec.action_spec.is_discrete():
print(f"\nThere are {spec.action_spec.discrete_size} discrete actions.")Primera observación visual del Agente:
decision_steps, terminal_steps = env.get_steps(behavior_name)
env.set_actions(behavior_name, spec.action_spec.empty_action(len(decision_steps)))
env.step()Se usa matplotlib para reconstruir la imagen obtenida de las observaciones y mostrarla:
%matplotlib inline
for index, obs_spec in enumerate(spec.observation_specs):
if len(obs_spec.shape) == 3:
print("Here is the first visual observation.")
print(decision_steps.obs[index][0,:,:,:].shape)
plt.imshow(decision_steps.obs[index][0,:,:,:])
plt.show()
for index, obs_spec in enumerate(spec.observation_specs):
if len(obs_spec.shape) == 1:
print(f"First vector observations: {decision_steps.obs[index][0,:]}")
5. Definiciones de VisualQNetwork, Experiencia y Entrenamiento.
Definición de las funciones necesarias para implementar el Deep Q-Learning.
- Definición de la clase VisualQNetwork.
- Definición y explicación.
- Definición de la clase Experience.
- Definición y explicación.
- Definición de la clase Trainer.
- Definición y explicación.
Definición de la clase VisualQNetwork:
class VisualQNetwork(torch.nn.Module):
def __init__(
self,
input_shape: Tuple[int, int, int],
encoding_size: int,
output_size: int
):
"""
Crea una red neuronal que toma como input un batch de imagenes
(tensor tridimensional) y da como salida un batch de outputs
(tensor unidimensional).
"""
super(VisualQNetwork, self).__init__()
height = input_shape[0]
width = input_shape[1]
initial_channels = input_shape[2]
conv_1_hw = self.conv_output_shape((height, width), 8, 4)
conv_2_hw = self.conv_output_shape(conv_1_hw, 4, 2)
self.final_flat = conv_2_hw[0] * conv_2_hw[1] * 32
self.conv1 = torch.nn.Conv2d(initial_channels, 16, [8, 8], [4, 4])
self.batchNorm1 = torch.nn.BatchNorm2d(16)
self.conv2 = torch.nn.Conv2d(16, 32, [4, 4], [2, 2])
self.batchNorm2 = torch.nn.BatchNorm2d(32)
self.dense1 = torch.nn.Linear(self.final_flat, encoding_size)
self.dense2 = torch.nn.Linear(encoding_size,output_size)
def forward(self, visual_obs: torch.tensor):
visual_obs = visual_obs.permute(0, 3, 1, 2)
conv_1 = torch.relu(self.batchNorm1(self.conv1(visual_obs)))
conv_2 = torch.relu(self.batchNorm2(self.conv2(conv_1)))
hidden = self.dense1(conv_2.reshape([-1, self.final_flat]))
hidden = torch.relu(hidden)
hidden = self.dense2(hidden)
return hidden
@staticmethod
def conv_output_shape(
h_w: Tuple[int, int],
kernel_size: int = 1,
stride: int = 1,
pad: int = 0,
dilation: int = 1,
):
"""
Calcula la altura y el ancho de la salida de una convolution layer.
"""
h = floor(
((h_w[0] + (2 * pad) - (dilation * (kernel_size - 1)) - 1) / stride) + 1
)
w = floor(
((h_w[1] + (2 * pad) - (dilation * (kernel_size - 1)) - 1) / stride) + 1
)
return h, wDefinición de la clase Experience
class Experience(NamedTuple):
"""
Una experiencia contiene los datos de la transición de un agente.
-Observation
-Action
-Reward
-Done flag
-Next Observation
"""
obs: np.ndarray
action: np.ndarray
reward: float
done: bool
next_obs: np.ndarray
# Una trayectoria es una secuencia ordenada de experiencias.
Trajectory = List[Experience]
# Un búfer es una lista desordenada de experiencias de múltiples trayectorias.
Buffer = List[Experience]Definición de la clase Trainer:
class Trainer:
@staticmethod
def generate_trajectories(
env: BaseEnv, q_net: VisualQNetwork, buffer_size: int, epsilon: float
):
"""
Dado un Unity Environment y una Q-Network, este método generará un búfer de
Experiencias obtenidas al ejecutar el Environment con la Policy derivada de
la Q-Network.
:param BaseEnv: El UnityEnvironment usado.
:param q_net: La Q-Network usada para recolectar la data.
:param buffer_size: El tamaño mínimo del buffer que devolverá este método.
:param epsilon: Agregará una variable normal aleatoria con desviación estándar.
epsilon a los valores de la Q-Network para fomentar la exploración.
:returns: una Tuple que contiene el búfer creado y el promedio acumulado de los
agentes obtenidos.
"""
# Crear un buffer vacio.
buffer: Buffer = []
# Reiniciar el environment.
env.reset()
# Leer y guardar el nombre del comportamiento del env.
behavior_name = list(env.behavior_specs)[0]
# Leer y guardar las específicaciones del comportamiento del env.
spec = env.behavior_specs[behavior_name]
# Mapping AgentID -> trajectories. Ayuda a crear trayectorias para cada agente.
dict_trajectories_from_agent: Dict[int, Trajectory] = {}
# Mapping AgentId -> "last observation".
dict_last_obs_from_agent: Dict[int, np.ndarray] = {}
# Mapping AgentId -> "last action".
dict_last_action_from_agent: Dict[int, np.ndarray] = {}
# Mapping AgentId -> cumulative reward (Solo para reporte).
dict_cumulative_reward_from_agent: Dict[int, float] = {}
# Lista que guarda las recompensas acumuladas hasta el momento.
cumulative_rewards: List[float] = []
# Mientras no exista suficiente data en el buffer.
while len(buffer) < buffer_size:
# Obtener los Decision Steps y Terminal Steps del agente.
decision_steps, terminal_steps = env.get_steps(behavior_name)
# Para todos los agentes con un Terminal Step:
for agent_id_terminated in terminal_steps:
# Se crea la última experiencia.
last_experience = Experience(
obs=dict_last_obs_from_agent[agent_id_terminated].copy(),
reward=terminal_steps[agent_id_terminated].reward,
done=not terminal_steps[agent_id_terminated].interrupted,
action=dict_last_action_from_agent[agent_id_terminated].copy(),
next_obs=terminal_steps[agent_id_terminated].obs[0],
)
# Se limpia la última observación y la última acción. (La trayectoria termino).
dict_last_obs_from_agent.pop(agent_id_terminated)
dict_last_action_from_agent.pop(agent_id_terminated)
# Se reporta la recompensa acumulada.
cumulative_reward = (
dict_cumulative_reward_from_agent.pop(agent_id_terminated)
+ terminal_steps[agent_id_terminated].reward
)
cumulative_rewards.append(cumulative_reward)
# Se añade la Trayectoria y la última experiencia al buffer.
buffer.extend(dict_trajectories_from_agent.pop(agent_id_terminated))
buffer.append(last_experience)
# Para todos los agentes con Decision Step:
for agent_id_decisions in decision_steps:
# Si el agente no tiene una trayectoria, se crea una vacia.
if agent_id_decisions not in dict_trajectories_from_agent:
dict_trajectories_from_agent[agent_id_decisions] = []
dict_cumulative_reward_from_agent[agent_id_decisions] = 0
# Si el agente solicita una decisión con la última observación
if agent_id_decisions in dict_last_obs_from_agent:
# Se crea una experiencia de la última observación y el Decision Step.
exp = Experience(
obs=dict_last_obs_from_agent[agent_id_decisions].copy(),
reward=decision_steps[agent_id_decisions].reward,
done=False,
action=dict_last_action_from_agent[agent_id_decisions].copy(),
next_obs=decision_steps[agent_id_decisions].obs[0],
)
# Se actualiza la trayectoria del agente y su recompensa acumulada.
dict_trajectories_from_agent[agent_id_decisions].append(exp)
dict_cumulative_reward_from_agent[agent_id_decisions] += (
decision_steps[agent_id_decisions].reward
)
# Guarda la observación como la nueva última observación.
dict_last_obs_from_agent[agent_id_decisions] = (
decision_steps[agent_id_decisions].obs[0]
)
# Se genera una acción para cada agente que solicit una decisión.
# Se calculan los valores para cada acción dada la observación.
actions_values = (
q_net(torch.from_numpy(decision_steps.obs[0])).detach().numpy()
)
# Se añade algo de ruido epsilon a los valores.
actions_values += epsilon * (
np.random.randn(actions_values.shape[0], actions_values.shape[1])
).astype(np.float32)
# Selecciona la mejor acción usando argmax.
actions = np.argmax(actions_values, axis=1)
actions.resize((len(decision_steps), 1))
# Guarda la acción, se pondrá en una trayectoría después.
for agent_index, agent_id in enumerate(decision_steps.agent_id):
dict_last_action_from_agent[agent_id] = actions[agent_index]
# Se establecen las acciones del environment.
# Los Unity Environments esperan instancias del tipo ActionTuple.
action_tuple = ActionTuple()
action_tuple.add_discrete(actions)
env.set_actions(behavior_name, action_tuple)
# Se realiza un paso en la simulación.
env.step()
return buffer, np.mean(cumulative_rewards)
@staticmethod
def update_q_net(
q_net: VisualQNetwork,
optimizer: torch.optim,
buffer: Buffer,
action_size: int
):
"""
Realiza una actualización de Q-Network utilizando el optimizador y el búfer proporcionados
"""
BATCH_SIZE = 1000
NUM_EPOCH = 3
GAMMA = 0.9
batch_size = min(len(buffer), BATCH_SIZE)
random.shuffle(buffer)
# Se separa el buffer en batches.
batches = [
buffer[batch_size * start : batch_size * (start + 1)]
for start in range(int(len(buffer) / batch_size))
]
for _ in range(NUM_EPOCH):
for batch in batches:
# Create the Tensors that will be fed in the network
obs = torch.from_numpy(np.stack([ex.obs for ex in batch]))
reward = torch.from_numpy(
np.array([ex.reward for ex in batch], dtype=np.float32).reshape(-1, 1)
)
done = torch.from_numpy(
np.array([ex.done for ex in batch], dtype=np.float32).reshape(-1, 1)
)
action = torch.from_numpy(np.stack([ex.action for ex in batch]))
next_obs = torch.from_numpy(np.stack([ex.next_obs for ex in batch]))
# Use the Bellman equation to update the Q-Network
target = (
reward
+ (1.0 - done)
* GAMMA
* torch.max(q_net(next_obs).detach(), dim=1, keepdim=True).values
)
mask = torch.zeros((len(batch), action_size))
mask.scatter_(1, action, 1)
prediction = torch.sum(qnet(obs) * mask, dim=1, keepdim=True)
criterion = torch.nn.MSELoss()
loss = criterion(prediction, target)
# Perform the backpropagation
optimizer.zero_grad()
loss.backward()
optimizer.step()6. Entrenamiento.
- Código en Python del entrenamiento.
# Cierra un env si no ha sido cerrado antes.
try:
env.close()
except:
pass
env = UnityEnvironment(file_name = None, base_port=5004)
print("Environment created")
# Se crea una Q-Network.
qnet_input = (84,84,4)
qnet_output = 3
qnet_encoding_size = 126
qnet = VisualQNetwork(qnet_input, qnet_encoding_size, qnet_output)
# Se crea un bufer para las experiencias.
experiences: Buffer = []
# Se define el optimizador (Adam).
optim = torch.optim.Adam(qnet.parameters(), lr= 0.001)
# Contenedor para las recompensas acumuladas.
cumulative_rewards: List[float] = []
# Se definen el número de steps a realizar (70).
NUM_TRAINING_STEPS = 50
# Se define el número de experiencias a recolectar en cada step de entrenamiento.
NUM_NEW_EXP = 1000
# Se define el tamaño máximo del buffer.
BUFFER_SIZE = 10000
for n in range(NUM_TRAINING_STEPS):
new_exp,_ = Trainer.generate_trajectories(env, qnet, NUM_NEW_EXP, epsilon = 0.1)
random.shuffle(experiences)
if len(experiences) > BUFFER_SIZE:
experiences = experiences[:BUFFER_SIZE]
experiences.extend(new_exp)
Trainer.update_q_net(qnet, optim, experiences, 3)
_, rewards = Trainer.generate_trajectories(env, qnet, 100, epsilon = 0)
cumulative_rewards.append(rewards)
print("Training step", n+1, "\treward", rewards)
env.close()
print("Closed environment.")
# Muestra el gráfico de entrenamiento.
plt.plot(range(NUM_TRAINING_STEPS), cumulative_rewards)Muestra del agente en su etapa de entrenamiento, logrando su primer giro, y entrenado con una GPU:
Para mayor información, consultar documentación en el siguiente PDF:
Jupyter notebook del proyecto:
Deep-Q Learning with Unity and Pytorch.
Referencias.
· Technologies, U. (2020, 5 junio). Unity — Manual: Unity User Manual (2019.3). Unity 2019 -Documentation. https://docs.unity3d.com/2019.3/Documentation/Manual/index.html
· U. (2020). Unity-Technologies/ml-agents. GitHub MLAgents Toolkit. https://github.com/Unity-Technologies/ml-agents
· U. (2020). Unity-Technologies/ml-agents. GitHub MLAgents Python API. https://github.com/Unity-Technologies/ml-agents/blob/main/docs/Python-APIDocumentation.md
· Juliani, A., Berges, V., Teng, E., Cohen, A., Harper, J., Elion, C., Goy, C., Gao, Y., Henry, H., Mattar, M., Lange, D.(2020). Unity: A General Platform for Intelligent Agents. arXiv preprint arXiv:1809.02627. https://github.com/Unity-Technologies/ml-agents.
· Kenney • Racing Kit. (2010). Kenney Racing Kit. https://www.kenney.nl/assets/racing-kit
· Reinforcement Learning Series Intro — Syllabus Overview. (2018). Deeplizard. https://deeplizard.com/learn/video/nyjbcRQ-uQ8
· Sutton, R. S., & Barto, A. G. (2018). Reinforcement Learning, Second Edition: An Introduction (2nd ed.) [Libro electrónico]. Bradford Book. http://incompleteideas.net/book/RLbook2020.pdf
· GitHub del Proyecto: https://github.com/jabc300/Deep-Q-Learning-with-Unity-and-Pytorch
