Auto-GPT/agbenchmark/utils.py

# radio charts, logs, helper functions for tests, anything else relevant.
import glob
import math
import os
import re
import sys
from datetime import datetime
from pathlib import Path
from typing import Any

from dotenv import load_dotenv

load_dotenv()

from agbenchmark.challenges.define_task_types import DIFFICULTY_MAP, DifficultyLevel

AGENT_NAME = os.getenv("AGENT_NAME")
HOME_ENV = os.getenv("HOME_ENV")


def calculate_info_test_path(reports_path: Path) -> str:
    report_location = os.getenv("REPORT_LOCATION", ".")
    if report_location:
        reports_path = Path(os.getcwd()) / report_location

    command = sys.argv

    if not reports_path.exists():
        reports_path.mkdir(parents=True, exist_ok=True)

    json_files = glob.glob(str(reports_path / "*.json"))

    # Default naming scheme
    file_count = len(json_files)
    run_name = f"file{file_count + 1}_{datetime.now().strftime('%m-%d-%H-%M')}.json"

    # # If "--test" is in command
    if "--test" in command:
        test_index = command.index("--test")
        try:
            test_arg = command[test_index + 1]  # Argument after --test
        except IndexError:
            raise ValueError("Expected an argument after --test")

        # Get all files that include the string that is the argument after --test
        related_files = [f for f in json_files if test_arg in f]
        related_file_count = len(related_files)

        # Determine the prefix based on the existing files
        if related_file_count == 0:
            # Try to find the highest prefix number among all files, then increment it
            all_prefix_numbers = []
            for f in json_files:
                number = float(Path(f).stem.split("_")[0])
                all_prefix_numbers.append(math.floor(number))

            max_prefix = max(all_prefix_numbers, default=0)
            run_name = f"{max_prefix + 1}_{test_arg}.json"
        else:
            # Take the number from before the _ and add the .{number}
            prefix_str = Path(related_files[0]).stem.rsplit("_", 1)[0].split(".")[0]
            prefix = math.floor(float(prefix_str))
            run_name = f"{prefix}.{related_file_count}_{test_arg}.json"

    print("run_namerun_namerun_name", run_name)
    new_file_path = reports_path / run_name
    return str(new_file_path)


def replace_backslash(value: Any) -> Any:
    if isinstance(value, str):
        return re.sub(
            r"\\+", "/", value
        )  # replace one or more backslashes with a forward slash
    elif isinstance(value, list):
        return [replace_backslash(i) for i in value]
    elif isinstance(value, dict):
        return {k: replace_backslash(v) for k, v in value.items()}
    else:
        return value


def calculate_success_percentage(results: list[bool]) -> float:
    # Take the last 10 results or all if less than 10
    last_results = results[-10:] if len(results) > 10 else results
    success_count = last_results.count(True)
    total_count = len(last_results)
    if total_count == 0:
        return 0
    success_percentage = (success_count / total_count) * 100  # as a percentage
    return round(success_percentage, 2)


def get_highest_success_difficulty(data: dict) -> str:
    highest_difficulty = None
    highest_difficulty_level = 0

    for test_name, test_data in data.items():
        if test_data["metrics"]["success"]:
            # Replace 'medium' with 'intermediate' for this example
            difficulty_str = test_data["metrics"]["difficulty"]

            try:
                difficulty_enum = DifficultyLevel[difficulty_str.lower()]
                difficulty_level = DIFFICULTY_MAP[difficulty_enum]

                if difficulty_level > highest_difficulty_level:
                    highest_difficulty = difficulty_enum
                    highest_difficulty_level = difficulty_level
            except KeyError:
                print(
                    f"Unexpected difficulty level '{difficulty_str}' in test '{test_name}'"
                )

    if highest_difficulty is not None:
        highest_difficulty_str = highest_difficulty.name  # convert enum to string
    else:
        highest_difficulty_str = ""

    if highest_difficulty_level:
        return f"{highest_difficulty_str}: {highest_difficulty_level}"
    return "No successful tests"


def assign_paths(folder_path: Path) -> tuple[str, str, str]:
    CONFIG_PATH = str(folder_path / "config.json")
    REGRESSION_TESTS_PATH = str(folder_path / "regression_tests.json")

    if HOME_ENV == "ci" and AGENT_NAME:
        INFO_TESTS_PATH = calculate_info_test_path(
            Path(os.getcwd()) / "agbenchmark" / "reports" / AGENT_NAME
        )
    else:
        INFO_TESTS_PATH = calculate_info_test_path(folder_path / "reports")

    return CONFIG_PATH, REGRESSION_TESTS_PATH, INFO_TESTS_PATH


def calculate_dynamic_paths() -> tuple[Path, str, str, str]:
    # the default home is where you're running from
    HOME_DIRECTORY = Path(os.getcwd())
    benchmarks_folder_path = HOME_DIRECTORY / "agbenchmark"

    if AGENT_NAME and HOME_ENV == "ci":
        if "/Auto-GPT-Benchmarks/agent" in str(HOME_DIRECTORY):
            raise Exception("Must run from root of benchmark repo if HOME_ENV is ci")

        # however if the env is local and the agent name is defined, we want to run that agent from the repo and then get the data in the internal agbenchmark directory
        # this is for the ci/cd pipeline
        benchmarks_folder_path = HOME_DIRECTORY / "agent" / AGENT_NAME / "agbenchmark"

        CONFIG_PATH, REGRESSION_TESTS_PATH, INFO_TESTS_PATH = assign_paths(
            benchmarks_folder_path
        )

        # we want to run the agent from the submodule
        HOME_DIRECTORY = Path(os.getcwd()) / "agent" / AGENT_NAME

    elif AGENT_NAME and not os.path.join("Auto-GPT-Benchmarks", "agent") in str(
        HOME_DIRECTORY
    ):
        # if the agent name is defined but the run is not from the agent repo, then home is the agent repo
        # used for development of both a benchmark and an agent
        HOME_DIRECTORY = Path(os.getcwd()) / "agent" / AGENT_NAME
        benchmarks_folder_path = HOME_DIRECTORY / "agbenchmark"

        CONFIG_PATH, REGRESSION_TESTS_PATH, INFO_TESTS_PATH = assign_paths(
            benchmarks_folder_path
        )

    else:
        # otherwise the default is when home is an agent (running agbenchmark from agent/agent_repo)
        # used when its just a pip install
        CONFIG_PATH, REGRESSION_TESTS_PATH, INFO_TESTS_PATH = assign_paths(
            benchmarks_folder_path
        )

    if not benchmarks_folder_path.exists():
        benchmarks_folder_path.mkdir(exist_ok=True)

    return (
        HOME_DIRECTORY,
        CONFIG_PATH,
        REGRESSION_TESTS_PATH,
        INFO_TESTS_PATH,
    )