Removed Null move possibility in the middle of a game

Author: Tesla2000

.gitignore

@@ -1,3 +1,5 @@
 /sandbox.py
 /data/
 /models/
+/evaluation_data/
+/training_data/

Config.py

@@ -7,22 +7,22 @@
 
 class _ConfigPaths:
     root = Path(__file__).parent
-    training_data_path = root / 'training_data'
+    training_data_path = root / "training_data"
     training_data_path.mkdir(exist_ok=True)
-    evaluation_data_path = root / 'evaluation_data'
+    evaluation_data_path = root / "evaluation_data"
     evaluation_data_path.mkdir(exist_ok=True)
-    model_path = root / 'models'
+    model_path = root / "models"
     model_path.mkdir(exist_ok=True)
 
 
 class _ConfigAgent:
     # hidden_sizes = (256, 128, 64, 32)
     hidden_sizes = (256,)
     # hidden_sizes = tuple()
-    c = .1
-    learning_rate = 1e-3
-    debug = False
-    pretrain = True
+    c = 0.1
+    learning_rate = 1e-4
+    debug = True
+    pretrain = False
 
 
 class Config(_ConfigPaths, _ConfigAgent):
@@ -35,10 +35,11 @@ class Config(_ConfigPaths, _ConfigAgent):
     n_simulations = 100
     n_games = None
     n_players = 2
-    n_actions = 46
+    n_actions = 45
+    eval_rate = 0.2
 
 
 if Config.debug:
-    random.seed(42)
-    np.random.seed(42)
-    torch.random.manual_seed(42)
+    random.seed(69)
+    np.random.seed(69)
+    torch.random.manual_seed(69)

agent/Agent.py

@@ -15,7 +15,7 @@ def __init__(
         self,
         n_players: int,
         hidden_sizes: tuple = Config.hidden_sizes,
-        n_moves: int = 46,
+        n_moves: int = Config.n_actions,
     ):
         super().__init__()
         self.tanh = nn.Tanh()

agent/RLDataset.py

@@ -12,5 +12,8 @@ def __len__(self):
         return len(self.examples)
 
     def __getitem__(self, index) -> tuple[np.array, ...]:
-        return np.array(self.examples[index][0]), np.array(self.examples[index][1]), np.array(
-            [self.examples[index][2] * 2 - 1])
+        return (
+            np.array(self.examples[index][0]),
+            np.array(self.examples[index][1]),
+            np.array([self.examples[index][2] * 2 - 1]),
+        )

agent/pretrain.py

@@ -0,0 +1,40 @@
+import operator
+from collections import deque
+from functools import reduce
+from itertools import islice
+
+import torch
+
+from Config import Config
+from agent.Agent import Agent
+from agent.train_agent import train_agent
+
+
+def pretrain(agents: deque[Agent]):
+    for agent, checkpoint_index in zip(
+        islice(reversed(agents), 1, None),
+        sorted(
+            (int(path.name.split(".")[0]) for path in Config.model_path.iterdir()),
+            reverse=True,
+        ),
+    ):
+        agent.load_state_dict(
+            torch.load(Config.model_path.joinpath(f"{checkpoint_index}.pth"))
+        )
+    newest = Config.model_path.joinpath(
+        f"{max((*tuple(int(path.name.split('.')[0]) for path in Config.model_path.iterdir()), 0))}.pth"
+    )
+    if newest.exists():
+        agents[-1].load_state_dict(torch.load(newest))
+    training_buffer = reduce(
+        operator.add,
+        (
+            deque(eval(path.read_text()))
+            for path in sorted(
+                Config.training_data_path.iterdir(), key=lambda path: int(path.name)
+            )
+        ),
+        deque(maxlen=Config.training_buffer_len),
+    )
+    train_agent(agents[-1], training_buffer)
+    return training_buffer

agent/save.py

@@ -0,0 +1,9 @@
+from Config import Config
+
+
+def save_temp_buffer(buffer, train: bool):
+    path = Config.training_data_path if train else Config.evaluation_data_path
+    index = max((*tuple(int(path.name) for path in path.iterdir()), -1)) + 1
+    path.joinpath(str(index)).write_text(
+        str(list((list(sample[0]), list(sample[1]), sample[2]) for sample in buffer))
+    )

agent/search.py

@@ -40,4 +40,4 @@ def search(
         N[state][action] + 1
     )
     N[state][action] += 1
-    return -v
+    return -v

agent/self_play.py

@@ -11,34 +11,50 @@
 from src.Game import Game
 
 
-def self_play(agents: deque[Agent]) -> tuple[list[tuple[np.array, np.array, int]], list[Agent]]:
+def self_play(
+    agents: deque[Agent],
+) -> tuple[list[tuple[np.array, np.array, int]], list[Agent]]:
     states = []
     winners = []
     initial_state = Game(n_players=Config.n_players)
     for agent in agents:
         agent.eval()
-    for agents_in_order in islice(windowed(cycle(agents), Config.n_players), Config.n_players):
+    for agents_in_order in islice(
+        windowed(cycle(agents), Config.n_players), Config.n_players
+    ):
         game = initial_state.copy()
-        id_to_agent = dict((player.id, agent) for agent, player in zip(agents_in_order, game.players))
+        id_to_agent = dict(
+            (player.id, agent) for agent, player in zip(agents_in_order, game.players)
+        )
         results, winner = _perform_game(game, [], id_to_agent)
         states += results
         winners.append(winner)
     return states, winners
 
 
-def _perform_game(game: Game, states: list, id_to_agent: dict[int, Agent]) -> tuple[
-    list[tuple[np.array, np.array, int]], Agent]:
-    for _ in tqdm(count()):
+def _perform_game(
+    game: Game, states: list, id_to_agent: dict[int, Agent]
+) -> tuple[list[tuple[np.array, np.array, int]], Agent]:
+    for turn in tqdm(count()):
         agent = id_to_agent[game.current_player.id]
         pi, action = policy(game, agent, Config.c, Config.n_simulations)
         action_index = game.all_moves.index(action)
         onehot_encoded_action = np.zeros(Config.n_actions)
         onehot_encoded_action[action_index] = 1
-        states.append((game, onehot_encoded_action, 0))
+        states.append((game, action, 0))
         game = game.perform(action)
         if game.is_terminal():
             result = game.get_results()
             return (
                 list(
-                    (state[0].get_state(), state[1], int(result[state[0].current_player.id] == 1)) for state in states),
-                id_to_agent[next(player.id for player in game.players if result[player.id])])
+                    (
+                        state[0].get_state(),
+                        (onehot_encoded_action := np.zeros(Config.n_actions), onehot_encoded_action.__setitem__(game.all_moves.index(state[1]), 1))[0],
+                        int(result[state[0].current_player.id] == 1),
+                    )
+                    for state in states
+                ),
+                id_to_agent[
+                    next(player.id for player in game.players if result[player.id])
+                ],
+            )

agent/train_agent.py

@@ -18,7 +18,11 @@ def train_agent(agent: Agent, train_data: deque[tuple[tuple, np.array, int]]):
     loader = DataLoader(dataset, batch_size=Config.train_batch_size)
     for batch in loader:
         state, policy, win_probability = batch
-        state, policy, win_probability = state.float(), policy.float(), win_probability.float()
+        state, policy, win_probability = (
+            state.float(),
+            policy.float(),
+            win_probability.float(),
+        )
         optimizer.zero_grad()
         output_policy, output_v = agent(state)
         bce = mse(output_v, win_probability)

main.py

@@ -1,49 +1,77 @@
+import random
 import re
 from collections import deque
 from copy import deepcopy
-from itertools import count, islice
-from pathlib import Path
+from itertools import count
 
 import torch
 
 from Config import Config
 from agent.Agent import Agent
+from agent.pretrain import pretrain
+from agent.save import save_temp_buffer
 from agent.self_play import self_play
 from agent.train_agent import train_agent
 
 
 def main():
     training_buffer = deque(maxlen=Config.training_buffer_len)
-    agents = deque((Agent(Config.n_players) for _ in range(Config.n_players)), maxlen=Config.n_players)
+    agents = deque(
+        (Agent(Config.n_players) for _ in range(Config.n_players)),
+        maxlen=Config.n_players,
+    )
     if Config.pretrain:
-        for agent, checkpoint_index in zip(islice(reversed(agents), 1, None), sorted((int(path.name.split('.')[0]) for path in Config.model_path.iterdir()), reverse=True)):
-            agent.load_state_dict(torch.load(Config.model_path.joinpath(f'{checkpoint_index}.pth')))
-        newest = Config.model_path.joinpath(
-            f"{max((*tuple(int(path.name.split('.')[0]) for path in Config.model_path.iterdir()), 0))}.pth")
-        if newest.exists():
-            agents[-1].load_state_dict(torch.load(newest))
-        training_buffer += list(map(eval, map(Path.read_text, sorted(Config.training_data_path.iterdir(), key=lambda path: int(path.name), reverse=True)[:Config.training_buffer_len])))
-        train_agent(agents[-1], training_buffer)
+        pretrain(agents)
     scores = deque(maxlen=Config.max_results_held)
-    for _ in (count() if Config.n_games is None else range(Config.n_games)):
+    for _ in count() if Config.n_games is None else range(Config.n_games):
         buffer, winners = self_play(agents)
-        start_index = max((*tuple(int(path.name) for path in Config.training_data_path.iterdir()), -1)) + 1
-        for start_index, sample in enumerate(buffer, start_index + 1):
-            Config.training_data_path.joinpath(str(start_index)).write_text(str((list(sample[0]), list(sample[1]), sample[2])))
+        to_train = random.random() > Config.eval_rate
+        save_temp_buffer(buffer, to_train)
         for winner in winners:
             scores.append(agents[-1] is winner)
-        if (len(scores) < Config.min_games_to_replace_agents and sum(scores) >= Config.minimal_relative_agent_improvement * Config.min_games_to_replace_agents / len(agents)) or (len(scores) >= Config.min_games_to_replace_agents and sum(scores) >= Config.minimal_relative_agent_improvement * len(scores) / len(agents)):
-            torch.save(agents[-1].state_dict(), Config.model_path.joinpath(str(max(map(int, (*re.findall(r'\d+', ''.join(map(str, Config.model_path.iterdir()))), -1))) + 1) + ".pth"))
+        if (
+            len(scores) < Config.min_games_to_replace_agents
+            and sum(scores)
+            > Config.minimal_relative_agent_improvement
+            * Config.min_games_to_replace_agents
+            / len(agents)
+        ) or (
+            len(scores) > Config.min_games_to_replace_agents
+            and sum(scores)
+            >= Config.minimal_relative_agent_improvement * len(scores) / len(agents)
+        ):
+            torch.save(
+                agents[-1].state_dict(),
+                Config.model_path.joinpath(
+                    str(
+                        max(
+                            map(
+                                int,
+                                (
+                                    *re.findall(
+                                        r"\d+",
+                                        "".join(map(str, Config.model_path.iterdir())),
+                                    ),
+                                    -1,
+                                ),
+                            )
+                        )
+                        + 1
+                    )
+                    + ".pth"
+                ),
+            )
             agents.append(Agent(Config.n_players))
-            agents[-1].load_state_dict(deepcopy(agents[-1].state_dict()))
+            agents[-1].load_state_dict(deepcopy(agents[-2].state_dict()))
             agents[-1].training = True
             scores = deque(maxlen=Config.max_results_held)
         elif len(scores) >= Config.min_games_to_replace_agents:
-            print(f'{len(scores)} {sum(scores) / len(scores):.2f}')
+            print(f"{len(scores)} {sum(scores) / len(scores):.2f}")
         else:
-            print(f'{len(scores)} {sum(scores)}/{len(scores)}')
-        training_buffer += buffer
-        train_agent(agents[-1], training_buffer)
+            print(f"{len(scores)} {sum(scores)}/{len(scores)}")
+        if to_train:
+            training_buffer += buffer
+            train_agent(agents[-1], training_buffer)
 
 
 if __name__ == "__main__":