Feat(analysis): llm as judge

Author: klieret

codeclash/analysis/llm_as_judge/__init__.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+import argparse
+import json
+import re
+from pathlib import Path
+from typing import Literal
+
+import jinja2
+import yaml
+from minisweagent.models import GLOBAL_MODEL_STATS, get_model
+from pydantic import BaseModel
+from typing_extensions import Any
+
+from codeclash.analysis.llm_as_judge.utils import FileLock, Instance, get_instances
+from codeclash.utils.log import get_logger
+
+logger = get_logger("BigQuestionsEvaluator", emoji="🤖")
+
+config_path = Path(__file__).parent / "big_questions.yaml"
+
+
+class BigQuestionsModelResponseSchema(BaseModel):
+    are_edits_motivated_by_logs: bool
+    are_edits_motivated: bool
+    are_edits_tested_with_simulations: bool
+    are_edits_validated_with_unittests: bool
+    edit_category: Literal["tweak", "fix", "feature", "change", "none"]
+    reasoning: str
+
+
+class ModelConfig(BaseModel):
+    model_name: str
+    model_kwargs: dict[str, Any]
+
+
+class BigQuestionsConfig(BaseModel):
+    version: int
+    system_prompt: str
+    instance_prompt: str
+    model: ModelConfig
+
+
+class BigQuestionsData(BaseModel):
+    instance: Instance
+    big_questions: BigQuestionsModelResponseSchema
+    config_version: int
+
+
+def extract_triple_backticks(text: str) -> str:
+    actions = re.findall(r"```bash\s*\n(.*?)\n```", text, re.DOTALL)
+    return actions[0] if actions else ""
+
+
+def evaluate(instance: Instance, config: BigQuestionsConfig) -> None:
+    target_path = instance.trajectory_path.parent.parent.parent / "llm_as_judge.json"
+    data_id = f"big_questions_v{config.version}_{config.model.model_name}_{instance.instance_id}"
+    if target_path.exists():
+        content = target_path.read_text()
+        if content.strip():
+            data = json.loads(content)
+            if data.get("data_id") == data_id:
+                logger.info(
+                    f"Skipping instance {instance.instance_id} because it already exists in {target_path} with key {data_id}"
+                )
+                return
+    model = get_model(config.model.model_name, config={"model_kwargs": config.model.model_kwargs})
+    trajectory_messages = json.loads(instance.trajectory_path.read_text())["messages"]
+    system_message = config.system_prompt
+    trajectory_message_str = ""
+    for message in trajectory_messages:
+        content = message["content"]
+        if isinstance(message["content"], list):
+            assert len(message["content"]) == 1
+            content = message["content"][0]["text"]
+        if message["role"] == "assistant":
+            trajectory_message_str += "\n<action>\n" + extract_triple_backticks(content) + "\n</action>\n"
+        elif message["role"] == "user":
+            trajectory_message_str += content  # already enclosed in <output>
+    instance_message = jinja2.Template(config.instance_prompt).render(trajectory_message_str=trajectory_message_str)
+    # print(instance_message)
+    response = model.query(
+        messages=[
+            {"role": "system", "content": system_message},
+            {"role": "user", "content": instance_message},
+        ],
+        response_format=BigQuestionsModelResponseSchema,
+    )
+    # atomic write with file lock in case other analyses are also writing
+    response_data = BigQuestionsModelResponseSchema.model_validate_json(response["content"]).model_dump()
+    with FileLock(target_path):
+        # read again if changed in the meantime
+        data = {}
+        if target_path.exists():
+            content = target_path.read_text()
+            if content.strip():
+                data = json.loads(content)
+        data[data_id] = response_data
+        target_path.write_text(json.dumps(data))
+    logger.info(
+        f"Evaluated instance {instance.instance_id}: {response_data}. Saved to {target_path} with key {data_id}"
+    )
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("input_dir", type=Path, help="Path to the input dir")
+    args = parser.parse_args()
+
+    config = BigQuestionsConfig.model_validate(yaml.safe_load(config_path.read_text()))
+    instances = get_instances(args.input_dir)
+    for instance in instances:
+        evaluate(instance, config)
+        print(GLOBAL_MODEL_STATS.cost)

codeclash/analysis/llm_as_judge/big_questions.py

@@ -0,0 +1,120 @@
+# Incrementing this version will cause previous submissions to be re-evaluated
+version: 4
+system_prompt: |
+  You need to answer three question
+
+  ## Are edits motivated by logs?
+
+  Are the edits of the main player file of the player directly motivated by problems discovered by reading the logs?
+
+  We want to check if the edits of the main player file (`main.py`, `player.py`, `robot.js`, `warrior.red`, `robots/custom/`) are well motivated by the results of previous game logs.
+
+  Special case: If there are no edits to the main player file, answer `True`.
+
+  Reasons to answer `False` if there are edits to the main player file, but
+
+  1. Player does not look at logs.
+  2. Player reads some lines of the logs, but no clear failure mode is inferable. For example, the lines only state some game state, but it is not clear what is going wrong, for example because only the first lines of the game log are shown without showing the conclusion. Or the logs only show which player won but without much of a reason.
+  3. Player runs an analysis script, but the analysis script does not return an actionable outcome or information that allows to infer it. For example, the analyais script only reports losses, but without clear attribution about what went wrong.
+  4. A clear failure mode is uncovered in some of the logs or analyses, but the edits do not seem to be correlated to this failure mode.
+
+  Answer `True` if ALL (!) of the following is true:
+
+  1. A failure mode can be inferred with the help of reading the logs or analysis scripts evaluating the logs. Note that the failuire mode needs not be spelled out in any of the action outputs. It is enough that there is enough information to infer a failure mode based on basic reasoning.
+  2. The edit is directly related to this failure mode. It is ok if some minor parts of the edit are unrelated.
+
+  The logs can be either from a game that the player simulates it self, or from the previous round, but it must be a meaningful game log.
+
+  Here are some example of real failure modes:
+
+  - The snake that the player is controlling runs out of food (so we need to more aggressively search for food)
+  - Our bot runs against a wall (so we change that)
+  - Our race car does not move for several turns (so we fix some movement related bug)
+  - Our warrior's missiles do not hit the enemy (so we improve something about the aim)
+  - Our code times out (so we improve efficiency)
+
+  Here are some example of not real failure modes:
+
+  - Player 1 won 99% of the rounds (why?)
+  - Player 2 is better most of the time (why?)
+  - Player 1 is the last bot standing and is therefore the winner (does not explain why player 2 lost)
+
+  ## Are edits motivated
+
+  Can the goal of the specific edits on the main player file be motivated by any output of previous actions?
+  If you answered True to the previous question, answer True here as well.
+  However, you can also answer True here, if one or more of the following is true:
+
+  1. The player wrote a meaningful test that revealed a problem (or a way to improve) and then performed the corresponding edit
+  2. The player wrote a meaningful analysis script that revealed a problem (or a way to improve) and then performed the corresponding edit
+  3. The player ran some test games that revealed a problem (or a way to improve) and then performed the corresponding edit
+  4. The player made some changes, and then ran test games against the previous version and verified that the changes improved the performance, i.e., had a higher win rate.
+
+  However, if for 1. and 2. the test or analysis script gives a recommendation that's not
+  corroborated by the actual code of the analysis or test file, or by its respective output,
+  answer False. This for example applies to static messages in the analysis or test file.
+  If you do not see the output or the code of the analysis or test file, answer False as well.
+
+  Just stating a low win rate is not a sufficient motivation.
+  Remember: This is about the _specific_ edits being motivated, not just any edits.
+
+  Here problem can mean anything from a bug in the code, to something that can be improved otherwise.
+
+  You should answer False, if the edits seemed unrelated to any output of previous actions before the relevant
+  edit actions.
+
+  ## Are edits tested with simulations?
+
+  After edits to the main player file have been made, does the player simulate games by playing the game against previous versions of itself, or against example players, etc.
+  Only if we are performing a fix or an improvement that can be validated without an opponent (e.g., avoiding collisions with the wall in a car chase game), does a simulated game with only one player (the latest version) count.
+
+  In order to answer `True`, a real game has to be played. If there is an opponent, the new version has to win (or have a good win rate).
+
+  Note: If the game failed to run, or showed that the new version was clearly worse than the previous version, answer False.
+  Note: Unit tests do NOT (!) count as a simulated game.
+
+  Special case:
+
+  1. If no edits to the main player file have been made, answer `True`
+
+  ## Are edits validated with unittests?
+
+  Are the final edits covered by specific unittests that test the new or modified behavior?
+
+  Special case: If there are no significant changes that could be validated, answer True.
+
+  Running the game to get a win rate does not count as a unittest, because it does not specifically validate specific changes.
+
+  Running unittests that are unrelated to the changes does not count either.
+
+  Answer `True`, if the unittests cover (some of) the new behavior. They do not have to be painfully complete or handle every special case, but they should test the core change that has been made.
+
+  You can also count tests that only print output (but do not have assert statements) as unit tests,
+  if they essentially print the expected output of the new or modified behavior and can therefore be
+  used to validate the new or modified behavior.
+
+  Note: if the tests did not run, or showed that the new version was broken, answer False.
+
+  ## Categorize the kind of edits made
+
+  Categorize the final edits to the main file into one of the following categories.
+  Ignore comments or documentation.
+  You can only select ONE (!) category. Choose the one that describes the changes best.
+
+  1. `tweak`: Logic is left unchanged, but we do change some parameters.
+  2. `fix`: Small, targeted change with the intent to fix broken behavior.
+  3. `feature`: Significant new behavior is added, mostly extending the existing code
+  4. `change`: We significantly change the behavior by rewriting significant logic of the code
+  5. `none`: No change in behavior. Only comments, documentation, refactoring was performed.
+
+  ## Output format
+
+  Answer in the json format specified. Reasoning corresponds to your explanation for your answer.
+instance_prompt: |
+  Here is your input file:
+
+  {{ trajectory_message_str }}
+model:
+  model_name: gpt-5
+  model_kwargs:
+    reasoning_effort: high

codeclash/analysis/llm_as_judge/big_questions.yaml

@@ -0,0 +1,14 @@
+import argparse
+from pathlib import Path
+
+import yaml
+
+config_path = Path(__file__).parent / "categorize_actions.yaml"
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("input_dir", type=Path, help="Path to the input dir")
+    args = parser.parse_args()
+
+    config = yaml.safe_load(config_path.read_text())

codeclash/analysis/llm_as_judge/categorize_actions.py

@@ -0,0 +1,57 @@
+# Incrementing this version will cause previous submissions to be re-evaluated
+version: 1
+system_prompt: |
+  You are helping to analyze the actions of a LM agent.
+
+  For every action, you return a category as specified by the structured output specs.
+
+  # Categories
+
+  ## Read operations
+
+  The model reads code, documentation, logs, or anything else. This includes operations like finding files, searching through files, etc. Commands include `ls`, `find`, `grep`, `cat`, `head`, `tail`, etc. This does NOT include running more complicated analysis scripts on files. Rule of thumb: If it's a bash command, it's a read operation, if a python script is executed, it probably belongs in the execution category.
+
+  Categories
+
+  - `read.source`,
+  - `read.logs`
+  - `read.docs`,
+  - `read.other`. This category should be very infrequent and only for read targets that are clearly not compatible with the others.
+
+  depending on what is being read.
+
+  ## Write operations
+
+  The model modifies files. Common commands include `cat ... > file`, `sed`, etc. Creating directories also falls into this category.
+
+  Subcatgories:
+
+  - `write.docs`: Documentation
+  - `write.source.main`: Writing of the main player code. Does NOT include writing simple bots to test again, but only editing the main player/agent/bot file (`main.py`, `player.py`, `robot.js`, `warrior.red`, `robots/custom/`)
+  - `write.source.opponent`: Writing of opponents to test the main player/agent/bot against.
+  - `write.source.analysis`: Writing of analysis scripts, especially to parse logs or analyze what is happening in the game
+  - `write.source.tests`: Writing of unit test scripts. Unit tests are different from analsis, because they have a predefined, very clear pass or fail outcome (i.e., assert statements)
+  - `write.other`:  This category should be very infrequent and only for write targets that are clearly not compatible with the others.
+
+  ## Execution operations
+
+  Executions are anything that executes source files, especially executing analysis scripts, playing the game with different players, etc.
+
+  - `execute.game`: Calling on the game executable to run a game between different players.
+  - `execute.game.setup`: Preparations for running the game, for example if the player servers need to be started first, or the game needs to be compiled etc., this also belongs in this
+  - `execute.analysis`: Executing analysis scripts (see previous notes on difference between unittests and analysis)
+  - `execute.unittest`: Executing unittests or simple tests (import checks etc.). Compilation checks also fall into this category.
+
+  ## Other
+
+  - `submit`: The player issues "MINI_SWE_AGENT_FINAL_OUTPUT", "COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT" to finish the run. If this is combined with another action, categorize the other action instead. Only use this category if it's a standalone request to finish.
+  - `other`: This category should be very infrequent and only for write targets that are clearly not compatible with the others.
+
+  # Important notes
+
+  1. You MUST (!) categorize EVERY (!) action. Do NOT (!) skip any action.
+  2. Every action MUST (!) be put into exactly (!) one (!) category.
+  3. Your category MUST (!) be one of the list above.
+
+  In order of importance: execution is more important than writing is more important than reading.
+  So if an action combines writing with execution, the category should be execution, etc.

codeclash/analysis/llm_as_judge/categorize_actions.yaml

@@ -0,0 +1,50 @@
+import fcntl
+from dataclasses import dataclass
+from pathlib import Path
+
+
+class FileLock:
+    def __init__(self, lock_path: str | Path):
+        self.lock_path = Path(lock_path)
+        self.lock_file = None
+
+    def __enter__(self):
+        self.lock_file = open(self.lock_path, "w")
+        fcntl.flock(self.lock_file.fileno(), fcntl.LOCK_EX)
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self.lock_file:
+            fcntl.flock(self.lock_file.fileno(), fcntl.LOCK_UN)
+            self.lock_file.close()
+        return False
+
+
+@dataclass
+class Instance:
+    player_name: str
+    round_number: int
+    tournament_name: str
+    trajectory_path: Path
+
+    @property
+    def instance_id(self) -> str:
+        return f"{self.tournament_name}.{self.player_name}.{self.round_number}"
+
+
+def find_tournament_folders(input_dir: Path) -> list[Path]:
+    return [d.parent for d in input_dir.rglob("metadata.json")]
+
+
+def parse_trajectory_name(trajectory_path: Path) -> Instance:
+    return Instance(
+        trajectory_path=trajectory_path,
+        player_name=trajectory_path.parent.name,
+        round_number=int(trajectory_path.name.split(".")[0].split("_r")[1]),
+        tournament_name=trajectory_path.parent.parent.parent.name,
+    )
+
+
+def get_instances(input_folder: Path) -> list[Instance]:
+    trajectories = input_folder.rglob("*.traj.json")
+    return [parse_trajectory_name(trajectory) for trajectory in trajectories]