Added evaluation

Author: Tesla2000

Config.py

@@ -16,20 +16,26 @@ class _ConfigPaths:
 
 
 class _ConfigAgent:
-    # hidden_sizes = (256, 128, 64, 32)
-    hidden_sizes = (256,)
-    # hidden_sizes = tuple()
-    c = 0.1
-    learning_rate = 1e-4
+    # hidden_sizes = (
+    #     256,
+    #     128,
+    #     64,
+    #     32,
+    # )
+    # hidden_sizes = (256,)
+    hidden_sizes = tuple()
+    c = 0.2
+    learning_rate = 1e-5
     debug = False
     pretrain = True
 
 
 class Config(_ConfigPaths, _ConfigAgent):
+    train = False
     max_results_held = 100
     minimal_relative_agent_improvement = 1.1
     min_games_to_replace_agents = 40
-    train_batch_size = 64
+    train_batch_size = 128
     training_buffer_len = 100_000
     min_n_points_to_finish = 15
     n_simulations = 100

agent/RLDataset.py

@@ -1,11 +1,11 @@
-from collections import deque
+from typing import Sequence
 
 import numpy as np
 from torch.utils.data import Dataset
 
 
 class RLDataset(Dataset):
-    def __init__(self, examples: deque[tuple[tuple, np.array, int]]):
+    def __init__(self, examples: Sequence[tuple[tuple, np.array, int]]):
         self.examples = examples
 
     def __len__(self):

agent/self_play.py

@@ -20,7 +20,8 @@ def self_play(
     for agent in agents:
         agent.eval()
     id_to_agent = dict(
-        (player.id, agent) for agent, player in zip(random.sample(agents, Config.n_players), game.players)
+        (player.id, agent)
+        for agent, player in zip(random.sample(agents, Config.n_players), game.players)
     )
     results, winner = _perform_game(game, [], id_to_agent)
     states += results
@@ -48,7 +49,8 @@ def _perform_game(
                         np.eye(Config.n_actions)[game.all_moves.index(state[1])],
                         int(result[state[0].current_player.id] == 1),
                     )
-                    for state in states if state[1] != game.null_move
+                    for state in states
+                    if state[1] != game.null_move
                 ),
                 id_to_agent[
                     next(player.id for player in game.players if result[player.id])

agent/train_agent.py

@@ -1,6 +1,7 @@
-from collections import deque
+from typing import Sequence
 
 import numpy as np
+from sklearn.metrics import accuracy_score
 from torch import nn, optim
 from torch.utils.data import DataLoader
 
@@ -9,24 +10,49 @@
 from .RLDataset import RLDataset
 
 
-def train_agent(agent: Agent, train_data: deque[tuple[tuple, np.array, int]]):
+def train_agent(agent: Agent, train_data: Sequence[tuple[tuple, np.array, int]]):
     agent.train()
+    optimizer = optim.Adam(agent.parameters(), lr=Config.learning_rate)
+    _loop(agent, train_data, optimizer)
+
+
+def eval_agent(agent: Agent, eval_set: Sequence[tuple[tuple, np.array, int]]):
+    agent.eval()
+    return _loop(agent, eval_set, batch_size=len(eval_set))
+
+
+def _loop(
+    agent: Agent,
+    dataset: Sequence[tuple[tuple, np.array, int]],
+    optimizer: optim.Optimizer = None,
+    batch_size=Config.train_batch_size,
+):
+    is_optimizer = optimizer is not None
     categorical_cross_entropy = nn.CrossEntropyLoss()
     mse = nn.MSELoss()
-    optimizer = optim.Adam(agent.parameters(), lr=Config.learning_rate)
-    dataset = RLDataset(train_data)
-    loader = DataLoader(dataset, batch_size=Config.train_batch_size)
-    for batch in loader:
-        state, policy, win_probability = batch
+    dataset = RLDataset(dataset)
+    loader = DataLoader(dataset, batch_size=batch_size)
+    for index, (state, policy, win_probability) in enumerate(loader):
         state, policy, win_probability = (
             state.float(),
             policy.float(),
             win_probability.float(),
         )
-        optimizer.zero_grad()
+        if is_optimizer:
+            optimizer.zero_grad()
         output_policy, output_v = agent(state)
         bce = mse(output_v, win_probability)
         cce = categorical_cross_entropy(output_policy, policy)
-        bce.backward(retain_graph=True)
-        cce.backward()
-        optimizer.step()
+        if is_optimizer:
+            bce.backward(retain_graph=True)
+            cce.backward()
+            optimizer.step()
+        else:
+            print(
+                accuracy_score(win_probability, np.sign(output_v.detach().numpy())),
+                accuracy_score(
+                    np.argmax(policy.detach().numpy(), axis=1),
+                    np.argmax(output_policy.detach().numpy(), axis=1),
+                ),
+            )
+            return bce.item(), cce.item()

main.py

@@ -1,7 +1,9 @@
+import operator
 import random
 import re
 from collections import deque
 from copy import deepcopy
+from functools import reduce
 from itertools import count
 
 import torch
@@ -11,10 +13,10 @@
 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
+from agent.train_agent import train_agent, eval_agent
 
 
-def main():
+def train_loop():
     training_buffer = deque(maxlen=Config.training_buffer_len)
     agents = deque(
         (Agent(Config.n_players) for _ in range(Config.n_players)),
@@ -74,5 +76,32 @@ def main():
             train_agent(agents[-1], training_buffer)
 
 
+def evaluation():
+    agent = Agent(Config.n_players)
+    train_set = reduce(
+        operator.add,
+        (eval(path.read_text()) for path in Config.training_data_path.iterdir()),
+    )
+    eval_set = reduce(
+        operator.add,
+        (eval(path.read_text()) for path in Config.evaluation_data_path.iterdir()),
+    )
+    prev_bce, prev_cce = float("inf"), float("inf")
+    while True:
+        train_agent(agent, train_set)
+        bce, cce = eval_agent(agent, eval_set)
+        if bce >= prev_bce and cce >= prev_bce:
+            break
+        prev_bce = bce
+        prev_cce = cce
+
+
+def main():
+    if Config.train:
+        train_loop()
+    else:
+        evaluation()
+
+
 if __name__ == "__main__":
     main()