from __future__ import annotations import argparse import json import traceback from pathlib import Path import gradio as gr from api.client import GaiaApiClient from config import get_settings from graph.builder import build_graph from graph.state import initial_state def validate_answers( answers: list[dict[str, str]], expected_count: int = 20, ) -> None: """Validate answers before saving or submitting.""" if len(answers) != expected_count: raise ValueError( f"Expected {expected_count} answers, " f"but received {len(answers)}." ) seen_task_ids: set[str] = set() empty_task_ids: list[str] = [] for index, item in enumerate(answers, start=1): task_id = str(item.get("task_id", "")).strip() submitted_answer = str( item.get("submitted_answer", "") ).strip() if not task_id: raise ValueError( f"Answer number {index} is missing task_id." ) if task_id in seen_task_ids: raise ValueError( f"Duplicate task_id detected: {task_id}" ) seen_task_ids.add(task_id) if not submitted_answer: empty_task_ids.append(task_id) if empty_task_ids: raise ValueError( "Submission stopped because these tasks have empty answers:\n" + "\n".join(empty_task_ids) ) def solve_all( username: str = "", agent_code: str = "", submit: bool = False, ) -> tuple[str, list[dict[str, str]]]: """Fetch and solve all course questions.""" settings = get_settings() client = GaiaApiClient(settings) graph = build_graph( settings=settings, api_client=client, ) questions = client.get_questions() answers: list[dict[str, str]] = [] logs: list[str] = [] for index, question in enumerate(questions, start=1): header = ( f"[{index}/{len(questions)}] " f"Solving {question.task_id}" ) print(header, flush=True) logs.append(header) try: state = initial_state(question) result = graph.invoke( state, config={ "configurable": { "thread_id": question.task_id, } }, ) answer = str( result.get("final_answer") or "" ).strip() if not answer: graph_error = result.get("error") feedback = result.get("feedback") draft_answer = result.get("draft_answer") logs.append(" ERROR: final_answer is empty") logs.append(f" graph error: {graph_error!r}") logs.append(f" feedback: {feedback!r}") logs.append(f" draft answer: {draft_answer!r}") print(" ERROR: final_answer is empty", flush=True) print(f" graph error: {graph_error!r}", flush=True) print(f" feedback: {feedback!r}", flush=True) print(f" draft answer: {draft_answer!r}", flush=True) answers.append( { "task_id": question.task_id, "submitted_answer": answer, } ) logs.append(f" answer: {answer!r}") print(f" answer: {answer!r}", flush=True) except Exception as exc: error_message = ( f" ERROR: {type(exc).__name__}: {exc}" ) logs.append(error_message) logs.append(traceback.format_exc()) print(error_message, flush=True) answers.append( { "task_id": question.task_id, "submitted_answer": "", } ) if submit: validate_answers( answers, expected_count=len(questions), ) if not username.strip(): raise ValueError( "A Hugging Face username is required." ) if not agent_code.strip(): raise ValueError( "A public Space URL ending in /tree/main " "is required." ) response = client.submit( username=username, agent_code=agent_code, answers=answers, ) response_text = json.dumps( response, ensure_ascii=False, indent=2, ) logs.append( "Submission response:\n" + response_text ) print( "Submission response:\n" + response_text, flush=True, ) return "\n".join(logs), answers def submit_existing_answers( input_path: str, username: str, agent_code: str, ) -> dict: """Submit an existing answers file without rerunning the agent.""" path = Path(input_path) if not path.exists(): raise FileNotFoundError( f"Answers file does not exist: {path}" ) with path.open("r", encoding="utf-8") as file: answers = json.load(file) if not isinstance(answers, list): raise ValueError( "The answers file must contain a JSON list." ) validate_answers(answers) client = GaiaApiClient(get_settings()) return client.submit( username=username, agent_code=agent_code, answers=answers, ) def run_ui() -> None: settings = get_settings() with gr.Blocks( title="Vertex Agent – GAIA" ) as demo: gr.Markdown( "# Vertex Agent\n" "Solve the GAIA course questions, " "review the answers, then submit." ) username = gr.Textbox( label="Hugging Face username" ) agent_code = gr.Textbox( label="Public Space code URL", value=settings.agent_code, ) submit = gr.Checkbox( label="Submit after solving", value=False, ) run_button = gr.Button( "Run agent", variant="primary", ) output = gr.Textbox( label="Run log", lines=20, ) answers_output = gr.JSON( label="Answers" ) run_button.click( fn=solve_all, inputs=[ username, agent_code, submit, ], outputs=[ output, answers_output, ], ) demo.launch() def run_cli(args: argparse.Namespace) -> None: if args.submit_existing: result = submit_existing_answers( input_path=args.output, username=args.username.strip(), agent_code=args.agent_code.strip(), ) print( json.dumps( result, ensure_ascii=False, indent=2, ) ) return log, answers = solve_all( username=args.username, agent_code=args.agent_code, submit=args.submit, ) output_path = Path(args.output) output_path.write_text( json.dumps( answers, ensure_ascii=False, indent=2, ), encoding="utf-8", ) print(f"Saved answers to {output_path}") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Run the Vertex GAIA Agent" ) parser.add_argument( "--cli", action="store_true", help="Run without Gradio", ) parser.add_argument( "--submit", action="store_true", help="Solve all questions and submit afterward", ) parser.add_argument( "--submit-existing", action="store_true", help=( "Submit the existing answers.json file " "without solving again" ), ) parser.add_argument( "--username", default="", help="Hugging Face username", ) parser.add_argument( "--agent-code", default="", help="Public Space URL ending in /tree/main", ) parser.add_argument( "--output", default="answers.json", help="Answers JSON file", ) return parser.parse_args() if __name__ == "__main__": parsed = parse_args() if parsed.cli: run_cli(parsed) else: run_ui()