import json import os from langchain.agents import Tool from langchain_google_genai import ChatGoogleGenerativeAI from langchain.agents import initialize_agent, AgentType import time import pandas as pd from dotenv import load_dotenv from agent.agent import CustomAgent 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 # # _______________----------Charger les variables d'environnement--------------___________ load_dotenv() # # ___________===================Tools LangChain config=============________________ # read_file_content=HuggingFaceFileReader(huggingface_token=os.getenv("HUGGINGFACE_TOKEN"),base_url=os.getenv("HUGGINGFACE_BASE_URL")) # web_search=SearchTool() # youtube=VideoTool(search_provider=YouTubeRapidAPISearchProvider(rapid_api_key=os.getenv("RAPIDAPI_KEY")),transcript_provider=YouTubeTranscriptProvider()) # analyze_image=BlipImageAnalyzer(huggingface_token=os.getenv("HUGGINGFACE_TOKEN"), base_url=os.getenv("HUGGINGFACE_BASE_URL")) # audio_transcriptor=WhisperAudioTranscription(huggingface_token=os.getenv("HUGGINGFACE_TOKEN"), base_url=os.getenv("HUGGINGFACE_BASE_URL")) # run_python_code_with_file=PythonCodeExecutor(huggingface_token=os.getenv("HUGGINGFACE_TOKEN"),base_url=os.getenv("HUGGINGFACE_BASE_URL")) # tools_LangChain = [ # Tool( # name="query_wikipedia", # func=web_search.query_wikipedia, # description="Cherche des informations encyclopédiques sur un sujet." # ), # Tool( # name="search_youtube", # func=youtube.search, # description="Trouve des vidéos pertinentes sur YouTube." # ), # Tool( # name="find_transcripton", # func=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=analyze_image.analyze, # description="Analyse une image fournie (description ou contenu)." # ), # Tool( # name="audio_file_checker", # func=audio_transcriptor.transcribe, # description="Vérifie si un fichier audio peut être traité, sinon affiche un message d'erreur." # ), # Tool( # name="defaultScrapping", # func=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=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=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'." # ) # ) # ] # # prompt # 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". # """ # # Configuration Gemini # llm_gemini = ChatGoogleGenerativeAI( # model="gemini-2.5-flash", # temperature=0, # google_api_key= os.getenv("GEMINI_API_KEY") # ) # tools = tools_LangChain # # Initialisation de l'agent # agent_gemini = 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": prefix} # ) # ======================----------------------------==================== if __name__=="__main__": # question = quest[1].get('question') # task_id="1f975693-876d-457b-a649-393859e79bf3.mp3" # task_id="cca530fc-4052-43b2-b130-b30968d8aa44" quest=json.load(open(os.getenv("FILE_PATH"))) df = pd.DataFrame(columns=['reponse', 'id']) params=AgentParamsDataClass agent_gemini=CustomAgent(params=AgentParamsDataClass) for question in quest[:1]: questionF = question.get('question') if question.get('file_name'): questionF += f",file_name={question.get('file_name')}" print("questionF:", questionF) response = agent_gemini.get_agent().run(input=f"{questionF}", tools=agent_gemini.tools_LangChain) df = pd.concat([df, pd.DataFrame({'reponse': [response], 'id': [question.get('task_id')]})], ignore_index=True) time.sleep(60) # response = agent_HF.run(question) print("\n Réponse finale :\n", response)