import os import re import gradio as gr import requests import inspect import pandas as pd # from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI # from llama_index.core.agent.workflow import AgentWorkflow # from llama_index.core.tools import FunctionTool from agent_llama import all_tools from agent_graph import build_graph from langchain_core.messages import HumanMessage from langchain_google_genai import ChatGoogleGenerativeAI # --- Constants --- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" # --- Basic Agent Definition --- # A custom agent class that wraps an LLM and agent workflow from llama index # class BasicAgent: # def __init__(self): # print("BasicAgent initialized.") # self.llm = HuggingFaceInferenceAPI(model_name="HuggingFaceH4/zephyr-7b-beta") # self.agent = AgentWorkflow.from_tools_or_functions( # all_tools, # make sure all_tools are sync functions # llm=self.llm, # system_prompt="You are a general AI assistant. Think step-by-step, and return only the final answer on the last line." # ) # def __call__(self, question: str) -> str: # try: # response = self.agent.run(question) # sync version of arun() # return str(response) # except Exception as e: # return f"Agent error: {e}" # Using LangGraph class BasicAgent: """A langgraph agent.""" def __init__(self): print("BasicAgent initialized.") self.graph = build_graph() def __call__(self, question: str) -> str: print(f"Agent received question (first 50 chars): {question[:50]}...") user_message = [HumanMessage(content=question)] result = self.graph.invoke({"messages": user_message}) answer = result['messages'][-1].content # Use regex to extract only the final answer match = re.search(r"FINAL ANSWER:\s*(.*)", answer) return match.group(1).strip() if match else answer def run_and_submit_all(profile: gr.OAuthProfile | None): """ Runs agent across GAIA questions, submits the answers and returns the results """ # Retrive HF space ID from enviornment variables space_id = os.getenv("SPACE_ID") # Check if user is logged in if profile: # Populated after log in gr.LoginButton username = f"{profile.username}" print(f"User logged in: {username}") else: print("User not logged in.") return "Please Login to Hugging Face with the button.", None # Initialize GAIA question and submission urls api_url = DEFAULT_API_URL questions_url = f"{api_url}/questions" submit_url = f"{api_url}/submit" # Initialize agent try: agent = BasicAgent() except Exception as e: print(f"Error instantiating agent: {e}") return f"Error initializing agent: {e}", None # Initialize agent repository to be used for agent code agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" print(agent_code) # Fetching questions print(f"Fetching questions from: {questions_url}") try: response = requests.get(questions_url, timeout=15) response.raise_for_status() # JSON containing GAIA questions questions_data = response.json() # Guard clause - Check for empty list or errors 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 Exception as e: print(f"Error fetching questions: {e}") return f"Error fetching questions: {e}", None # Initialize empty logs for results and answers to be submitted answers_payload = [] # Task ID + Submitted Answer - Used for evaluation results_log = [] # Task ID + Question + Submitted Answer - Used for display # Run agent on questions print(f"Running agent on {len(questions_data)} questions...") # For loop to pull individual questions as agent input for item in questions_data: 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}") continue try: # Submit question to agent 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}") results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"}) 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) # Initialize submission data submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload} print(f"Submitting {len(answers_payload)} answers to: {submit_url}") # POST submission data 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', '?')}/{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 Exception as e: status_message = f"Submission Failed: {e}" print(status_message) results_df = pd.DataFrame(results_log) return status_message, results_df # Initialize Gradio app with gr.Blocks() as demo: # Markdown text blocks gr.Markdown("# Basic Agent Evaluation Runner") gr.Markdown( """ **Instructions:** 1. Log in to your Hugging Face account using the button below. 2. Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, and see the score. """ ) # Adds a login button for authentication gr.LoginButton() # A button that triggers evaluation logic when clicked run_button = gr.Button("Run Evaluation & Submit All Answers") status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False) # Non interactive textbox to show result status results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True) #Gives run_button.click( fn=run_and_submit_all, outputs=[status_output, results_table] ) if __name__ == "__main__": print("\n" + "-"*30 + " App Starting " + "-"*30) space_host_startup = os.getenv("SPACE_HOST") space_id_startup = os.getenv("SPACE_ID") 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(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("🔧 Running startup checks...\n") # Check WikipediaLoader try: from langchain_community.document_loaders import WikipediaLoader print("✅ WikipediaLoader imported successfully.") # Try fetching a test page test_docs = WikipediaLoader(query="Alan Turing", load_max_docs=1).load() if test_docs and test_docs[0].page_content.strip(): print("✅ WikipediaLoader can fetch content.\n") else: print("⚠️ WikipediaLoader returned no content.\n") except Exception as e: print("❌ WikipediaLoader failed:", e, "\n") # Check Google Gemini LLM try: from langchain_google_genai import ChatGoogleGenerativeAI import os if os.getenv("GOOGLE_API_KEY"): llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0) print("✅ Google Gemini model instantiated successfully.\n") else: print("⚠️ GOOGLE_API_KEY not found in environment.\n") except Exception as e: print("❌ langchain-google-genai or Gemini setup failed:", e, "\n") print("-"*(60 + len(" App Starting ")) + "\n") print("Launching Gradio Interface for Basic Agent Evaluation...") demo.launch(debug=True, share=False)