src/gaia_agent/cli.py

import argparse
import json
import logging
from pathlib import Path
from typing import Any

from gaia_agent.agent import GaiaAgent
from gaia_agent.client import ScoringClient
from gaia_agent.models import Answer

USERNAME = "AyhamJo7"
AGENT_CODE_URL = "https://huggingface.co/spaces/AyhamJo7/Final_Assignment_Template/tree/main"


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Run the GAIA course final-assignment agent")
    subparsers = parser.add_subparsers(dest="command", required=True)
    evaluate = subparsers.add_parser("evaluate")
    evaluate.add_argument("--output", type=Path, default=Path("answers.json"))
    evaluate.add_argument("--downloads", type=Path, default=Path("downloads"))
    submit = subparsers.add_parser("submit")
    submit.add_argument("--answers", type=Path, default=Path("answers.json"))
    return parser


def evaluate(output: Path, downloads: Path) -> None:
    agent = GaiaAgent()
    answers: list[dict[str, str]] = []
    with ScoringClient() as client:
        questions = client.questions()
        for index, question in enumerate(questions, start=1):
            logging.info("Solving %s/%s: %s", index, len(questions), question.task_id)
            attachment = client.download_attachment(question, downloads)
            record = agent.solve(question, attachment)
            answers.append(
                {"task_id": question.task_id, "submitted_answer": record.submitted_answer}
            )
            _write_answers(output, answers)


def _write_answers(output: Path, answers: list[dict[str, str]]) -> None:
    output.write_text(json.dumps(answers, indent=2, ensure_ascii=True) + "\n", encoding="utf-8")


def submit(path: Path) -> None:
    raw: Any = json.loads(path.read_text(encoding="utf-8"))
    answers = [Answer.model_validate(item) for item in raw]
    with ScoringClient() as client:
        score = client.submit(USERNAME, AGENT_CODE_URL, answers)
    print(score.model_dump_json(indent=2))


def main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    args = _parser().parse_args()
    if args.command == "evaluate":
        evaluate(args.output, args.downloads)
    else:
        submit(args.answers)


if __name__ == "__main__":
    main()