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,
) |