Spaces:
Sleeping
Sleeping
| import os | |
| import gradio as gr | |
| import requests | |
| import inspect | |
| import pandas as pd | |
| import asyncio | |
| import aiohttp | |
| import time | |
| import random | |
| import json | |
| from smolagents import FinalAnswerTool, Tool, tool, OpenAIServerModel, DuckDuckGoSearchTool, CodeAgent, VisitWebpageTool | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| # (Keep Constants as is) | |
| # --- Constants --- | |
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" | |
| OPENAI_TOKEN = os.getenv("OPENAI_API_KEY") | |
| # --- Custom Tools --- | |
| class ReliableSearchTool(Tool): | |
| name = "reliable_search" | |
| description = "Search the web for information with built-in retry and fallback mechanisms" | |
| inputs = {"query": {"type": "string", "description": "The search query to look up"}} | |
| output_type = "string" | |
| def __init__(self): | |
| super().__init__() | |
| self.ddg_tool = DuckDuckGoSearchTool() | |
| self.max_retries = 2 | |
| self.timeout = 8 | |
| self.is_initialized = True | |
| def forward(self, query: str) -> str: | |
| """Search the web with retry logic and fallbacks.""" | |
| # Try Wikipedia first | |
| try: | |
| import requests | |
| wiki_url = "https://en.wikipedia.org/api/rest_v1/page/summary/" + query.replace(" ", "_") | |
| response = requests.get(wiki_url, timeout=5) | |
| if response.status_code == 200: | |
| data = response.json() | |
| if 'extract' in data and data['extract']: | |
| return f"Wikipedia: {data['extract']}" | |
| except Exception as e: | |
| print(f"Wikipedia search failed: {e}") | |
| # Try Wikipedia search API | |
| try: | |
| search_url = "https://en.wikipedia.org/api/rest_v1/page/search/" + query.replace(" ", "%20") | |
| response = requests.get(search_url, timeout=5) | |
| if response.status_code == 200: | |
| data = response.json() | |
| if 'pages' in data and data['pages']: | |
| return f"Wikipedia search: {data['pages'][0].get('description', 'No description available')}" | |
| except Exception as e: | |
| print(f"Wikipedia search API failed: {e}") | |
| # Fallback to DuckDuckGo | |
| try: | |
| result = self.ddg_tool(query) | |
| if result and len(result) > 50: | |
| return result | |
| except Exception as e: | |
| print(f"DuckDuckGo search failed: {e}") | |
| return f"Search unavailable for '{query}'. Please use your existing knowledge to answer." | |
| # --- Basic Agent Definition --- | |
| # ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------ | |
| class SlpMultiAgent: | |
| def __init__(self): | |
| print("BasicAgent initialized.") | |
| async def __call__(self, question: str) -> str: | |
| print(f"Agent received question (first 50 chars): {question[:50]}...") | |
| fixed_answer = "This is a default answer." | |
| print(f"Agent returning fixed answer: {fixed_answer}") | |
| # Truncate question to avoid exceeding model context length | |
| MAX_QUESTION_LENGTH = 1000 | |
| short_question = question # [:MAX_QUESTION_LENGTH] | |
| # Use GPT-3.5-turbo model with optimized settings | |
| model = OpenAIServerModel( | |
| model_id="gpt-3.5-turbo-16k", | |
| temperature=0.1, # Slight randomness for better reasoning | |
| max_tokens=1000 # Keep higher tokens for complex reasoning | |
| ) | |
| # Here you can implement your agent logic, tools, and model calls | |
| web_agent = CodeAgent( | |
| tools=[ReliableSearchTool(), VisitWebpageTool()], # Use custom reliable search tool | |
| model=model, | |
| additional_authorized_imports=["pandas", "time", "json", "requests", "urllib"], | |
| max_steps=5, # Keep sufficient steps for complex tasks | |
| name="WebAgent", | |
| verbosity_level=0, | |
| description="An agent that can search the web and visit webpages to find information." | |
| ) | |
| manager_agent = CodeAgent( | |
| model=OpenAIServerModel( | |
| model_id="gpt-3.5-turbo-16k", | |
| temperature=0.1, | |
| max_tokens=1000 | |
| ), | |
| tools=[], | |
| managed_agents=[web_agent], | |
| name="ManagerAgent", | |
| description="A manager agent that can delegate tasks to other agents and manage their execution.", | |
| additional_authorized_imports=[ | |
| "pandas", | |
| "time", | |
| "json", | |
| "requests" | |
| ], | |
| planning_interval=3, | |
| verbosity_level=1, | |
| max_steps=10, # Keep full steps for complex reasoning | |
| final_answer_checks=[check_reasoning] | |
| ) | |
| # Create a task for the agent run with retry mechanism for rate limits | |
| max_retries = 3 | |
| result = None | |
| for attempt in range(max_retries): | |
| try: | |
| loop = asyncio.get_event_loop() | |
| result = await loop.run_in_executor( | |
| None, | |
| lambda: manager_agent.run(f""" | |
| Question: {short_question} | |
| Answer this question step by step. When you need to write code, use this exact format: | |
| <code> | |
| # Your Python code here | |
| </code> | |
| When you have the final answer, use: | |
| <code> | |
| final_answer("Your answer here") | |
| </code> | |
| Be precise and factual. Use search tools only when needed. | |
| """) | |
| ) | |
| break # Success, exit retry loop | |
| except Exception as e: | |
| print(f"Attempt {attempt+1}/{max_retries} failed: {e}") | |
| if "rate limit" in str(e).lower() and attempt < max_retries - 1: | |
| # Add jitter to avoid synchronized retries | |
| wait_time = (attempt + 1) * 10 + random.uniform(0, 5) | |
| print(f"Rate limit hit. Waiting {wait_time:.2f} seconds before retry...") | |
| await asyncio.sleep(wait_time) | |
| elif attempt < max_retries - 1: | |
| await asyncio.sleep(5) # Wait before general retry | |
| else: | |
| print(f"All attempts failed. Returning default answer.") | |
| return "I apologize, but I'm currently experiencing technical difficulties. Please try again later." | |
| # If we couldn't get a result after all retries | |
| if result is None: | |
| return "I apologize, but I'm currently experiencing technical difficulties. Please try again later." | |
| # Return the result from the agent | |
| return result | |
| def check_reasoning(final_answer, agent_memory): | |
| try: | |
| multimodal_model = OpenAIServerModel( | |
| model_id="gpt-3.5-turbo", | |
| max_tokens=100 # Reduced tokens for cost efficiency | |
| ) | |
| # More focused validation prompt | |
| prompt = f"Rate answer quality 1-10: {final_answer[:200]}..." | |
| messages = [ | |
| { | |
| "role": "user", | |
| "content": prompt | |
| } | |
| ] | |
| # Add retry mechanism for rate limits | |
| max_retries = 2 # Reduced retries | |
| for attempt in range(max_retries): | |
| try: | |
| output = multimodal_model(messages) | |
| if hasattr(output, 'content'): | |
| # Actually check the response instead of always returning True | |
| response = output.content.lower() | |
| # Look for quality indicators | |
| if any(word in response for word in ['7', '8', '9', '10', 'good', 'correct']): | |
| return True | |
| elif any(word in response for word in ['1', '2', '3', '4', 'poor', 'wrong']): | |
| return False | |
| return True # Default to pass if unclear | |
| break | |
| except Exception as e: | |
| if attempt < max_retries - 1: | |
| print(f"Retry {attempt+1}/{max_retries} due to: {e}") | |
| time.sleep(3) # Reduced wait time | |
| else: | |
| print(f"Final attempt failed: {e}") | |
| return True # Default to passing if we can't check properly | |
| except Exception as e: | |
| print(f"Error in reasoning check: {e}") | |
| return True # Default to passing on errors | |
| async def run_and_submit_all(profile): | |
| """ | |
| Fetches all questions, runs the BasicAgent on them, submits all answers, | |
| and displays the results asynchronously. | |
| """ | |
| # --- Determine HF Space Runtime URL and Repo URL --- | |
| space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code | |
| # Handle different profile types | |
| if profile: | |
| if hasattr(profile, 'username'): | |
| # It's an OAuthProfile object | |
| username = profile.username | |
| else: | |
| # It's a string or other type | |
| username = str(profile) | |
| print(f"User logged in: {username}") | |
| else: | |
| print("User not logged in.") | |
| return "Please Login to Hugging Face with the button.", None | |
| api_url = DEFAULT_API_URL | |
| questions_url = f"{api_url}/questions" | |
| submit_url = f"{api_url}/submit" | |
| # 1. Instantiate Agent ( modify this part to create your agent) | |
| try: | |
| agent = SlpMultiAgent() | |
| except Exception as e: | |
| print(f"Error instantiating agent: {e}") | |
| return f"Error initializing agent: {e}", None | |
| # In the case of an app running as a hugging Face space, this link points toward your codebase ( usefull for others so please keep it public) | |
| agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" | |
| print(agent_code) | |
| # 2. Fetch Questions | |
| print(f"Fetching questions from: {questions_url}") | |
| try: | |
| async with aiohttp.ClientSession() as session: | |
| async with session.get(questions_url, timeout=15) as response: | |
| response.raise_for_status() | |
| questions_data = await response.json() | |
| if not questions_data: | |
| print("Fetched questions list is empty.") | |
| return "Fetched questions list is empty or invalid format.", None | |
| print(f"Fetched {len(questions_data)} questions.") | |
| except aiohttp.ClientError as e: | |
| print(f"Error fetching questions: {e}") | |
| return f"Error fetching questions: {e}", None | |
| except ValueError as e: # JSON decode error | |
| print(f"Error decoding JSON response from questions endpoint: {e}") | |
| return f"Error decoding server response for questions: {e}", None | |
| except Exception as e: | |
| print(f"An unexpected error occurred fetching questions: {e}") | |
| return f"An unexpected error occurred fetching questions: {e}", None | |
| # 3. Run your Agent | |
| results_log = [] | |
| answers_payload = [] | |
| print(f"Running agent on {len(questions_data)} questions...") | |
| # Process questions with optimized concurrency | |
| semaphore = asyncio.Semaphore(2) # Process 2 questions at a time for better efficiency | |
| async def process_question(item): | |
| task_id = item.get("task_id") | |
| question_text = item.get("question") | |
| if not task_id or question_text is None: | |
| print(f"Skipping item with missing task_id or question: {item}") | |
| return None | |
| async with semaphore: | |
| max_retries = 3 | |
| for attempt in range(max_retries): | |
| try: | |
| print(f"Processing task {task_id}, attempt {attempt+1}/{max_retries}") | |
| submitted_answer = await agent(question_text) | |
| return {"task_id": task_id, "submitted_answer": submitted_answer, | |
| "log": {"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer}} | |
| except Exception as e: | |
| print(f"Error running agent on task {task_id}, attempt {attempt+1}: {e}") | |
| if "rate limit" in str(e).lower() and attempt < max_retries - 1: | |
| # Exponential backoff with jitter | |
| wait_time = (2 ** attempt) * 5 + random.uniform(0, 3) | |
| print(f"Rate limit hit. Waiting {wait_time:.2f} seconds before retry...") | |
| await asyncio.sleep(wait_time) | |
| elif attempt < max_retries - 1: | |
| await asyncio.sleep(5) # Reduced wait time | |
| else: | |
| # All retries failed, return default answer | |
| default_answer = "This is a default answer." | |
| return {"task_id": task_id, "submitted_answer": default_answer, | |
| "log": {"Task ID": task_id, "Question": question_text, "Submitted Answer": default_answer}} | |
| # Create tasks for all questions | |
| tasks = [process_question(item) for item in questions_data] | |
| results = await asyncio.gather(*tasks) | |
| # Process results | |
| for result in results: | |
| if result is not None: | |
| answers_payload.append({"task_id": result["task_id"], "submitted_answer": result["submitted_answer"]}) | |
| results_log.append(result["log"]) | |
| if not answers_payload: | |
| print("Agent did not produce any answers to submit.") | |
| return "Agent did not produce any answers to submit.", pd.DataFrame(results_log) | |
| # 4. Prepare Submission | |
| submission_data = {"username": str(username).strip(), "agent_code": agent_code, "answers": answers_payload} | |
| status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..." | |
| print(status_update) | |
| # 5. Submit | |
| print(f"Submitting {len(answers_payload)} answers to: {submit_url}") | |
| try: | |
| async with aiohttp.ClientSession() as session: | |
| async with session.post(submit_url, json=submission_data, timeout=60) as response: | |
| response.raise_for_status() | |
| result_data = await 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', '?')}/{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 aiohttp.ClientResponseError as e: | |
| error_detail = f"Server responded with status {e.status}." | |
| try: | |
| error_text = await e.response.text() | |
| try: | |
| error_json = await e.response.json() | |
| error_detail += f" Detail: {error_json.get('detail', error_text)}" | |
| except ValueError: | |
| error_detail += f" Response: {error_text[:500]}" | |
| except: | |
| pass | |
| status_message = f"Submission Failed: {error_detail}" | |
| print(status_message) | |
| results_df = pd.DataFrame(results_log) | |
| return status_message, results_df | |
| except asyncio.TimeoutError: | |
| status_message = "Submission Failed: The request timed out." | |
| print(status_message) | |
| results_df = pd.DataFrame(results_log) | |
| return status_message, results_df | |
| except aiohttp.ClientError 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 | |
| # --- Build Gradio Interface using Blocks --- | |
| with gr.Blocks() as demo: | |
| gr.Markdown("# Basic Agent Evaluation Runner") | |
| gr.Markdown( | |
| """ | |
| **Instructions:** | |
| 1. Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ... | |
| 2. Log in to your Hugging Face account using the button below. This uses your HF username for submission. | |
| 3. Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score. | |
| --- | |
| **Disclaimers:** | |
| Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions). | |
| This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async. | |
| """ | |
| ) | |
| login_button = gr.LoginButton() | |
| run_button = gr.Button("Run Evaluation & Submit All Answers") | |
| status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False) | |
| # Removed max_rows=10 from DataFrame constructor | |
| results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True) | |
| def sync_wrapper(profile): | |
| # This wrapper ensures we have access to the profile | |
| if not profile: | |
| print("No profile available in sync_wrapper") | |
| return "Please Login to Hugging Face with the button.", None | |
| print(f"Profile type in wrapper: {type(profile)}") | |
| try: | |
| return asyncio.run(run_and_submit_all(profile)) | |
| except Exception as e: | |
| print(f"Error in sync_wrapper: {e}") | |
| return f"Error processing request: {e}", None | |
| run_button.click( | |
| fn=sync_wrapper, | |
| inputs=login_button, | |
| outputs=[status_output, results_table] | |
| ) | |
| if __name__ == "__main__": | |
| print("\n" + "-"*30 + " App Starting " + "-"*30) | |
| # Check for SPACE_HOST and SPACE_ID at startup for information | |
| space_host_startup = os.getenv("SPACE_HOST") | |
| space_id_startup = os.getenv("SPACE_ID") # Get SPACE_ID at startup | |
| if space_host_startup: | |
| print(f"✅ SPACE_HOST found: {space_host_startup}") | |
| print(f" Runtime URL should be: https://{space_host_startup}.hf.space") | |
| else: | |
| print("ℹ️ SPACE_HOST environment variable not found (running locally?).") | |
| if space_id_startup: # Print repo URLs if SPACE_ID is found | |
| print(f"✅ SPACE_ID found: {space_id_startup}") | |
| print(f" Repo URL: https://huggingface.co/spaces/{space_id_startup}") | |
| print(f" Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main") | |
| else: | |
| print("ℹ️ SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined.") | |
| print("-"*(60 + len(" App Starting ")) + "\n") | |
| print("Launching Gradio Interface for Basic Agent Evaluation...") | |
| demo.launch(debug=True, share=False) |