File size: 2,681 Bytes
21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f 9b927a8 11a3890 dd5985f 11a3890 612fa38 dd5985f 21132c7 dd5985f 21132c7 dd5985f 21132c7 dd5985f |
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 |
import os
from smolagents import CodeAgent, OpenAIServerModel, PythonInterpreterTool
from tools.fetch import fetch_webpage, search_web
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 myprompts import output_format
import dotenv
dotenv.load_dotenv()
# --- Models ---
openai_model = OpenAIServerModel(
model_id="gpt-3.5-turbo",
api_key=os.environ["OPENAI_API_KEY"],
api_base="https://api.openai.com/v1",
)
# --- Tool Agents ---
web_agent = CodeAgent(
model=openai_model,
tools=[search_web, fetch_webpage],
name="web_agent",
description="Searches the web and fetches content from webpages.",
max_steps=6,
verbosity_level=1,
)
audiovideo_agent = CodeAgent(
model=openai_model,
tools=[
get_youtube_transcript,
get_youtube_title_description,
get_text_transcript_from_audio_file,
analyze_image
],
name="audiovideo_agent",
description="Processes audio, video, and image content.",
max_steps=6,
verbosity_level=1,
)
# --- Answer Checker ---
def check_final_answer(answer: str, _) -> bool:
return isinstance(answer, str) and len(answer.strip()) <= 200
# --- Manager Agent ---
manager_agent = CodeAgent(
model=openai_model,
tools=[PythonInterpreterTool()],
managed_agents=[web_agent, audiovideo_agent],
name="manager_agent",
description="Manages sub-agents to answer complex queries.",
final_answer_checks=[check_final_answer],
verbosity_level=2,
max_steps=12,
)
# --- Entry Class ---
class MultiAgent:
def __init__(self):
print("✅ MultiAgent initialized")
def __call__(self, question: str) -> str:
mylog("MultiAgent", question)
try:
system_instruction = (
"You are the lead agent in a multi-agent AI system. Use the sub-agents wisely."
" Answer using reasoning, planning, and tools where needed."
" Stay concise and always format the answer exactly as required."
)
prompt = f"""
{system_instruction}
THE QUESTION:
{question.strip()}
{output_format.strip()}
"""
return str(manager_agent.run(prompt))
except Exception as e:
error_msg = f"An error occurred while processing the question: {e}"
print(error_msg)
return error_msg
if __name__ == "__main__":
example_question = "What was the actual enrollment of the Malko competition in 2023?"
agent = MultiAgent()
print(agent(example_question)) |