File size: 2,225 Bytes
65822ef
 
c42da51
65822ef
 
c42da51
65822ef
 
 
 
 
 
 
 
 
 
 
 
 
c42da51
65822ef
 
 
 
 
 
 
 
 
 
 
c42da51
 
 
65822ef
 
c42da51
65822ef
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c42da51
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
65822ef
 
 
 
 
c42da51
 
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import os

import whisper
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.reasoning import ReasoningTools
from utils import fetch_file_content
from agno.tools.wikipedia import WikipediaTools

BASE_STORAGE_ROOT = os.getenv("AGENT_STORAGE_ROOT", os.path.join(os.getcwd(), "agent_storage"))


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: Path to the downloaded file.
    """
    task_dir = os.path.join(BASE_STORAGE_ROOT, task_id)
    os.makedirs(task_dir, exist_ok=True)

    filename = task_id
    file_path = os.path.join(task_dir, filename)

    if os.path.exists(file_path):
        print("[INFO] Using cached file:", file_path)
        return file_path

    # Use the utility function to fetch content
    result = fetch_file_content(task_id)
    content = result["content"]

    with open(file_path, "wb") as f:
        f.write(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()


def convert_audio_to_text(task_id: str) -> str:
    """
    Use this to download an audio and convert it to text

     Args:
        task_id (str): Identifier that points to the remote file.

    Returns:
        str: the transcript of the audio in text
    """

    result = fetch_file_content(task_id, temp=True)
    model = whisper.load_model("turbo")
    result_whisper = model.transcribe(audio=result["path"])
    print("[convert_audio_to_text]", result_whisper["text"])
    return result_whisper["text"]


tools = [
    ReasoningTools(think=True, add_few_shot=True),
    DuckDuckGoTools(fixed_max_results=5),
    WikipediaTools(),
    get_file_from_task_id,
    read_file_from_task_id,
    convert_audio_to_text
]