r00ta1m/Final_Assignment_Template
0
1import os2import gradio as gr3import requests4import pandas as pd5import threading6import re7import base648from bs4 import BeautifulSoup9from smolagents import CodeAgent, DuckDuckGoSearchTool, tool, InferenceClientModel, PythonInterpreterTool10 11# --- Constants ---12DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"13 14SYSTEM_PROMPT = """You are a precise AI assistant solving GAIA benchmark tasks.15STRICT ANSWER FORMAT RULES:16- Return ONLY the bare answer: a number, name, date, word, or short phrase.17- NO explanations, NO prefixes like "The answer is", NO punctuation unless it is part of the answer.18- If asked "how many" → just the number e.g. 319- If asked for a name → just the name e.g. Marie Curie20- If asked for a date → just the date e.g. 1969-07-2021- If asked for a comma separated list → alphabetize unless told otherwise e.g. apple, banana, cherry22- If a number has units → only include units if the question explicitly asks for them23TOOL STRATEGY:24- If the question mentions an attached file (.mp3, .xlsx, .csv, .py, .pdf etc), ALWAYS call get_task_file(task_id) first.25- If a URL is in the question, call visit_webpage on it immediately.26- For factual/historical questions, use wikipedia_search first, then web_search to confirm.27- Use Python (python_interpreter) for ALL math, counting, sorting, table operations.28- Never guess. Always verify with tools before answering.29- If first search fails, try different keywords.30SPECIAL CASES:31- For reversed text questions: reverse the string in Python and read it.32- For math/logic table questions: use Python to check all pairs systematically.33- For botany questions: remember botanical fruits (tomato, pepper, zucchini, corn, green beans, peanuts, acorns) are NOT vegetables.34- For Excel/CSV files: use Python with the csv module to parse and calculate.35- For Wikipedia questions: use the wikipedia_search tool with the exact article name.36"""37 38# -----------------------39# Tools40# -----------------------41 42@tool43def visit_webpage(url: str) -> str:44 """Visit a webpage and return its readable text content. Use whenever a URL appears in the question.45 Args:46 url: The full URL to visit.47 """48 try:49 headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}50 r = requests.get(url, headers=headers, timeout=15)51 soup = BeautifulSoup(r.text, "html.parser")52 for tag in soup(["script", "style", "nav", "footer", "header"]):53 tag.decompose()54 text = soup.get_text(separator=" ", strip=True)55 return text[:20000]56 except Exception as e:57 return f"Failed to visit {url}: {str(e)}"58 59 60@tool61def get_task_file(task_id: str) -> str:62 """Download the file attached to a GAIA task. ALWAYS use this when the question mentions a file (.mp3, .xlsx, .csv, .py, .pdf, image, etc).63 Returns file content as text, or base64 for binary files.64 Args:65 task_id: The GAIA task ID.66 """67 try:68 url = f"{DEFAULT_API_URL}/files/{task_id}"69 headers = {"User-Agent": "Mozilla/5.0"}70 r = requests.get(url, headers=headers, timeout=20)71 if r.status_code != 200:72 return f"No file found for task {task_id} (status {r.status_code})"73 74 content_type = r.headers.get("content-type", "")75 76 # Text-based files77 if any(t in content_type for t in ["text", "json", "csv", "python", "javascript"]):78 return r.text[:25000]79 80 # Excel files — return base64 so Python can parse with openpyxl81 if "spreadsheet" in content_type or "excel" in content_type or task_id.endswith(".xlsx"):82 b64 = base64.b64encode(r.content).decode()83 return f"EXCEL_BASE64:{b64[:50000]}"84 85 # PDF86 if "pdf" in content_type:87 return f"PDF file ({len(r.content)} bytes). Try downloading and parsing with PyPDF2 if needed. Raw text attempt:\n{r.text[:5000]}"88 89 # Audio — return base64 hint90 if "audio" in content_type or "mp3" in content_type:91 return f"AUDIO file ({len(r.content)} bytes, type: {content_type}). This is an audio file that cannot be transcribed directly. Try searching for the answer using context from the question."92 93 # Default: try text94 try:95 return r.text[:25000]96 except Exception:97 b64 = base64.b64encode(r.content).decode()98 return f"BINARY_BASE64:{b64[:50000]}"99 100 except Exception as e:101 return f"Failed to get task file: {str(e)}"102 103 104@tool105def wikipedia_search(query: str) -> str:106 """Search Wikipedia and return the full article text. Best for facts, biographies, history, science.107 For discography questions, search 'Artist Name discography'.108 Args:109 query: Wikipedia search query e.g. 'Mercedes Sosa discography'110 """111 try:112 # Search for article113 search_params = {114 "action": "query",115 "list": "search",116 "srsearch": query,117 "format": "json",118 "srlimit": 5119 }120 r = requests.get("https://en.wikipedia.org/w/api.php", params=search_params, timeout=10)121 results = r.json().get("query", {}).get("search", [])122 if not results:123 return "No Wikipedia results found."124 125 title = results[0]["title"]126 127 # Get full article text (not just intro)128 content_params = {129 "action": "query",130 "titles": title,131 "prop": "extracts",132 "explaintext": True,133 "format": "json"134 }135 content_r = requests.get("https://en.wikipedia.org/w/api.php", params=content_params, timeout=10)136 pages = content_r.json().get("query", {}).get("pages", {})137 page = next(iter(pages.values()))138 extract = page.get("extract", "No content available.")139 return f"Wikipedia article: {title}\n\n{extract[:10000]}"140 except Exception as e:141 return f"Wikipedia search failed: {str(e)}"142 143 144@tool145def download_file_from_url(url: str) -> str:146 """Download any file from a URL and return its text content.147 Args:148 url: The full URL of the file.149 """150 try:151 headers = {"User-Agent": "Mozilla/5.0"}152 r = requests.get(url, headers=headers, timeout=15)153 return r.text[:25000]154 except Exception as e:155 return f"Download failed: {str(e)}"156 157 158@tool159def cve_search(cve_id: str) -> dict:160 """Retrieve CVE vulnerability details from NVD.161 Args:162 cve_id: CVE identifier e.g. CVE-2021-44228163 """164 headers = {"User-Agent": "Mozilla/5.0"}165 try:166 r = requests.get(167 f"https://services.nvd.nist.gov/rest/json/cves/2.0?cveId={cve_id}",168 headers=headers, timeout=10169 )170 data = r.json()171 if "vulnerabilities" not in data or not data["vulnerabilities"]:172 return {"error": f"{cve_id} not found"}173 vuln = data["vulnerabilities"][0]["cve"]174 desc = vuln["descriptions"][0]["value"] if vuln.get("descriptions") else "N/A"175 metrics = vuln.get("metrics", {})176 for key in ["cvssMetricV31", "cvssMetricV30", "cvssMetricV2"]:177 if key in metrics:178 cvss_data = metrics[key][0]["cvssData"]179 break180 else:181 cvss_data = {}182 return {183 "cve_id": cve_id,184 "description": desc[:800],185 "cvss_score": cvss_data.get("baseScore", "N/A"),186 "severity": cvss_data.get("baseSeverity", "N/A"),187 "references": [r["url"] for r in vuln.get("references", [])][:3]188 }189 except Exception as e:190 return {"error": str(e)}191 192 193# -----------------------194# Agent Builder195# -----------------------196 197def build_agent():198 model = InferenceClientModel(199 model_id="Qwen/Qwen2.5-72B-Instruct",200 max_tokens=2048,201 temperature=0.1,202 token=os.getenv("HF_TOKEN")203 )204 return CodeAgent(205 model=model,206 tools=[207 DuckDuckGoSearchTool(),208 visit_webpage,209 get_task_file,210 wikipedia_search,211 download_file_from_url,212 cve_search,213 PythonInterpreterTool()214 ],215 max_steps=10,216 verbosity_level=1,217 additional_authorized_imports=[218 "re", "json", "datetime", "math", "collections",219 "csv", "io", "string", "itertools", "functools",220 "base64", "openpyxl", "pandas"221 ]222 )223 224 225# -----------------------226# Run agent with timeout227# -----------------------228 229def run_agent_with_timeout(agent, question, timeout=180):230 result = [None]231 error = [None]232 233 def target():234 try:235 result[0] = str(agent.run(question))236 except Exception as e:237 error[0] = str(e)238 239 t = threading.Thread(target=target)240 t.start()241 t.join(timeout)242 243 if t.is_alive():244 return "TIMEOUT"245 if error[0]:246 return f"AGENT ERROR: {error[0]}"247 return result[0]248 249 250# -----------------------251# Submission Function252# -----------------------253 254def run_and_submit_all(profile: gr.OAuthProfile | None):255 space_id = os.getenv("SPACE_ID")256 257 if profile:258 username = profile.username259 print(f"User logged in: {username}")260 else:261 return "Please login to Hugging Face with the button.", None262 263 api_url = DEFAULT_API_URL264 agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"265 print(f"Agent code link: {agent_code}")266 267 # Fetch questions268 try:269 response = requests.get(f"{api_url}/questions", timeout=15)270 response.raise_for_status()271 questions_data = response.json()272 print(f"Fetched {len(questions_data)} questions.")273 except Exception as e:274 return f"Error fetching questions: {e}", None275 276 # Build agent once277 try:278 agent = build_agent()279 except Exception as e:280 return f"Error initializing agent: {e}", None281 282 # Run agent on each question283 results_log = []284 answers_payload = []285 286 for item in questions_data:287 task_id = item.get("task_id")288 question_text = item.get("question")289 if not task_id or question_text is None:290 continue291 292 print(f"\n📋 Task {task_id}: {question_text[:100]}...")293 294 augmented_question = (295 f"{SYSTEM_PROMPT}\n\n"296 f"Task ID: {task_id}\n"297 f"Question: {question_text}\n\n"298 f"If this question mentions any file or attachment, call get_task_file('{task_id}') first.\n"299 f"Return ONLY the bare final answer with no explanation."300 )301 302 submitted_answer = run_agent_with_timeout(agent, augmented_question, timeout=180)303 print(f"✅ Answer: {submitted_answer[:100]}")304 305 answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})306 results_log.append({307 "Task ID": task_id,308 "Question": question_text,309 "Submitted Answer": submitted_answer310 })311 312 if not answers_payload:313 return "Agent produced no answers.", pd.DataFrame(results_log)314 315 # Submit316 submission_data = {317 "username": username.strip(),318 "agent_code": agent_code,319 "answers": answers_payload320 }321 322 try:323 response = requests.post(f"{api_url}/submit", json=submission_data, timeout=60)324 response.raise_for_status()325 result_data = response.json()326 final_status = (327 f"Submission Successful!\n"328 f"User: {result_data.get('username')}\n"329 f"Overall Score: {result_data.get('score', 'N/A')}% "330 f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"331 f"Message: {result_data.get('message', 'No message received.')}"332 )333 return final_status, pd.DataFrame(results_log)334 except Exception as e:335 return f"Submission Failed: {e}", pd.DataFrame(results_log)336 337 338# -----------------------339# Gradio UI340# -----------------------341 342with gr.Blocks() as demo:343 gr.Markdown("# 🛡️ CVE Intelligence Agent — GAIA Evaluation")344 gr.Markdown(345 """346 **Instructions:**347 1. Log in with your Hugging Face account below.348 2. Click **Run Evaluation & Submit All Answers**.349 3. Wait for all 20 questions (~15-25 min).350 4. Your score will appear when done.351 ---352 ⚠️ Do not close this tab while running. Each question has a 3 min timeout.353 """354 )355 356 gr.LoginButton()357 358 run_button = gr.Button("🚀 Run Evaluation & Submit All Answers", variant="primary")359 status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)360 results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)361 362 run_button.click(363 fn=run_and_submit_all,364 outputs=[status_output, results_table]365 )366 367if __name__ == "__main__":368 space_id = os.getenv("SPACE_ID")369 if space_id:370 print(f"✅ SPACE_ID: {space_id}")371 print(f" Repo URL: https://huggingface.co/spaces/{space_id}/tree/main")372 demo.launch(debug=True, share=False)