"""Improved AI Agent Evaluation Runner using only free Hugging Face models.""" import os from dotenv import load_dotenv load_dotenv() import logging from typing import Optional, Tuple, List import gradio as gr import requests import pandas as pd from langchain_core.messages import HumanMessage from agent import build_graph # Logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" MAX_RETRIES = 3 TIMEOUT = 30 class AgentEvaluator: """Evaluates the agent with robust error handling and progress tracking.""" def __init__(self): self.agent = None self.results_log = [] def initialize_agent(self) -> Tuple[bool, str]: """Initialize agent with robust error handling.""" try: logger.info("Initializing agent...") self.agent = BasicAgent() return True, "Agent initialized successfully" except Exception as e: error_msg = f"Failed to initialize agent: {str(e)}" logger.error(error_msg) return False, error_msg def fetch_questions(self, api_url: str) -> Tuple[bool, List[dict], str]: """Fetch questions from the evaluation server.""" questions_url = f"{api_url}/questions" for attempt in range(MAX_RETRIES): try: logger.info(f"Fetching questions (attempt {attempt + 1}/{MAX_RETRIES})") response = requests.get(questions_url, timeout=TIMEOUT) response.raise_for_status() questions_data = response.json() if not questions_data: return False, [], "No questions received from server" logger.info(f"Successfully fetched {len(questions_data)} questions") return True, questions_data, f"Fetched {len(questions_data)} questions" except requests.exceptions.Timeout: logger.warning(f"Timeout on attempt {attempt + 1}") if attempt == MAX_RETRIES - 1: return False, [], "Request timed out after multiple attempts" except requests.exceptions.RequestException as e: logger.error(f"Request failed on attempt {attempt + 1}: {e}") if attempt == MAX_RETRIES - 1: return False, [], f"Failed to fetch questions: {str(e)}" return False, [], "Unexpected error in fetch_questions" def process_questions(self, questions_data: List[dict], progress_callback=None) -> Tuple[List[dict], List[dict]]: """Process each question and track progress.""" results_log = [] answers_payload = [] total_questions = len(questions_data) for i, item in enumerate(questions_data): task_id = item.get("task_id") question_text = item.get("question") if not task_id or question_text is None: logger.warning(f"Skipping invalid item: {item}") continue try: if progress_callback: progress = (i + 1) / total_questions progress_callback(progress, f"Processing question {i + 1}/{total_questions}") logger.info(f"Processing question {i + 1}/{total_questions}: {task_id}") submitted_answer = self.agent(question_text) answers_payload.append({ "task_id": task_id, "submitted_answer": submitted_answer }) results_log.append({ "Task ID": task_id, "Question": question_text[:100] + "..." if len(question_text) > 100 else question_text, "Submitted Answer": submitted_answer, "Status": "✅ Success" }) except Exception as e: error_msg = f"ERROR: {str(e)}" logger.error(f"Failed to process question {task_id}: {e}") results_log.append({ "Task ID": task_id, "Question": question_text[:100] + "..." if len(question_text) > 100 else question_text, "Submitted Answer": error_msg, "Status": "❌ Failed" }) return results_log, answers_payload def submit_answers(self, answers_payload: List[dict], username: str, agent_code: str, api_url: str) -> Tuple[bool, str]: """Submit answers and report results.""" if not answers_payload: return False, "No answers to submit" submit_url = f"{api_url}/submit" submission_data = { "username": username.strip(), "agent_code": agent_code, "answers": answers_payload } try: logger.info(f"Submitting {len(answers_payload)} answers for user '{username}'") response = requests.post(submit_url, json=submission_data, timeout=TIMEOUT * 2) response.raise_for_status() result_data = response.json() final_status = ( f"🎉 Submission Successful!\n" f"User: {result_data.get('username', 'Unknown')}\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', 'No message received.')}" ) logger.info("Submission successful") return True, final_status 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" - {error_json.get('detail', 'Unknown error')}" except Exception: error_detail += f" - {e.response.text[:200]}" logger.error(f"Submission failed: {error_detail}") return False, f"❌ Submission Failed: {error_detail}" except Exception as e: logger.error(f"Unexpected submission error: {e}") return False, f"❌ Submission Failed: {str(e)}" class BasicAgent: """Wraps the build_graph() agent and guarantees FINAL ANSWER formatting.""" def __init__(self): logger.info("Initializing BasicAgent...") try: self.graph = build_graph() logger.info("Agent graph built successfully") except Exception as e: logger.error(f"Failed to build agent graph: {e}") raise def __call__(self, question: str) -> str: """Process a question and always output FINAL ANSWER.""" if not question.strip(): return "Error: Empty question provided" try: logger.debug(f"Processing question: {question[:50]}...") messages = [HumanMessage(content=question)] result = self.graph.invoke({"messages": messages}) if not result or 'messages' not in result or not result['messages']: return "Error: No response from agent" answer = result['messages'][-1].content # Guarantee FINAL ANSWER: if "FINAL ANSWER:" not in answer: answer = f"FINAL ANSWER: {answer}" logger.debug(f"Agent response: {answer[:50]}...") return answer except Exception as e: error_msg = f"Agent processing error: {str(e)}" logger.error(error_msg) return error_msg def run_evaluation_async(profile: gr.OAuthProfile) -> Tuple[str, Optional[pd.DataFrame]]: """Orchestrates the entire evaluation process with the current logged-in user.""" if not profile: return "❌ Please log in to Hugging Face to continue.", None username = profile.username space_id = os.getenv("SPACE_ID", "unknown-space") agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" api_url = DEFAULT_API_URL evaluator = AgentEvaluator() # 1. Initialize agent success, message = evaluator.initialize_agent() if not success: return f"❌ {message}", None # 2. Fetch questions success, questions_data, message = evaluator.fetch_questions(api_url) if not success: return f"❌ {message}", None # 3. Process questions try: results_log, answers_payload = evaluator.process_questions(questions_data) if not answers_payload: return "❌ No valid answers generated", pd.DataFrame(results_log) # 4. Submit answers success, final_status = evaluator.submit_answers( answers_payload, username, agent_code, api_url ) results_df = pd.DataFrame(results_log) return final_status, results_df except Exception as e: logger.error(f"Evaluation process failed: {e}") return f"❌ Evaluation failed: {str(e)}", None # ----------------------- UI -------------------------- with gr.Blocks(title="AI Agent Evaluator", theme=gr.themes.Soft()) as demo: gr.Markdown(""" # 🤖 AI Agent Evaluation System **Welcome to the AI Agents Course Unit 4 Assignment!** This system evaluates your AI agent's performance on a variety of tasks including: - 🔍 Research and information retrieval - 🧮 Mathematical calculations - 📊 Data analysis - 🌐 Web search and synthesis ### Instructions: 1. **Clone this space** and modify the agent logic in `agent.py` 2. **Log in** to your Hugging Face account below 3. **Run the evaluation** to test your agent's performance ### What happens when you click "Run Evaluation"? - Fetches test questions from the evaluation server - Runs your agent on each question - Submits answers and receives a score - Shows detailed results for analysis """) with gr.Row(): gr.LoginButton() with gr.Row(): run_btn = gr.Button( "🚀 Run Evaluation & Submit Answers", variant="primary", size="lg" ) with gr.Row(): with gr.Column(): status_output = gr.Textbox( label="📋 Evaluation Status", lines=8, interactive=False, placeholder="Click 'Run Evaluation' to start..." ) with gr.Row(): results_table = gr.DataFrame( label="📊 Detailed Results", wrap=True, interactive=False ) run_btn.click( fn=run_evaluation_async, outputs=[status_output, results_table], show_progress=True ) gr.Markdown(""" ### 💡 Tips for Success: - **Understand the task**: Each question tests different capabilities - **Implement tools**: Use search, calculation, and retrieval tools effectively - **Handle errors gracefully**: Make your agent robust to unexpected inputs - **Optimize for accuracy**: Focus on getting the right answers consistently ### 🔧 Technical Notes: - Your agent code should be in `agent.py` - Modify the `build_graph()` function to implement your logic - Use the provided tools or add your own - Follow the required answer format for best results """) if __name__ == "__main__": logger.info("Starting AI Agent Evaluation System...") space_host = os.getenv("SPACE_HOST") space_id = os.getenv("SPACE_ID") if space_host: logger.info(f"✅ Running on Hugging Face Spaces: https://{space_host}.hf.space") else: logger.info("ℹī¸ Running locally") if space_id: logger.info(f"📁 Repository: https://huggingface.co/spaces/{space_id}") else: logger.warning("⚠ī¸ SPACE_ID not found - repository links may not work") demo.launch( debug=True, share=False, show_error=True, server_name="0.0.0.0" if space_host else "127.0.0.1" )