File size: 7,021 Bytes
acfd37c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
142599f
 
acfd37c
 
fbe1f2c
acfd37c
142599f
 
fbe1f2c
acfd37c
 
 
 
 
 
142599f
 
 
acfd37c
 
 
 
 
 
142599f
 
 
acfd37c
 
 
 
142599f
acfd37c
fbe1f2c
acfd37c
 
 
 
142599f
acfd37c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
142599f
acfd37c
 
 
 
 
 
 
 
 
 
fbe1f2c
acfd37c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
import os
from smolagents import CodeAgent, LiteLLMModel, load_tool, ToolCollection, HfApiModel, InferenceClientModel, TransformersModel, OpenAIServerModel
from smolagents import ToolCallingAgent, PythonInterpreterTool, tool, WikipediaSearchTool
from smolagents import DuckDuckGoSearchTool, FinalAnswerTool, VisitWebpageTool, SpeechToTextTool
from mcp import StdioServerParameters
from huggingface_hub import HfApi, login
from dotenv import load_dotenv
from typing import Optional
import requests
import re
import string
import random
import textwrap
import nltk
import spacy

DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"


@tool
def download_file(task_id: str) -> str:
    """
    Returns the file path of the downloaded file.

    Args:
        task_id: the ID of the task to download the file for.
    """
    data = requests.get(f"{DEFAULT_API_URL}/files/{task_id}")
    if data.status_code == 200:
        file_path = f"/tmp/{task_id}"
        with open(file_path, "wb") as file:
            file.write(data.content)
        return file_path
    else:
        raise Exception(f"Failed to download file: {data.status_code}")

@tool  
def get_file_content_as_text(task_id: str) -> str:
    """
    Returns the content of the file as text.

    Args:
        task_id: the ID of the task to get the file content for.
    """
    data = requests.get(f"{DEFAULT_API_URL}/files/{task_id}")
    if data.status_code == 200:
        return data.text
    else:
        raise Exception(f"Failed to get file content: {data.status_code}")



def load_hf_model(modelName: str):
    """
    Loads a model from the hugging face hub
    :param modelName: Name of the model
    :return: model
    """
    load_dotenv()

    # for local usage, we might use a hf token to log in
    # hf_token = os.getenv("hugging_face")
    # login(token=hf_token)  # Login at hugging face
    model = HfApiModel(model_id=modelName)
    return model


def load_ollama_model(modelName: str):
    """
    Loads the requested model in ollama
    :param modelName: Name of the model
    :return: model (via OpenAI compatible API)
    """
    model = OpenAIServerModel(model_id=modelName, api_base="http://localhost:11434/v1")
    return model

def load_lmStudio_model(modelName: str):
    """
    Loads the requested model into lm studio
    :param modelName: Name of the model
    :return: model, accessible through the OpenAI compatible API
    """
    model = OpenAIServerModel(model_id=modelName, api_base="http://localhost:1234/v1")
    return model

def load_gemini_model(model_name: str):
    """
    Loads the gemini model
    :return: model
    """
    try:
        print(f"Gemini API Key: {os.getenv('GEMINI_API_KEY')}")
        model = LiteLLMModel(model_id=f"gemini/{model_name}",
                         api_key=os.getenv("GEMINI_API_KEY"))
        return model
    except Exception as e:
        print("Error loading Gemini model:", e)
        return None



def get_agent(model_name:str, model_type:str) -> Optional[CodeAgent]:
    
    match model_type:
        case "hugging face":
            model = load_hf_model(model_name)
        case "Ollama":
            model = load_ollama_model(model_name)
        case "Gemini":
            model = load_gemini_model(model_name)
        case "LMStudio":
            model = load_lmStudio_model(model_name)
        case _:
            print("Model type not supported.")
            return None
    
    # Tools laden
    web_search_tool = DuckDuckGoSearchTool()
    final_answer_tool = FinalAnswerTool()
    visit_webpage_tool = VisitWebpageTool()


    variation_agent = CodeAgent(
        model=model,
        tools=[PythonInterpreterTool()],
        name="variation_agent",
        description="Get the user question and checks if the given question makes sense at all, if not, we try to modify the text like reverse. Provide the content / the questin as the 'task' argument." \
        "The agent can write professional python code, focused on modifiying texts." \
        "It has access to the following libraries: re, string, random, textwrap, nltk and spacy." \
        "The goal is to find out, if a user question is a trick, and we might modify the content.",
        additional_authorized_imports=[
            "re",
            "string",
            "random",
            "textwrap",
            "nltk",
            "spacy"
        ]
        )
    variation_agent.system_prompt = "You are a text variation agent. You can write professional python code, focused on modifiying texts." \
    "You can use the following libraries: re, string, random, textwrap, nltk and spacy." \
    "Your goal is to find out, if a user question is a trick, and we might modify the content."

    code_agent = CodeAgent(
        name="code_agent",
        description="Can generate code an run it. It provides the possibility to download additional files if needed.",
        model=model,
        tools=[download_file, PythonInterpreterTool(), get_file_content_as_text],
        additional_authorized_imports=[
            "geopandas",
            "plotly",
            "shapely",
            "json",
            "pandas",
            "numpy",
        ],
        verbosity_level=2,
        #final_answer_checks=[FinalAnswerTool()],
        max_steps=5,
    )

    final_answer_tool = FinalAnswerTool()
    final_answer_tool.description = "You are a general AI assistant. I will ask you a question. Report your thoughts, and finish your answer with the following template: FINAL ANSWER: [YOUR FINAL ANSWER]. YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings. If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string."

    tool_agent = CodeAgent(
        model=model,
        tools=[web_search_tool, visit_webpage_tool, WikipediaSearchTool(), final_answer_tool],
        verbosity_level=2,
        max_steps=15,
        managed_agents=[code_agent, variation_agent],
        planning_interval=5,
    )

    return tool_agent
#    return tool_agent

    manager_agent = CodeAgent(
        #model=HfApiModel("deepseek-ai/DeepSeek-R1", provider="together", max_tokens=8096),
        model=model,
        tools=[web_search_tool, visit_webpage_tool],
#        managed_agents=[mcp_tool_agent],
        additional_authorized_imports=[
            "geopandas",
            "plotly",
            "shapely",
            "json",
            "pandas",
            "numpy",
        ],
        planning_interval=5,
        verbosity_level=2,
        #final_answer_checks=[FinalAnswerTool()],
        max_steps=15
    )
    
    return manager_agent