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 | |
] | |