import os import re import json import time import requests import pandas as pd import gradio as gr from pathlib import Path # --- Constants --- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" TASK_FILES_DIR = Path("./task_files") TASK_FILES_DIR.mkdir(exist_ok=True) AUTHORIZED_IMPORTS = [ "os", "re", "json", "csv", "math", "datetime", "collections", "itertools", "pathlib", "pandas", "numpy", "statistics", "unicodedata", ] GAIA_RULES = """ ## GAIA BENCHMARK — STRICT ANSWER RULES You MUST follow these rules EXACTLY. Violations = WRONG answer. 1. Return ONLY the final answer. No "Final Answer:", no reasoning, no explanation. 2. Numbers: digits only ("42", not "forty-two"), unless words are requested. 3. Yes/No: exactly "Yes" or "No". 4. Lists: comma-separated ("a, b, c"), in the order the question implies. 5. Match plural/singular exactly as asked. 6. Names: exact spelling, full names, proper capitalization. 7. Cities: full name ("Los Angeles", not "LA"). 8. No units unless asked (no "$", "%", "km"). 9. No trailing period. No markdown, no quotes, no backticks. 10. If a file is provided: read it with Python first. 11. If unsure: give your single best guess (never leave blank). """ def clean_answer_for_gaia(raw: str) -> str: if not raw: return "" s = raw.strip() lower = s.lower() if lower.startswith("error:"): return s prefixes = [ "final answer:", "final answer -", "the answer is", "answer:", "answer -", "ответ:", "финальный ответ:", ] for p in prefixes: if lower.startswith(p): s = s[len(p):].strip() lower = s.lower() s = s.strip("`").strip() if s.startswith("```") and s.endswith("```"): s = s[3:-3].strip() if len(s) >= 2 and s[0] == s[-1] and s[0] in ('"', "'"): s = s[1:-1].strip() if s.endswith("."): s = s[:-1].rstrip() s = re.sub(r"\s+", " ", s) for marker in ["\n\n", "\nExplanation:", "\nNote:"]: if marker in s: s = s.split(marker, 1)[0].strip() return s def download_task_file(task_id: str): try: r = requests.get(f"{DEFAULT_API_URL}/files/{task_id}", timeout=30) if r.status_code != 200 or not r.content: return None cd = r.headers.get("content-disposition", "") filename = f"{task_id}.bin" m = re.search(r'filename="?([^";]+)"?', cd) if m: filename = m.group(1) path = TASK_FILES_DIR / filename path.write_bytes(r.content) return path except Exception as e: print(f"[download_task_file] {task_id}: {e}") return None # --- Real Agent --- class BasicAgent: def __init__(self): print("BasicAgent initializing...") from smolagents import ( CodeAgent, OpenAIServerModel, WebSearchTool, VisitWebpageTool, ) # Берем ключ из Secrets (переменных окружения) Space deepseek_key = os.environ.get("DEEPSEEK_API_KEY") if not deepseek_key: raise ValueError("DEEPSEEK_API_KEY is not set in Space Secrets! Please add it in Settings -> Variables and Secrets.") # ✅ ИНИЦИАЛИЗАЦИЯ DEEPSEEK model = OpenAIServerModel( model_id="deepseek-chat", api_base="https://api.deepseek.com/v1", # Используем официальный эндпоинт api_key=deepseek_key, max_tokens=2048, temperature=0.0, ) self._agent = CodeAgent( model=model, tools=[WebSearchTool(), VisitWebpageTool()], max_steps=10, verbosity_level=2, additional_authorized_imports=AUTHORIZED_IMPORTS, ) try: base = self._agent.prompt_templates.get("system_prompt", "") self._agent.prompt_templates["system_prompt"] = base + GAIA_RULES except Exception as e: print(f"[prompt_templates] {e}") print("BasicAgent ready.") def __call__(self, question: str) -> str: print(f"Agent received question (first 80 chars): {question[:80]}...") try: raw = self._agent.run(question) except Exception as e: print(f"[agent.run] FATAL ERROR: {e}") raw = f"ERROR: {str(e)}" answer = clean_answer_for_gaia(str(raw)) print(f"Agent answer: {answer!r}") return answer def run_and_submit_all(profile: gr.OAuthProfile | None): space_id = os.getenv("SPACE_ID") if profile: username = f"{profile.username}" else: 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" try: agent = BasicAgent() except Exception as e: return f"Error initializing agent: {e}", None agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" try: response = requests.get(questions_url, timeout=30) response.raise_for_status() questions_data = response.json() if not questions_data: return "Fetched questions list is empty.", None except Exception as e: return f"Error fetching questions: {e}", None results_log = [] answers_payload = [] for i, item in enumerate(questions_data, 1): task_id = item.get("task_id") question_text = item.get("question") if not task_id or question_text is None: continue file_path = download_task_file(task_id) prompt = question_text if file_path: prompt = ( f"{question_text}\n\n" f"[FILE] The associated file is saved at: `{file_path}`. " f"Read it with Python before answering." ) try: submitted_answer = agent(prompt) except Exception as e: submitted_answer = f"ERROR: {e}" answers_payload.append( {"task_id": task_id, "submitted_answer": submitted_answer} ) results_log.append({ "Task ID": task_id, "Question": question_text[:100], "Submitted Answer": submitted_answer, }) print(f"[{i}/{len(questions_data)}] {task_id} -> {submitted_answer!r}") if not answers_payload: return "Agent did not produce any answers.", pd.DataFrame(results_log) submission_data = { "username": username.strip(), "agent_code": agent_code, "answers": answers_payload, } try: response = requests.post(submit_url, json=submission_data, timeout=120) 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.')}" ) return final_status, pd.DataFrame(results_log) except Exception as e: return f"Submission Failed: {e}", pd.DataFrame(results_log) with gr.Blocks() as demo: gr.Markdown("# GAIA Agent — Unit 4 Final Project") gr.Markdown("1. **Log in**\n2. Click **Run Evaluation**\n3. Check the table below.") gr.LoginButton() run_button = gr.Button("Run Evaluation & Submit All Answers", variant="primary") status_output = gr.Textbox(label="Status / Result", lines=8, interactive=False) results_table = gr.DataFrame(label="Questions and Answers", wrap=True) run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table]) if __name__ == "__main__": demo.launch(debug=True, share=False)