|
|
import requests |
|
|
import json |
|
|
import os |
|
|
from smolagents import tool |
|
|
from typing import Any, Optional |
|
|
from smolagents.tools import Tool |
|
|
|
|
|
class FinalAnswerTool(Tool): |
|
|
name = "final_answer" |
|
|
description = "Provides a final answer to the given problem." |
|
|
inputs = {'answer': {'type': 'any', 'description': 'The final answer to the problem'}} |
|
|
output_type = "any" |
|
|
|
|
|
def forward(self, answer: Any) -> Any: |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
return answer |
|
|
|
|
|
def __init__(self, *args, **kwargs): |
|
|
self.is_initialized = False |
|
|
|
|
|
@tool |
|
|
def fetch_questions(api_url: str) -> dict | None: |
|
|
""" A tool that fetches the questions from the agents-course API and returns the JSON as a dict. |
|
|
Args: |
|
|
api_url: The base URL of the agents-course API. |
|
|
Returns: |
|
|
A dictionary containing the questions if successful, or None if there was an error. |
|
|
""" |
|
|
questions_url = f"{api_url}/questions" |
|
|
|
|
|
try: |
|
|
response = requests.get(questions_url) |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
except requests.RequestException as e: |
|
|
print(f"Error fetching questions: {e}") |
|
|
return None |
|
|
|
|
|
@tool |
|
|
def fetch_file(api_url:str, task_id: str, file_name: str) -> None: |
|
|
""" A tool that fetches a file for a given task ID from the agents-course API. |
|
|
Args: |
|
|
api_url: The base URL of the agents-course API. |
|
|
task_id: The ID of the task for which the file is to be fetched. |
|
|
file_name: The name of the file to be downloaded. |
|
|
Returns: |
|
|
None if the file is downloaded successfully, or an error message if there was an issue. |
|
|
""" |
|
|
file_url = f"{api_url}/files/{task_id}" |
|
|
downsloads_dir = "files" |
|
|
os.makedirs(downsloads_dir, exist_ok=True) |
|
|
|
|
|
file_path = os.path.join(downsloads_dir, file_name) |
|
|
if not os.path.exists(file_path): |
|
|
print(f"Downloading file {file_name} for task {task_id} ...") |
|
|
file_response = requests.get(file_url, timeout=30) |
|
|
if file_response.status_code == 200: |
|
|
with open(file_path, 'wb') as f: |
|
|
f.write(file_response.content) |
|
|
print(f"File {file_name} downloaded successfully.") |
|
|
else: |
|
|
print(f"Failed to download file {file_name} for task {task_id}. Status code: {file_response.status_code}") |
|
|
return None |
|
|
|