Working evaluation for single player mode

Author: klieret

codeclash/agents/abstract.py

@@ -7,7 +7,8 @@
 
 from codeclash.agents.utils import GameContext
 from codeclash.constants import GH_ORG
-from codeclash.utils.environment import assert_zero_exit_code
+from codeclash.tournaments.utils.git_utils import filter_git_diff
+from codeclash.utils.environment import assert_zero_exit_code, create_file_on_container
 from codeclash.utils.log import get_logger
 
 load_dotenv()
@@ -81,6 +82,43 @@ def push(self) -> None:
             f"Pushed {self.name} commit history to remote repository (branch {self._branch_name})"
         )
 
+    def reset_and_apply_patch(
+        self, patch: str, *, base_commit: str = "", filter_patch: bool = True
+    ) -> None:
+        """Clean all uncommited changes. If base_commit is provided, reset to that commit.
+        Then apply the patch to the codebase.
+        """
+        # Need to clean before we copy over the patch (else it's gonna be removed by git clean)
+        self.logger.debug(
+            assert_zero_exit_code(
+                self.environment.execute(
+                    f"git reset --hard {base_commit} && git clean -fd"
+                )
+            )
+        )
+
+        patch = filter_git_diff(patch) if filter_patch else patch
+
+        if not patch.strip():
+            self.logger.debug("No patch to apply, skipping")
+            return
+
+        create_file_on_container(
+            container=self.environment,  # type: ignore
+            content=patch,
+            dest_path="tmp_patch.txt",
+        )
+
+        self.logger.debug(f"Applying patch to agent's codebase: {patch}")
+
+        commands = ["git status", "git apply tmp_patch.txt", "rm -f tmp_patch.txt"]
+        for cmd in commands:
+            self.logger.debug(f"Executing command: {cmd}")
+            out = assert_zero_exit_code(
+                self.environment.execute(cmd), logger=self.logger
+            )
+            self.logger.debug(out)
+
     # --- Helper methods ---
 
     def _tag_round(self, round: int) -> None:

codeclash/tournaments/abstract.py

@@ -1,8 +1,10 @@
 import getpass
 import time
+import traceback
 from pathlib import Path
 
 from codeclash.constants import DIR_LOGS
+from codeclash.utils.environment import create_file_on_container
 from codeclash.utils.log import get_logger
 
 
@@ -24,3 +26,18 @@ def __init__(self, config: dict, *, name: str, **kwargs):
 
     def get_metadata(self) -> dict:
         return self._metadata
+
+    def _copy_game_log_to_agent(self, agent, round_num: int, log_output: str) -> None:
+        """Copy round log to agent environment."""
+        try:
+            create_file_on_container(
+                container=agent.environment,
+                content=log_output,
+                dest_path=f"logs/round_{round_num}.log",
+            )
+        except Exception:
+            self.logger.error(
+                f"Error creating round log in {agent.name}'s container: {traceback.format_exc()}"
+            )
+        else:
+            self.logger.info(f"Created round log in {agent.name}'s container.")

codeclash/tournaments/pvp_training.py

@@ -2,14 +2,11 @@
 PvP training mode where multiple agents compete against each other.
 """
 
-import traceback
-
 from codeclash.agents import get_agent
 from codeclash.agents.abstract import Player
 from codeclash.games import get_game
 from codeclash.games.abstract import CodeGame
 from codeclash.tournaments.abstract import AbstractTournament
-from codeclash.utils.environment import create_file_on_container
 from codeclash.utils.log import get_logger
 
 
@@ -55,35 +52,20 @@ def run_training_round(self, round_num: int) -> None:
         round_log_path.write_text(log_output)
 
         # Copy log to agent environments
-        self._post_round_setup(self.agents, round_num, log_output)
+        for agent in self.agents:
+            self._copy_game_log_to_agent(agent, round_num, log_output)
 
         for agent in self.agents:
             self.run_agent(agent, round_num)
 
+        self.logger.info("Round completed.")
+
     def run_agent(self, agent: Player, round_num: int) -> None:
         """Run a single agent for the current round."""
         agent.pre_run_hook(new_round=round_num)
         agent.run()
         agent.post_run_hook(round=round_num)
 
-    def _post_round_setup(self, agents: list, round_num: int, log_output: str) -> None:
-        """Copy round logs to agent environments and local directory."""
-        for agent in agents:
-            try:
-                create_file_on_container(
-                    container=agent.environment,
-                    content=log_output,
-                    dest_path=f"logs/round_{round_num}.log",
-                )
-            except Exception:
-                self.logger.error(
-                    f"Error creating round log in {agent.name}'s container: {traceback.format_exc()}"
-                )
-            else:
-                self.logger.info(f"Created round log in {agent.name}'s container.")
-
-        self.logger.info("Round completed.")
-
     def cleanup(self) -> None:
         """Clean up game resources and push agents if requested."""
         self.game.end(self.cleanup_on_end)

codeclash/tournaments/single_player_training.py

@@ -3,15 +3,13 @@
 """
 
 import copy
-import traceback
 
 from codeclash.agents import get_agent
 from codeclash.agents.abstract import Player
 from codeclash.games import get_game
 from codeclash.games.abstract import CodeGame
 from codeclash.tournaments.abstract import AbstractTournament
 from codeclash.tournaments.utils.git_utils import filter_git_diff
-from codeclash.utils.environment import assert_zero_exit_code, create_file_on_container
 from codeclash.utils.log import get_logger
 
 
@@ -45,6 +43,7 @@ def run(self):
         try:
             for round_num in range(1, self.rounds + 1):
                 self.run_training_round(round_num)
+            self.evaluate()
         finally:
             self.cleanup()
 
@@ -65,76 +64,65 @@ def run_training_round(self, round_num: int) -> None:
         round_log_path.write_text(log_output)
 
         # Copy log to main agent environment only
-        self._copy_game_log_to_agent([self.agent], round_num, log_output)
+        self._copy_game_log_to_agent(self.agent, round_num, log_output)
 
         self.run_main_agent(round_num)
-        self.run_mirror_agent(round_num)
+        mirror_agent_state = round_num - 1 if round_num > 1 else 0
+        self.set_mirror_state_to_round(mirror_agent_state)
+
+        self.logger.info("Round completed.")
 
     def run_main_agent(self, round_num: int):
         """Run the main agent for the current round."""
         self.agent.pre_run_hook(new_round=round_num)
         self.agent.run()
         self.agent.post_run_hook(round=round_num)
 
-    def run_mirror_agent(self, round_num: int):
+    def set_mirror_state_to_round(self, round_num: int):
         """Update mirror agent's codebase with the main agent's changes."""
-        if round_num == 1:
-            self.logger.info("Skipping updating mirror agent for round 1")
-            return
-
-        # Set mirror agent's codebase to the main agent's codebase of the previous round
-        full_diff = self.agent.get_metadata()["diff"][round_num - 1]
-
-        full_diff = filter_git_diff(full_diff)
-
-        if full_diff.strip():
-            self.logger.debug(
-                assert_zero_exit_code(
-                    self.mirror_agent.environment.execute(
-                        "git reset --hard && git clean -fd"
-                    )
-                )
-            )
-
-            create_file_on_container(
-                container=self.mirror_agent.environment,  # type: ignore
-                content=full_diff,
-                dest_path="tmp_patch.txt",
-            )
-
-            self.logger.info("Applying patch to mirror agent's codebase")
-            self.logger.debug(f"Full diff: {full_diff}")
-
-            commands = ["git status", "git apply tmp_patch.txt", "rm -f tmp_patch.txt"]
-            for cmd in commands:
-                self.logger.debug(f"Executing command: {cmd}")
-                out = assert_zero_exit_code(
-                    self.mirror_agent.environment.execute(cmd), logger=self.logger
-                )
-                self.logger.debug(out)
+        if round_num == 0:
+            full_diff = ""
         else:
-            self.logger.info("No diff found for mirror agent, skipping update")
-
-    def _copy_game_log_to_agent(
-        self, agents: list, round_num: int, log_output: str
-    ) -> None:
-        """Copy round logs to agent environments and local directory."""
-        for agent in agents:
-            try:
-                create_file_on_container(
-                    container=agent.environment,
-                    content=log_output,
-                    dest_path=f"logs/round_{round_num}.log",
-                )
-            except Exception:
-                self.logger.error(
-                    f"Error creating round log in {agent.name}'s container: {traceback.format_exc()}"
-                )
-            else:
-                self.logger.info(f"Created round log in {agent.name}'s container.")
+            full_diff = self.agent.get_metadata()["diff"][round_num]
+            full_diff = filter_git_diff(full_diff)
 
-        self.logger.info("Round completed.")
+        self.mirror_agent.reset_and_apply_patch(full_diff)
 
     def cleanup(self):
         """Clean up game resources."""
         self.game.end(self.cleanup_on_end)
+
+    def evaluate(self, n_repetitions: int = 3):
+        """Evaluate the agent's performance by
+        calculating the matrix of every round against each other.
+        """
+        p1 = get_agent(self.config["player"], self.config["prompts"], self.game)
+        p1.name = "p1"
+        p2 = get_agent(self.config["player"], self.config["prompts"], self.game)
+        p2.name = "p2"
+        matrix = {
+            p1_round: {p2_round: [] for p2_round in range(0, self.rounds + 1)}
+            for p1_round in range(0, self.rounds + 1)
+        }
+        for p1_round in range(0, self.rounds + 1):
+            for p2_round in range(0, self.rounds + 1):
+                self.logger.info(
+                    f"Evaluating agent at round {p1_round} against agent at round {p2_round}"
+                )
+                p1_patch = (
+                    self.agent.get_metadata()["diff"][p1_round] if p1_round > 0 else ""
+                )
+                p2_patch = (
+                    self.agent.get_metadata()["diff"][p2_round] if p2_round > 0 else ""
+                )
+                p1.reset_and_apply_patch(p1_patch)
+                p2.reset_and_apply_patch(p2_patch)
+                for i_repetition in range(n_repetitions):
+                    result = self.game.run_round([p1, p2])
+                    winner = result["winner"]
+                    self.logger.info(
+                        f"Round {p1_round} vs {p2_round} repetition {i_repetition} winner: {winner}"
+                    )
+                    matrix[p1_round][p2_round].append(winner)
+        self.logger.info(f"Evaluation matrix: {matrix}")
+        return matrix

configs/battlesnake_single_player.yaml

@@ -1,6 +1,6 @@
 game:
   name: BattleSnake
-  rounds: 5
+  rounds: 4
   args:
     width: 11
     height: 11