Spaces:
Sleeping
Sleeping
| """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" | |
| ) |