from typing import Dict, List, Tuple import re import tempfile from pathlib import Path import pandas as pd import requests from agent import GaiaAgent from pandas import DataFrame # --- Constants --- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" QUESTIONS_URL = f"{DEFAULT_API_URL}/questions" SUBMIT_URL = f"{DEFAULT_API_URL}/submit" FILE_PATH = f"{DEFAULT_API_URL}/files/" # --- Helper Methods --- def fetch_all_questions() -> Dict: """Fetches all questions from the specified API endpoint. This function retrieves a list of questions from the API, handles potential errors such as network issues, invalid responses, or empty question lists, and returns the questions as a dictionary. Returns: Dict: A dictionary containing the questions data retrieved from the API. Raises: UserWarning: If there is an error fetching the questions, such as network issues, invalid JSON response, or an empty question list. The exception message provides details about the specific error encountered. """ print(f"Fetching questions from: {QUESTIONS_URL}") response = requests.get(QUESTIONS_URL, timeout=15) try: response.raise_for_status() questions_data = response.json() if not questions_data: print("Fetched questions list is empty.") raise UserWarning("Fetched questions list is empty or invalid format.") print(f"Fetched {len(questions_data)} questions.") return questions_data except requests.exceptions.RequestException as e: print(f"Error fetching questions: {e}") raise UserWarning(f"Error fetching questions: {e}") except requests.exceptions.JSONDecodeError as e: print(f"Error decoding JSON response from questions endpoint: {e}") print(f"Response text: {response.text[:500]}") raise UserWarning(f"Error decoding server response for questions: {e}") except Exception as e: print(f"An unexpected error occurred fetching questions: {e}") raise UserWarning(f"An unexpected error occurred fetching questions: {e}") def submit_answers(submission_data: dict, results_log: list) -> Tuple[str, DataFrame]: """Submits answers to the scoring API and returns the submission status and results. This function sends the provided answers to the scoring API, handles potential errors such as network issues, server errors, or invalid responses, and returns a status message indicating the success or failure of the submission, along with a DataFrame containing the results log. Args: submission_data (dict): A dictionary containing the answers to be submitted. Expected to have a structure compatible with the scoring API. results_log (list): A list of dictionaries containing the results log. This log is converted to a Pandas DataFrame and returned. Returns: Tuple[str, DataFrame]: A tuple containing: - A status message (str) indicating the submission status and any relevant information or error messages. - A Pandas DataFrame containing the results log. """ try: response = requests.post(SUBMIT_URL, json=submission_data, timeout=60) response.raise_for_status() result_data = response.json() final_status = ( f"Submission Successful!\n" f"User: {result_data.get('username')}\n" f"Overall Score: {result_data.get('score', 'N/A')}% " f"({result_data.get('correct_count', '?')}/" f"{result_data.get('total_attempted', '?')} correct)\n" f"Message: {result_data.get('message', 'No message received.')}" ) print("Submission successful.") results_df = pd.DataFrame(results_log) return final_status, results_df except requests.exceptions.HTTPError as e: error_detail = f"Server responded with status {e.response.status_code}." try: error_json = e.response.json() error_detail += f" Detail: {error_json.get('detail', e.response.text)}" except requests.exceptions.JSONDecodeError: error_detail += f" Response: {e.response.text[:500]}" status_message = f"Submission Failed: {error_detail}" print(status_message) results_df = pd.DataFrame(results_log) return status_message, results_df except requests.exceptions.Timeout: status_message = "Submission Failed: The request timed out." print(status_message) results_df = pd.DataFrame(results_log) return status_message, results_df except requests.exceptions.RequestException as e: status_message = f"Submission Failed: Network error - {e}" print(status_message) results_df = pd.DataFrame(results_log) return status_message, results_df except Exception as e: status_message = f"An unexpected error occurred during submission: {e}" print(status_message) results_df = pd.DataFrame(results_log) return status_message, results_df def run_agent( gaia_agent: GaiaAgent, questions_data: List[Dict] ) -> Tuple[List[Dict], List[Dict]]: """Runs the agent on a list of questions and returns the results and answers. This function iterates through a list of questions, runs the provided agent on each question, and collects the results and answers. It handles potential errors during agent execution and returns the results log and the answers payload. Args: gaia_agent (GaiaAgent): An instance of the GaiaAgent class, which is responsible for generating answers to the questions. questions_data (List[Dict]): A list of dictionaries, where each dictionary represents a question and contains at least the 'task_id' and 'question' keys. Returns: Tuple[List[Dict], List[Dict]]: A tuple containing: - A list of dictionaries representing the results log, where each dictionary contains the 'Task ID', 'Question', and 'Submitted Answer'. - A list of dictionaries representing the answers payload, where each dictionary contains the 'task_id' and 'submitted_answer'. """ results_log = [] answers_payload = [] print(f"🚀 Running agent on {len(questions_data)} questions...") for item in questions_data: task_id = item.get("task_id") question_text = item.get("question") question_text = process_file(task_id, question_text) if not task_id or question_text is None: print(f"⚠️ Skipping invalid item (missing task_id or question): {item}") continue try: submitted_answer = gaia_agent(task_id, question_text) answers_payload.append( {"task_id": task_id, "submitted_answer": submitted_answer} ) except Exception as e: print(f"❌ Error running agent on task {task_id}: {e}") submitted_answer = f"AGENT ERROR: {e}" results_log.append( { "Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer, } ) return results_log, answers_payload def process_file(task_id: str, question_text: str) -> str: """ Attempt to download a file associated with a task from the API. - If the file exists (HTTP 200), it is saved to a temp directory and the local file path is returned. - If no file is found (HTTP 404), returns None. - For all other HTTP errors, the exception is propagated to the caller. """ file_url = f"{FILE_PATH}{task_id}" try: response = requests.get(file_url, timeout=30) response.raise_for_status() except requests.exceptions.RequestException as exc: print(f"Exception in download_file>> {str(exc)}") return question_text # Unable to get the file # Determine filename from 'Content-Disposition' header, fallback to task_id content_disposition = response.headers.get("content-disposition", "") filename = task_id match = re.search(r'filename="([^"]+)"', content_disposition) if match: filename = match.group(1) # Save file in a temp directory temp_storage_dir = Path(tempfile.gettempdir()) / "gaia_cached_files" temp_storage_dir.mkdir(parents=True, exist_ok=True) file_path = temp_storage_dir / filename file_path.write_bytes(response.content) return ( f"{question_text}\n\n" f"---\n" f"A file was downloaded for this task and saved locally at:\n" f"{str(file_path)}\n" f"---\n\n" )