File size: 5,324 Bytes
7d8f8f7 9ccd162 7d8f8f7 e2a980c 8cb575b 7d8f8f7 8cb575b 7d8f8f7 46b277b a15d2f2 a865540 7d8f8f7 8cb575b 46b277b 7d8f8f7 |
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 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 |
# 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
from functools import wraps
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")
# Base model
base_model = InferenceClientModel(
provider="groq",
api_key=grok_api_key,
model_id="qwen/qwen3-32b",
requests_per_minute=20,
temperature=0.0, # No randomness
top_p=1.0, # Don’t truncate the probability distribution
)
# Wrap with rate limiting
My_Agent = base_model
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}") |