Spaces:
Sleeping
Sleeping
| import os | |
| import httpx | |
| from agno.tools.duckduckgo import DuckDuckGoTools | |
| from agno.tools.reasoning import ReasoningTools | |
| from agno.tools.wikipedia import WikipediaTools | |
| BASE_STORAGE_ROOT = os.getenv("AGENT_STORAGE_ROOT", os.path.join(os.getcwd(), "agent_storage")) | |
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" | |
| def get_file_from_task_id(task_id: str) -> str: | |
| """ | |
| Use this tool to **download** the file linked to a given `task_id`. | |
| Args: | |
| task_id (str): Identifier that points to the remote file. | |
| Returns: | |
| str: task_id to be used by other tools to read the file | |
| """ | |
| # ensure storage directory exists | |
| task_dir = os.path.join(BASE_STORAGE_ROOT, task_id) | |
| os.makedirs(task_dir, exist_ok=True) | |
| # filename derived from task_id | |
| filename = task_id | |
| file_path = os.path.join(task_dir, filename) | |
| # if file already exists, return | |
| if os.path.exists(file_path): | |
| print("[INFO] Using cached file:", file_path) | |
| return file_path | |
| # fetch content from remote | |
| response = httpx.get(f"{DEFAULT_API_URL}/files/{task_id}", timeout=15, follow_redirects=True) | |
| response.raise_for_status() | |
| # write content to file | |
| with open(file_path, "wb") as f: | |
| f.write(response.content) | |
| return file_path | |
| def read_file_from_task_id(task_id: str) -> str: | |
| """ | |
| Args: | |
| task_id (str): Identifier that points to the remote file. | |
| Returns: | |
| str: Content of the downloaded (or cached) file. | |
| """ | |
| # expected local path | |
| file_path = os.path.join(BASE_STORAGE_ROOT, task_id, task_id) | |
| with open(file_path, "r", encoding="utf-8") as f: | |
| print("[INFO] Reading file:", file_path) | |
| return f.read() | |
| tools = [ | |
| ReasoningTools(think=True, add_few_shot=True), | |
| DuckDuckGoTools(fixed_max_results=5), | |
| WikipediaTools(), | |
| get_file_from_task_id, | |
| read_file_from_task_id | |
| ] | |