import os import gradio as gr import requests import pandas as pd from dotenv import load_dotenv import traceback from typing import Optional # Keep this import, good practice # --- Constants --- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" # --- Agent Integration --- AGENT_AVAILABLE = False AGENT_LOAD_ERROR = "" AGENT_FUNCTION_NAME = "run_gaia_task" # Define the target function name try: from final_agent import run_gaia_task print(f"Successfully imported {AGENT_FUNCTION_NAME} from final_agent.py") AGENT_AVAILABLE = True except ImportError as e: error_msg = f"ERROR: Could not import {AGENT_FUNCTION_NAME} from final_agent.py: {e}" print(error_msg) AGENT_LOAD_ERROR = error_msg except Exception as e: error_msg = f"ERROR during import or initial setup in final_agent.py: {e}" print(error_msg) traceback.print_exc() AGENT_LOAD_ERROR = error_msg if not AGENT_AVAILABLE: def run_gaia_task(task_description: str) -> str: """Dummy function used when the real agent fails to load.""" print(f"Executing dummy {AGENT_FUNCTION_NAME} because agent failed to load.") return f"ERROR: Agent function '{AGENT_FUNCTION_NAME}' could not be loaded. Details: {AGENT_LOAD_ERROR}" # --- Agent Runner Class --- class AgentRunner: def __init__(self): print("AgentRunner initialized.") if not AGENT_AVAILABLE: print(f"WARNING: Agent function failed to load during startup. Error: {AGENT_LOAD_ERROR}") def __call__(self, question: str) -> str: """Runs the imported agent function on a single question.""" print(f"\n--- AgentRunner received question: {question[:100]}... ---") try: final_answer = run_gaia_task(task_description=question) final_answer_str = str(final_answer) print(f"--- AgentRunner returning answer: {final_answer_str} ---") return final_answer_str except Exception as e: print(f"!!! ERROR calling {AGENT_FUNCTION_NAME} function: {e} !!!") traceback.print_exc() return f"ERROR: Agent function '{AGENT_FUNCTION_NAME}' failed during execution - {e}" # --- Submission Logic --- def run_and_submit_all( profile: gr.OAuthProfile | None): """Fetches questions, runs agent, submits answers.""" space_id = os.getenv("SPACE_ID") if not profile: print("User not logged in."); return "Please Login.", None username= f"{profile.username}"; print(f"User logged in: {username}") api_url = DEFAULT_API_URL; questions_url = f"{api_url}/questions"; submit_url = f"{api_url}/submit" # 1. Instantiate Agent Runner try: agent = AgentRunner() if not AGENT_AVAILABLE: return f"Agent function '{AGENT_FUNCTION_NAME}' failed to load. Check logs. Error: {AGENT_LOAD_ERROR}", None except Exception as e: print(f"Error instantiating AgentRunner: {e}"); return f"Init error: {e}", None agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" if space_id else "Code URL N/A" print(f"Agent code reference: {agent_code}") # 2. Fetch Questions print(f"Fetching questions from: {questions_url}") try: response = requests.get(questions_url, timeout=30); response.raise_for_status() questions_data = response.json() if not questions_data: print("Questions list empty."); return "Questions list empty.", None print(f"Fetched {len(questions_data)} questions.") except Exception as e: print(f"Error fetching questions: {e}"); return f"Fetch error: {e}", None # 3. Run Agent on each question results_log = []; answers_payload = [] print(f"Running agent on {len(questions_data)} questions...") question_count = len(questions_data) for i, item in enumerate(questions_data): task_id = item.get("task_id"); question_text = item.get("question") print(f"\n--- Processing Question {i+1}/{question_count} (ID: {task_id}) ---") if not task_id or question_text is None: print(f"Skipping item: {item}"); continue try: submitted_answer = agent(question_text) answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer}) results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer}) except Exception as e: print(f"!! Error running agent on task {task_id}: {e} !!"); traceback.print_exc() results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT RUN ERROR: {e}"}) answers_payload.append({"task_id": task_id, "submitted_answer": f"AGENT RUN ERROR: {e}"}) if not answers_payload: print("Agent produced no answers."); return "Agent produced no answers.", pd.DataFrame(results_log) # 4. Prepare Submission submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload} print(f"\nSubmitting {len(answers_payload)} answers for user '{username}'...") # 5. Submit try: response = requests.post(submit_url, json=submission_data, timeout=120); response.raise_for_status() result_data = response.json() final_status = (f"Submission Successful!\nUser: {result_data.get('username')}\n" f"Overall Score: {result_data.get('score', 'N/A')}% " f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n" f"Message: {result_data.get('message', 'N/A')}") print("Submission successful."); results_df = pd.DataFrame(results_log); return final_status, results_df except requests.exceptions.HTTPError as e: error_detail = f"Server error {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[:200]}" status_message = f"Submission Failed: {error_detail}" except requests.exceptions.Timeout: status_message = "Submission Failed: The request timed out (120 seconds)." except Exception as e: status_message = f"Submission Failed: Unexpected error - {e}"; traceback.print_exc() print(status_message); results_df = pd.DataFrame(results_log); return status_message, results_df # --- Build Gradio Interface --- with gr.Blocks() as demo: gr.Markdown("# GAIA Agent Evaluation Runner") gr.Markdown( """ **Instructions:** 1. Ensure your agent logic is in `final_agent.py` (exposing the `run_gaia_task` function) and dependencies in `requirements.txt`. Set secrets in Space settings (GROQ_API_KEY, TAVILY_API_KEY, OPENAI_API_KEY). 2. Log in to Hugging Face using the button below. 3. Click 'Run Evaluation & Submit All Answers' to run your agent. Check Logs for detailed progress. --- **Disclaimers:** Execution can take significant time depending on the number of questions and agent complexity. """ ) login_button = gr.LoginButton() # Assign to variable to access profile info implicitly run_button = gr.Button("Run Evaluation & Submit All Answers") status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False) results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True) # --- CORRECTED LINE --- # Remove the 'inputs' argument. The profile is passed implicitly because of LoginButton. run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table]) # --- Main execution block --- if __name__ == "__main__": print("\n" + "-"*30 + " App Starting " + "-"*30) if not AGENT_AVAILABLE: print(f"CRITICAL WARNING: Agent function '{AGENT_FUNCTION_NAME}' could not be loaded. The app will run but agent calls will fail.") print(f"Load Error Details: {AGENT_LOAD_ERROR}") print("Launching Gradio Interface...") demo.launch(debug=False, share=False)