Spaces:
Sleeping
Sleeping
File size: 1,924 Bytes
65822ef |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
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
]
|