submission/agent/agent.py

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