Spaces:
Sleeping
Sleeping
| import base64 | |
| import requests | |
| try: | |
| from smolagents import Tool | |
| except ImportError: | |
| Tool = object | |
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" | |
| HF_DATASET_BASE = "https://huggingface.co/datasets/gaia-benchmark/GAIA/resolve/main/2023/test" | |
| # Tool to download and read file attachments (text) from the scoring API | |
| # Plain class (no Tool inheritance) — called directly, not via CodeAgent | |
| class FileReaderTool: | |
| name = "file_reader" | |
| def __init__(self, api_url: str = DEFAULT_API_URL): | |
| self.api_url = api_url | |
| print("FileReaderTool initialized.") | |
| def __call__(self, task_id: str, file_name: str = "") -> str: | |
| # Try HF dataset URL first (actual file location), then scoring API fallback | |
| urls_to_try = [] | |
| if file_name: | |
| urls_to_try.append(f"{HF_DATASET_BASE}/{file_name}") | |
| urls_to_try.append(f"{self.api_url}/files/{task_id}") | |
| for url in urls_to_try: | |
| print(f"FileReaderTool trying: {url}") | |
| try: | |
| resp = requests.get(url, timeout=30) | |
| if resp.status_code == 200: | |
| content = resp.text | |
| print(f"FileReaderTool downloaded {len(content)} chars.") | |
| return content | |
| except requests.exceptions.RequestException: | |
| continue | |
| return f"Failed to download file for task {task_id}" | |
| # Tool to download image attachments and return them as base64 | |
| # Plain class (no Tool inheritance) — called directly, not via CodeAgent | |
| class ImageReaderTool: | |
| name = "image_reader" | |
| def __init__(self, api_url: str = DEFAULT_API_URL): | |
| self.api_url = api_url | |
| print("ImageReaderTool initialized.") | |
| def __call__(self, task_id: str, file_name: str = "") -> str: | |
| # Try HF dataset URL first (actual file location), then scoring API fallback | |
| urls_to_try = [] | |
| if file_name: | |
| urls_to_try.append(f"{HF_DATASET_BASE}/{file_name}") | |
| urls_to_try.append(f"{self.api_url}/files/{task_id}") | |
| for url in urls_to_try: | |
| print(f"ImageReaderTool trying: {url}") | |
| try: | |
| resp = requests.get(url, timeout=30) | |
| if resp.status_code == 200: | |
| content_type = resp.headers.get("Content-Type", "image/png") | |
| image_b64 = base64.b64encode(resp.content).decode("utf-8") | |
| print(f"ImageReaderTool downloaded image ({len(resp.content)} bytes, {content_type}).") | |
| return f"data:{content_type};base64,{image_b64}" | |
| except requests.exceptions.RequestException: | |
| continue | |
| return f"Failed to download image for task {task_id}" | |
| # Web search tool — uses ddgs directly to avoid smolagents DuckDuckGoSearchTool package check | |
| class WebSearchTool: | |
| name = "web_search" | |
| def __init__(self): | |
| print("WebSearchTool initialized.") | |
| def __call__(self, query: str) -> str: | |
| print(f"WebSearchTool received query (first 50 chars): {query[:50]}...") | |
| try: | |
| from ddgs import DDGS | |
| with DDGS() as ddgs: | |
| results = list(ddgs.text(query, max_results=5)) | |
| if not results: | |
| return "No results found." | |
| output = "\n\n".join( | |
| f"{r.get('title', '')}\n{r.get('href', '')}\n{r.get('body', '')}" | |
| for r in results | |
| ) | |
| except Exception as e: | |
| output = f"Search error: {e}" | |
| print(f"WebSearchTool returning result (first 100 chars): {output[:100]}...") | |
| return output | |