Create multiagents.py

multiagents.py

@@ -0,0 +1,183 @@
+
+# a multi agent proposal to solve HF agent course final assignment
+import os
+import dotenv
+from smolagents import CodeAgent
+#from smolagents import OpenAIServerModel
+from tools.fetch import fetch_webpage, search_web
+from smolagents import PythonInterpreterTool, InferenceClientModel
+from tools.yttranscript import get_youtube_transcript, get_youtube_title_description
+from tools.stt import get_text_transcript_from_audio_file
+from tools.image import analyze_image
+from common.mylogger import mylog
+from smolagents import LiteLLMModel  # Import LiteLLMModel instead of OpenAIServerModel
+import os
+#from huggingface_hub import InferenceClient
+import myprompts
+
+from groq_api import GrokApi
+
+
+dotenv.load_dotenv()
+
+# gemini_model = OpenAIServerModel(
+#     model_id="gemini-2.0-flash",
+#     api_key=os.environ["GEMINI_API_KEY"],
+#     # Google Gemini OpenAI-compatible API base URL
+#     api_base="https://generativelanguage.googleapis.com/v1beta/openai/",
+# )
+
+# vllm_model = OpenAIServerModel(
+#     model_id="Qwen/Qwen2.5-1.5B-Instruct",
+#     api_base="http://192.168.1.39:18000/v1",  
+#     api_key="token-abc123",  
+# )
+
+# openai_41nano_model = OpenAIServerModel(
+#     model_id="gpt-4.1-nano",
+#     api_base="https://api.openai.com/v1",
+#     api_key=os.environ["OPENAI_API_KEY"],
+# )
+
+# openai_41mini_model = OpenAIServerModel(
+#     model_id="gpt-4.1-mini",
+#     api_base="https://api.openai.com/v1",
+#     api_key=os.environ["OPENAI_API_KEY"],
+# )
+
+
+
+# # --- Agent wrappers ---
+# groq_model = LiteLLMModel(
+#     model_id="groq/qwen3-32b",  # or any other Groq model like groq/mixtral-8x7b-32768
+#     api_key = os.getenv("GROQ_API_KEY"),
+#     temperature=0.1,
+#     max_tokens=4000,
+# )
+
+# SETUP AND TEST
+
+
+
+grok_api_key = os.getenv("groq_api")
+
+#InferenceClientModel InferenceClient
+My_Agent = InferenceClientModel(
+    provider="groq",
+    api_key=grok_api_key,
+    model_id = "qwen/qwen3-32b"
+)
+
+# test_messages = [{"role": "user", "content": "What are the 3 laws of robotics"}]
+# output_test = My_Agent(test_messages)
+# print('1', output_test)
+
+
+
+# My_Agent = client.chat.completions.create(
+#     model="qewn/qwen3-32b",
+#     messages=[
+#         {
+#             "role": "user",
+#             "content": "How many 'G's in 'huggingface'?"
+#         }
+#     ],
+# )
+
+
+
+def check_final_answer(final_answer, agent_memory)  -> bool:
+    """
+    Check if the final answer is correct.
+    basic check on the length of the answer.
+    """
+    mylog("check_final_answer", final_answer)
+    # if return answer is more than 200 characters, we will assume it is not correct    
+    if len(str(final_answer)) > 200:
+        return False
+    else:
+        return True
+
+
+web_agent = CodeAgent(
+    model=My_Agent,
+    tools=[
+        search_web,
+        fetch_webpage,                
+    ],
+    name="web_agent",
+    description="Use search engine to find webpages related to a subject and get the page content",
+    additional_authorized_imports=["pandas", "numpy","bs4"],
+    verbosity_level=1,    
+    max_steps=7,
+)
+
+audiovideo_agent = CodeAgent(
+    model=My_Agent,
+    tools=[
+        get_youtube_transcript,
+        get_youtube_title_description,
+        get_text_transcript_from_audio_file,
+        analyze_image
+    ],
+    name="audiovideo_agent",
+    description="Extracts information from image, video or audio files from the web",
+    additional_authorized_imports=["pandas", "numpy","bs4", "requests"],
+    verbosity_level=1,
+    max_steps=7,
+)
+
+
+
+manager_agent = CodeAgent(
+    model=My_Agent,
+    tools=[ PythonInterpreterTool()],
+    managed_agents=[web_agent, audiovideo_agent],    
+    additional_authorized_imports=["pandas", "numpy","bs4"],
+    planning_interval=5,
+    verbosity_level=2,
+    final_answer_checks=[check_final_answer],
+    max_steps=15,
+    name="manager_agent",
+    description="A manager agent that coordinates the work of other agents to answer questions.",
+)
+
+class MultiAgent:
+    def __init__(self):
+        print("BasicAgent initialized.")
+
+    def __call__(self, question: str) -> str:
+        mylog(self.__class__.__name__, question)        
+
+        try:
+            prefix = """You are the top agent of a multi-agent system that can answer questions by coordinating the work of other agents.
+            You will receive a question and you will decide which agent to use to answer it.
+            You can use the web_agent to search the web for information and for fetching the content of a web page, or the audiovideo_agent to extract information from video or audio files.
+            You can also use your own knowledge to answer the question.
+            You need to respect the output format that is given to you.
+            Finding the correct answer to the question need reasoning and plannig, read the question carrefully, think step by step and do not skip any steps.
+            """
+
+            question = prefix + "\nTHE QUESTION:\n" + question + '\n' + myprompts.output_format
+
+            fixed_answer = ""
+
+            fixed_answer = manager_agent.run(question)
+            
+            return fixed_answer
+        except Exception as e:
+            error = f"An error occurred while processing the question: {e}"
+            print(error)
+            return error
+
+
+if __name__ == "__main__":
+    # Example usage
+
+    asyncio.run(main())
+    question = """
+What was the actual enrollment of the Malko competition in 2023?
+"""
+    agent = MultiAgent()
+    answer = agent(question)
+    print(f"Answer: {answer}")