import os import gradio as gr import requests import pandas as pd from langchain_groq import ChatGroq from langchain_core.tools import tool from langchain_core.messages import HumanMessage, SystemMessage from langchain_community.tools import DuckDuckGoSearchRun from langchain_community.tools import WikipediaQueryRun from langchain_community.utilities import WikipediaAPIWrapper from langgraph.prebuilt import create_react_agent # --- Constants --- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" # --- System prompt: this is what enforces the EXACT-MATCH formatting --- SYSTEM_PROMPT = """You are a precise assistant solving questions from the GAIA benchmark. You have tools: web search, Wikipedia, and a file downloader. Use them to find facts; reason step by step. Your FINAL message must contain ONLY the answer itself — no explanation, no preamble. The grader does an EXACT string match, so follow these rules strictly: - Never write "FINAL ANSWER" or any extra words in the final message. - If the answer is a number: digits only, no thousands separators, no units (unless the question explicitly asks for a unit). - If the answer is text: as few words as possible, no leading articles, no trailing period, no abbreviations unless required by the question. - If the answer is a comma-separated list: apply the rules above to each element. """ # --- Tools --- search_tool = DuckDuckGoSearchRun() wiki_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(top_k_results=2, doc_content_chars_max=3000)) @tool def fetch_task_file(task_id: str) -> str: """Download the file attached to a GAIA task by its task_id and return its text content. Use this only when the question refers to an attached file (a spreadsheet, csv, text file, code, etc.).""" url = f"{DEFAULT_API_URL}/files/{task_id}" try: r = requests.get(url, timeout=30) r.raise_for_status() except Exception as e: return f"Could not download file for task {task_id}: {e}" content_type = r.headers.get("content-type", "").lower() # Spreadsheets if "spreadsheet" in content_type or "excel" in content_type or url.endswith((".xlsx", ".xls")): try: import io df = pd.read_excel(io.BytesIO(r.content)) return f"Spreadsheet contents:\n{df.to_string()}" except Exception as e: return f"Got a spreadsheet but failed to parse it: {e}" # CSV if "csv" in content_type or url.endswith(".csv"): try: import io df = pd.read_csv(io.BytesIO(r.content)) return f"CSV contents:\n{df.to_string()}" except Exception as e: return f"Got a CSV but failed to parse it: {e}" # Try plain text try: text = r.content.decode("utf-8") return f"File contents:\n{text[:5000]}" except Exception: return (f"The attached file is binary ({content_type}, {len(r.content)} bytes) " f"and cannot be read as text in this version of the agent.") TOOLS = [search_tool, wiki_tool, fetch_task_file] # --- Agent --- class BasicAgent: def __init__(self): api_key = os.getenv("GROQ_API_KEY") if not api_key: raise ValueError("GROQ_API_KEY is not set. Add it in Settings -> Variables and secrets.") llm = ChatGroq(model="llama-3.3-70b-versatile", temperature=0, api_key=api_key) self.graph = create_react_agent(llm, TOOLS) print("LangGraph agent initialized.") def __call__(self, question: str, task_id: str | None = None) -> str: print(f"Agent received question (first 60 chars): {question[:60]}...") user_content = question if task_id: user_content += (f"\n\n[If this question refers to an attached file, " f"call fetch_task_file with task_id='{task_id}'.]") messages = [SystemMessage(content=SYSTEM_PROMPT), HumanMessage(content=user_content)] try: result = self.graph.invoke({"messages": messages}, config={"recursion_limit": 25}) answer = result["messages"][-1].content.strip() except Exception as e: print(f"Agent error: {e}") return f"AGENT ERROR: {e}" # Safety net: strip a stray "FINAL ANSWER" prefix if the model added one if answer.upper().startswith("FINAL ANSWER"): answer = answer.split(":", 1)[-1].strip() print(f"Agent answer: {answer[:100]}") return answer def run_and_submit_all(profile: gr.OAuthProfile | None): """Fetches all questions, runs the agent, submits answers, and shows results.""" space_id = os.getenv("SPACE_ID") if profile: 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 api_url = DEFAULT_API_URL questions_url = f"{api_url}/questions" submit_url = f"{api_url}/submit" # 1. Instantiate Agent try: agent = BasicAgent() except Exception as e: print(f"Error instantiating agent: {e}") return f"Error initializing agent: {e}", None 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: response = requests.get(questions_url, timeout=15) response.raise_for_status() questions_data = 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 Exception as e: print(f"Error fetching questions: {e}") return f"Error fetching questions: {e}", None # 3. Run Agent results_log = [] answers_payload = [] print(f"Running agent on {len(questions_data)} questions...") 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: submitted_answer = agent(question_text, task_id) 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) # 4. Prepare Submission submission_data = {"username": 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: 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.") return final_status, pd.DataFrame(results_log) except requests.exceptions.HTTPError as e: error_detail = f"Server responded with status {e.response.status_code}." try: error_detail += f" Detail: {e.response.json().get('detail', e.response.text)}" except requests.exceptions.JSONDecodeError: error_detail += f" Response: {e.response.text[:500]}" print(f"Submission Failed: {error_detail}") return f"Submission Failed: {error_detail}", pd.DataFrame(results_log) except Exception as e: print(f"An unexpected error occurred during submission: {e}") return f"An unexpected error occurred during submission: {e}", pd.DataFrame(results_log) # --- Gradio Interface --- with gr.Blocks() as demo: 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. --- Note: a full run takes a while (the agent goes through all questions one by one). """ ) gr.LoginButton() 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) 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}") else: print("ℹ️ SPACE_HOST not found (running locally?).") if space_id_startup: print(f"✅ SPACE_ID found: {space_id_startup}") else: print("ℹ️ SPACE_ID not found (running locally?).") print("-" * 74 + "\n") print("Launching Gradio Interface for Basic Agent Evaluation...") demo.launch(debug=True, share=False, ssr_mode=False)