Recientemente intenté implementar un agente de ajedrez basado en DQN.

Ahora, cualquiera que sepa cómo funcionan DQNS y Chess te diría que es una idea tonta.

Y ... lo fue, pero como principiante lo disfruté, sin embargo. En este artículo compartiré las ideas que aprendí mientras trabajaba en esto.

Comprender el medio ambiente.

Antes de comenzar a implementar el agente en sí, tuve que familiarizarme con el entorno que usaré y hacer un envoltorio personalizado encima para que pueda interactuar con el agente durante la capacitación.

• Usé el entorno de ajedrez de la biblioteca kaggle_environments.
  
  

   from kaggle_environments import make
   env = make("chess", debug=True)
  

• también usé Chessnut, que es una biblioteca de Python liviana que ayuda a analizar y validar los juegos de ajedrez.
  
  

   from Chessnut import Game
   initial_fen = env.state[0]['observation']['board']
   game=Game(env.state[0]['observation']['board'])
  

En este entorno, el estado de la Junta se almacena en el formato FEN.

proporciona una forma compacta de representar todas las piezas en el tablero y el reproductor actualmente activo. Sin embargo, dado que planeé alimentar la entrada a una red neuronal, tuve que modificar la representación del estado.

Convertir fen en formato de matriz

Dado que hay 12 tipos diferentes de piezas en un tablero, creé 12 canales de 8x8 cuadrículas para representar el estado de cada uno de esos tipos en el tablero.

Creación de un contenedor para el entorno

class EnvCust:
    def __init__(self):
        self.env = make("chess", debug=True)
        self.game=Game(env.state[0]['observation']['board'])
        print(self.env.state[0]['observation']['board'])
        self.action_space=game.get_moves();
        self.obs_space=(self.env.state[0]['observation']['board'])

    def get_action(self):
        return Game(self.env.state[0]['observation']['board']).get_moves();


    def get_obs_space(self):
        return fen_to_board(self.env.state[0]['observation']['board'])

    def step(self,action):
        reward=0
        g=Game(self.env.state[0]['observation']['board']);
        if(g.board.get_piece(Game.xy2i(action[2:4]))=='q'):
            reward=7
        elif g.board.get_piece(Game.xy2i(action[2:4]))=='n' or g.board.get_piece(Game.xy2i(action[2:4]))=='b' or g.board.get_piece(Game.xy2i(action[2:4]))=='r':
            reward=4
        elif g.board.get_piece(Game.xy2i(action[2:4]))=='P':
            reward=2
        g=Game(self.env.state[0]['observation']['board']);
        g.apply_move(action)
        done=False
        if(g.status==2):
            done=True
            reward=10
        elif g.status == 1:  
            done = True
            reward = -5 
        self.env.step([action,'None'])
        self.action_space=list(self.get_action())
        if(self.action_space==[]):
            done=True
        else:
            self.env.step(['None',random.choice(self.action_space)])
            g=Game(self.env.state[0]['observation']['board']);
            if g.status==2:
                reward=-10
                done=True

        self.action_space=list(self.get_action())
        return self.env.state[0]['observation']['board'],reward,done

El punto de este envoltorio era proporcionar una política de recompensa para el agente y una función de paso que se usa para interactuar con el entorno durante la capacitación.

Chessnut fue útil para obtener información como los movimientos legales posibles en el estado actual del tablero y también para reconocer a los compañeros de verificación durante el juego.

Traté de crear una política de recompensa para dar puntos positivos a los compañeros de verificación y eliminar las piezas enemigas mientras los puntos negativos por perder el juego.

Creando un búfer de repetición

Buffer de reproducción se usa durante el período de entrenamiento para guardar la salida (estado, acción, recompensa, siguiente estado) mediante la red Q y luego se usa al azar para la backpropagation de la red de destino

Funciones auxiliares

Chessnut devuelve acciones legales en formato UCI que se parece a 'A2A3', sin embargo, para interactuar con la red neuronal, convertí cada acción en un índice distinto usando un patrón básico. Hay un total de 64 cuadrados, por lo que decidí tener 64*64 índices únicos para cada movimiento.
Sé que no todos los 64*64 movimientos serían legales, pero podría manejar la legalidad usando Chessnut y el patrón era bastante simple.

Estructura de red neuronal

import torch
import torch.nn as nn
import torch.optim as optim

class DQN(nn.Module):
    def __init__(self):  
        super(DQN, self).__init__()
        self.conv_layers = nn.Sequential(
            nn.Conv2d(12, 32, kernel_size=3, stride=1, padding=
            nn.ReLU(),  
            nn.Conv2d(32, 64, kernel_size=3, stride=1, padding=1),  
            nn.ReLU()
        )
        self.fc_layers = nn.Sequential(
            nn.Flatten(),
            nn.Linear(64 * 8 * 8, 256),
            nn.ReLU(), 
            nn.Linear(256, 128),
            nn.ReLU(), 
            nn.Linear(128, 4096)
        )

    def forward(self, x):
        x = x.unsqueeze(0)
        x = self.conv_layers(x)  
        x = self.fc_layers(x)  
        return x

    def predict(self, state, valid_action_indices):
        with torch.no_grad(): 
            q_values = self.forward(state) 
            q_values = q_values.squeeze(0)  


            valid_q_values = q_values[valid_action_indices]


            best_action_relative_index = valid_q_values.argmax().item()  
            max_q_value=valid_q_values.argmax()
            best_action_index = valid_action_indices[best_action_relative_index] 

            return max_q_value, best_action_index

Esta red neuronal usa las capas convolucionales para tomar la entrada de 12 canales y también usa los índices de acción válidos para filtrar la predicción de salida de recompensa.

Implementación del agente

model = DQN().to(device)  # The current Q-network
target_network = DQN().to(device)  # The target Q-network
optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
replay_buffer = ReplayBuffer(buffer_size=10000)
epsilon = 0.5  
gamma = 0.99 
batch_size=15
def train(episodes):
    for ep in range(1,episodes 1):
        print('Episode Number:',ep)
        myenv=EnvCust()
        done=False
        state=myenv.obs_space
        i=0
        while not done and i batch_size:
                mini_batch = replay_buffer.sample(batch_size)
                for e in mini_batch:
                    state, action, reward, next_state, done = e
                    g=Game(next_state)
                    act=g.get_moves();
                    ind_a=action_index(act)
                    input_state=torch.tensor(fen_to_board(next_state), dtype=torch.float32, requires_grad=True).to(device)
                    tpred,_=target_network.predict(input_state,ind_a)
                    target = reward   gamma * tpred * (1 - done)

                    act_ind=uci_to_action_index(action)
                    input_state2=torch.tensor(fen_to_board(state), dtype=torch.float32, requires_grad=True).to(device)
                    current_q_value =model(input_state2)[0,act_ind]

                    loss = (current_q_value - target) ** 2
                    optimizer.zero_grad()
                    loss.backward()
                    optimizer.step()
            if ep % 5 == 0:
                target_network.load_state_dict(model.state_dict())

Este era obviamente un modelo muy básico que no tenía posibilidades de funcionar bien (y no lo hizo), pero me ayudó a entender cómo funcionan un poco mejor.