File size: 6,220 Bytes
90079bd
6de65a0
 
 
 
 
 
480fc0c
 
6de65a0
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7a0516a
6de65a0
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
ca345fd
6de65a0
 
 
7a0516a
6de65a0
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
cb36500
 
 
6de65a0
 
 
fc590bd
6de65a0
 
 
 
cb36500
 
 
6de65a0
cb36500
 
6de65a0
 
 
90079bd
cb36500
 
90079bd
6de65a0
 
8d1968e
6de65a0
 
 
 
 
 
 
 
 
 
 
 
 
cb36500
 
 
6de65a0
 
 
 
 
 
 
 
 
 
 
cb36500
fc590bd
cb36500
6de65a0
fc590bd
 
6de65a0
 
 
 
 
 
fc590bd
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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
from smolagents import CodeAgent, LiteLLMModel, GoogleSearchTool, WikipediaSearchTool, Tool, VisitWebpageTool, HfApiModel
import os
import json
import requests
import pandas as pd
from huggingface_hub import InferenceClient

class GetFileTool(Tool):
    name = "get_file_tool"
    description = "This tool allows to download the file attached to the question"
    output_type = "string"
    inputs = {
        "task_id": {
            "type": "string",
            "description": "The task id of the question",
        },
        "file_name": {
            "type": "string",
            "description": "The name of the file to download"
        }
    }

    def forward(self, task_id: str, file_name: str) -> str:
        # Download the file from the task id
        file = f"https://agents-course-unit4-scoring.hf.space/files/{task_id}"
        # Save the file with the file name
        with open(file_name, "wb") as f:
            f.write(requests.get(file).content)

        # Return the file name  
        return os.path.abspath(file_name)
    
class LoadXlsxFileTool(Tool):
    name = "load_xlsx_file_tool"
    description = """This tool loads xlsx file into pandas and returns it"""
    inputs = {
        "file_path": {"type": "string", "description": "File path"}
    }
    output_type = "object"

    def forward(self, file_path: str) -> object:
        return pd.read_excel(file_path)

class LoadTextFileTool(Tool):
    name = "load_text_file_tool"
    description = """This tool loads any text file"""
    inputs = {
        "file_path": {"type": "string", "description": "File path"}
    }
    output_type = "string"

    def forward(self, file_path: str) -> object:
        with open(file_path, 'r', encoding='utf-8') as file:
            return file.read() 

class AudioToTextTool(Tool):
    name = "audio_to_text_tool"
    description = """This tool transcribes audio files into text"""
    inputs = {
        "file_path": {"type": "string", "description": "File path"}
    }
    output_type = "string"

    def forward(self, file_path: str) -> str:
        try:
            # Check if file exists
            if not os.path.exists(file_path):
                return f"Error: File {file_path} does not exist"

            # Read the audio file as raw bytes
            with open(file_path, "rb") as f:
                audio_data = f.read()

            # Set up the API URL and headers
            api_url = "https://router.huggingface.co/hf-inference/models/openai/whisper-large-v3-turbo"
            headers = {
                "Authorization": f"Bearer {os.getenv('HF_TOKEN')}",
                "Content-Type": "audio/mpeg"  # Assuming MP3 format
            }

            # Make the API request
            response = requests.post(api_url, headers=headers, data=audio_data)
            response.raise_for_status()  # Raise an exception for bad status codes
            
            # Parse and return the response
            output = response.json()
            return output["text"]

        except Exception as e:
            return f"Error transcribing audio: {str(e)}"
    
class ImageAnalysisTool(Tool):
    name = "image_analysis_tool"
    description = """This tool analyzes images and returns the text and the information in the image"""
    inputs = {
        "task_id": {"type": "string", "description": "The task id of the question"},
    }
    output_type = "string"

    def forward(self, task_id: str) -> str:
        client = InferenceClient(
            provider="nebius",
            api_key=os.getenv("HF_TOKEN"),
        )
            
        completion = client.chat.completions.create(
            model="google/gemma-3-27b-it",
            messages=[
                {
                    "role": "user",
                    "content": [
                    {
                        "type": "text",
                        "text": "Describe this image in markdown format"
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"https://agents-course-unit4-scoring.hf.space/files/{task_id}"
                        }
                    }
                ]
            }
        ],
        )

        return completion.choices[0].message.content


def final_answer_formatting(answer, question):
    model = LiteLLMModel(
        model_id="gemini/gemini-2.0-flash",
        api_key=os.getenv("GOOGLE_API_KEY"),
    )

    prompt = f"""
    You are an AI assistant specialized in the GAIA benchmark. For the question provided, generate the answer in the exact format requested by the question. Do not include any other text or creative additions.
    Question: {question}
    Answer: {answer}
    """

    messages = [
    {"role": "user", "content": [{"type": "text", "text": prompt}]}
    ]

    output = model(messages).content
    return output


web_agent = CodeAgent(
    model=LiteLLMModel(
        model_id="gemini/gemini-2.0-flash",
        api_key=os.getenv("GOOGLE_API_KEY"),
    ),
    tools=[
        WikipediaSearchTool(),
        GoogleSearchTool(provider="serper"), 
        VisitWebpageTool()
    ],
    add_base_tools=False,
    additional_authorized_imports=[
        "os", "requests", "inspect", "pandas", 
        "datetime", "re", "bs4", "markdownify"
    ],
    max_steps=10,
    name="web_agent",
    description="This agent is used to search the web for information"
)

audio_agent = CodeAgent(
    model=LiteLLMModel(
        model_id="gemini/gemini-2.0-flash",
        api_key=os.getenv("GOOGLE_API_KEY"),
    ),
    tools=[AudioToTextTool()],
    add_base_tools=False,
    max_steps=10,
    name="audio_agent",
    description="This agent is used to analyze and transcribe audio files"
)


manager_agent = CodeAgent(
    name="manager_agent",
    model=LiteLLMModel(
        model_id="gemini/gemini-2.5-flash-preview-04-17",
        api_key=os.getenv("GOOGLE_API_KEY"),
    ),
    tools=[GetFileTool(), LoadXlsxFileTool(), LoadTextFileTool(), ImageAnalysisTool()],
    managed_agents=[web_agent, audio_agent],
    additional_authorized_imports=[
        "pandas"
    ],
    planning_interval=5,
    verbosity_level=1,
    max_steps=10,
)