import os from dotenv import load_dotenv load_dotenv() # Loads variables from .env into the environment # os.environ["LANGFUSE_HOST"] = os.getenv("LANGFUSE_HOST") # os.environ["LANGFUSE_PUBLIC_KEY"] = os.getenv("LANGFUSE_PUBLIC_KEY") # os.environ["LANGFUSE_SECRET_KEY"] = os.getenv("LANGFUSE_SECRET_KEY") # os.environ["HF_TOKEN"] = os.getenv("HF_token") # os.environ["GEMINI_API_KEY"] = os.getenv("GEMINI_API_KEY") #os.environ["TAVILY_API_KEY"] = os.getenv("TRAVILY_SECRET_KEY") from langfuse import get_client langfuse = get_client() # Verify connection if langfuse.auth_check(): print("Langfuse client is authenticated and ready!") else: print("Authentication failed. Please check your credentials and host.") from smolagents import OpenAIServerModel gemini_model = OpenAIServerModel( model_id="gemini-2.5-flash", api_base="https://generativelanguage.googleapis.com/v1beta/openai/", api_key=os.getenv("GEMINI_API_KEY") ) # from smolagents import TransformersModel # VL_model = TransformersModel( # model_id="Qwen/Qwen2.5-VL-32B-Instruct", # device_map="auto", # torch_dtype="auto", # ) # coder_model = TransformersModel( # model_id="Qwen/Qwen2.5-Coder-32B-Instruct", # device_map="auto", # torch_dtype="auto", # ) #from Tools.searchtool import WebSearchTool from Tools.mywebpagevisit import CustomVisitWebpageTool from smolagents import ( CodeAgent, WebSearchTool, #WikipediaSearchTool ) web_search = WebSearchTool() visit_webpage = CustomVisitWebpageTool() #wikipedia_search = WikipediaSearchTool() from openinference.instrumentation.smolagents import SmolagentsInstrumentor SmolagentsInstrumentor().instrument() web_agent = CodeAgent( tools=[WebSearchTool,CustomVisitWebpageTool], model= gemini_model, additional_authorized_imports = ['requests', 'bs4','json'], name="web_agent", description="This is an agent that can do web search using 'WebSearchTool' and visit a webpage using 'CustomVisitWebpageTool'." ) from Tools.ytdownload import youtube_audio_downloader,youtube_video_downloader yt_agent = CodeAgent( tools=[youtube_audio_downloader,youtube_video_downloader], model= gemini_model, name="youtube_downloader_agent", description="This is an agent that can download youtube audio file using 'youtube_audio_downloader'and youtube video file using 'youtube_video_downloader'." ) from Tools.transcriber import transcriber transcriber_agent = CodeAgent( tools=[transcriber], model= gemini_model, name="transcriber_agent", description="This is an agent that can transcribe a local audio file using 'transcriber'.\n\nDo not remove adjectives or descriptors of ingredients if not specifically mentioned." ) from Tools.visual_reasoner import video_reasoner, image_reasoner VL_agent = CodeAgent( tools=[video_reasoner,image_reasoner], model= gemini_model, name="VL_agent", description="This is an agent that can answer questions on a video using 'video_reasoner' and on a image using 'image_reasoner'." ) excel_agent = CodeAgent( tools=[], model= gemini_model, name="excel_agent", description="This is an agent that can work with excel.", add_base_tools=False, additional_authorized_imports = ['pandas', 'numpy', 'math'] ) from Tools.code_reader import code_reader manager_agent = CodeAgent( tools=[code_reader], model=gemini_model, managed_agents=[web_agent, yt_agent, transcriber_agent, VL_agent], additional_authorized_imports=[ "numpy","os","pandas",'scipy','regex','math' ], planning_interval=5, verbosity_level=2, max_steps=10, 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." ) # manager_agent.run( # "How many studio albums were published by Mercedes Sosa between 2000 and 2009 (included)? You can use the latest 2022 version of english wikipedia." # ) class MyAgent: def __init__(self): print("BasicAgent initialized.") def __call__(self, question: str) -> str: print(f"Agent received question (first 50 chars): {question[:50]}...") try: answer = manager_agent.run(question) return answer except Exception as e: error = f"An error occurred while processing the question: {e}" print(error) return error # if __name__ == "__main__": # question = """ # What was the actual enrollment of the Malko competition in 2023? # """ # agent = MyAgent() # answer = agent(question) # print(f"Answer: {answer}")