File size: 7,606 Bytes
b156b8a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
from dataclasses import dataclass
import os
from langchain.agents import Tool
from langchain_google_genai import ChatGoogleGenerativeAI

from dataclass.dataclass_agent_params import AgentParamsDataClass
from tools.tools_audio_transcriptor import WhisperAudioTranscription
from tools.tools_file_reader import HuggingFaceFileReader
from tools.tools_image_analyzer import BlipImageAnalyzer
from tools.tools_script_excuter import PythonCodeExecutor
from tools.tools_video import VideoTool
from tools.tools_web_scrapper import SearchTool
from tools.youtube_rapidapi_official_search_provider import YouTubeRapidAPISearchProvider
from tools.youtube_transcriptor_provider import YouTubeTranscriptProvider
from langchain.agents import initialize_agent, AgentType
from langchain.agents import AgentExecutor
from dotenv import load_dotenv


class CustomAgent:
    def __init__(self,params:AgentParamsDataClass):
        self.params=params
        self._organize_params()
        self.read_file_content=self.params.read_file_content_tool or self.read_file_content
        self.web_search=self.params.web_search_tool or self.web_search 
        self.analyze_image=self.params.analyze_image_tool or self.analyze_image 
        self.audio_transcriptor=self.params.audio_transcriptor_tool or self.audio_transcriptor 
        self.run_python_code_with_file=self.params.run_python_code_with_file_tool or self.run_python_code_with_file 
        self.tools_LangChain=self.params.tools_LangChain or self.default_tools_LangChain 
        self.youtube=self.youtube 
        self.prefix=self.params.prefix_tool or self.prefix 
        load_dotenv()



    def _organize_params(self):
            
        if not self.params.read_file_content_tool:
            self.read_file_content=HuggingFaceFileReader(
                huggingface_token=os.getenv("HUGGINGFACE_TOKEN"),
                base_url=os.getenv("HUGGINGFACE_BASE_URL")
                )
        if not self.params.web_search_tool:
            self.web_search=SearchTool()
        
        if not self.params.analyze_image_tool:
            print("token:",os.getenv("HUGGINGFACE_TOKEN"))
            print("url:",os.getenv("HUGGINGFACE_BASE_URL"))
            self.analyze_image=BlipImageAnalyzer(
                    huggingface_token=os.getenv("HUGGINGFACE_TOKEN"),
                    base_url=os.getenv("HUGGINGFACE_BASE_URL")
                )
        if not self.params.audio_transcriptor_tool:
            self.audio_transcriptor=WhisperAudioTranscription(
                    huggingface_token=os.getenv("HUGGINGFACE_TOKEN"),
                    base_url=os.getenv("HUGGINGFACE_BASE_URL")
                )
        if not self.params.search_provider_tool or not self.params.search_provider_tool :
            self.youtube=VideoTool(search_provider=YouTubeRapidAPISearchProvider(rapid_api_key=os.getenv("RAPIDAPI_KEY")),
                                            transcript_provider=YouTubeTranscriptProvider()
                                        )
        if  self.params.search_provider_tool and self.params.search_provider_tool :
            self.youtube=VideoTool(search_provider=self.params.search_provider_tool(rapid_api_key=os.getenv("RAPIDAPI_KEY")),
                                            transcript_provider=self.params.search_provider_tool()
                                        )
        if not self.params.run_python_code_with_file_tool :
            self.run_python_code_with_file=self.params.run_python_code_with_file_tool or PythonCodeExecutor(huggingface_token=os.getenv("HUGGINGFACE_TOKEN"),
                                                              base_url=os.getenv("HUGGINGFACE_BASE_URL")
                                                              )
        if not self.params.prefix_tool :
            self.prefix = """Tu es un agent intelligent qui résout des problèmes en suivant un raisonnement étape par étape.
                    Tu dois TOUJOURS répondre dans le format suivant :
                    Thought: ta réflexion
                    Action: le nom d’un outil parmi [{tools}]
                    Action Input: "ton entrée pour l’outil. Les pièces jointes sont préfixées par file_name"

                    OU, si tu as la réponse :

                    Thought: ta réflexion finale
                    Final Answer: la réponse

                    NE RÉPONDS JAMAIS avec un autre format. Même si tu ne sais pas, choisis un outil ou réponds Final Answer: "Je ne sais pas".
                    """
            
        if not self.params.tools_LangChain:
            self.default_tools_LangChain = [
                Tool(
                    name="query_wikipedia",
                    func=self.web_search.query_wikipedia,
                    description="Cherche des informations encyclopédiques sur un sujet."
                ),
                Tool(
                    name="search_youtube",
                    func=self.youtube.search,
                    description="Trouve des vidéos pertinentes sur YouTube."
                ),
                Tool(
                    name="find_transcripton",
                    func=self.youtube.transcript,
                    description="Utilise cet outil pour lire la transcription d'une vidéo YouTube à partir de son URL ou ID"
                ),
                Tool(
                    name="analyze_image",
                    func=self.analyze_image.analyze,
                    description="Analyse une image fournie (description ou contenu)."
                ),
                Tool(
                name="audio_file_checker",
                func=self.audio_transcriptor.transcribe,
                description="Vérifie si un fichier audio peut être traité, sinon affiche un message d'erreur."
                ),
                Tool(
                    name="defaultScrapping",
                    func=self.web_search.web_search,
                    description=(
                        "Outil par défaut pour chercher sur le web des informations générales "
                        "si aucun autre outil spécifique ne correspond à la requête."
                    )
                ),
                Tool(
                    name="readFileContent",
                    func=self.read_file_content.read,
                    description="Lit le contenu d’un fichier local (PDF, Excel, CSV ou TXT) et retourne son contenu texte."
                ), 
                Tool(
                    name="run_python_code_with_file",
                    func=self.run_python_code_with_file.run_code_with_file,
                    description=(
                        "Télécharge un fichier en utilisant file_name=... depuis HuggingFace, "
                        "puis exécute un script Python qui peut accéder à ce fichier via la variable 'file_path'."
                    )
                )
            ]        
            
    def get_agent(self)->AgentExecutor:
        # Configuration Gemini
        llm_gemini = ChatGoogleGenerativeAI(
            model="gemini-2.5-flash",
            temperature=0,
            google_api_key= os.getenv("GEMINI_API_KEY")
        )
        tools = self.tools_LangChain 
        # Initialisation de l'agent
        return initialize_agent(
            tools=tools,
            llm=llm_gemini,
            agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
            verbose=True,
            handle_parsing_errors=True,
            max_iterations=10,
            max_execution_time=60,
            agent_kwargs={"prefix": self.prefix}
        )