File size: 3,570 Bytes
b15be4b |
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 |
import os
import traceback
from dotenv import load_dotenv
from agno.agent import Agent
from agno.storage.agent.sqlite import SqliteAgentStorage
from agno.memory.agent import AgentMemory
from agno.models.nebius import Nebius
load_dotenv()
NEBIUS_API_KEY = os.getenv("NEBIUS_API_KEY")
DB_NAME = "hackathon.db"
storage = SqliteAgentStorage(table_name="hackathon_storage", db_file=DB_NAME)
memory = AgentMemory()
class AgentFactory:
def __init__(self, user_id, thread_id, agent_config: dict, knowledge_base):
self.user_id = user_id
self.thread_id = thread_id
self.agent_config = agent_config
self.knowledge_base = knowledge_base
async def routing_agent(self):
try:
routing_agent = Agent(
model=Nebius(
id="meta-llama/Meta-Llama-3.1-405B-Instruct",
temperature=0,
api_key=NEBIUS_API_KEY,
base_url="https://api.studio.nebius.com/v1/"
),
name="Routing Agent",
description="You are a helpful routing assistant. This agent is responsible for routing the user's message to the appropriate agent. Based on the question it has to provide response. If question relates to 'plot', 'chart', 'graph', 'visualize', 'visualization', 'visual','bar chart', 'line chart', 'pie chart', 'scatter plot', 'histogram', 'heatmap', 'dashboard', 'show me', 'display', 'draw', 'create chart','generate plot', 'make graph', 'data visualization', 'analytics','trends', 'comparison chart', 'infographic'. If the question relates to the visualization like above key points then respond with 'visualization' else respond 'normal'.",
instructions=[
"You should only respond with 'normal' or 'visualization'.",
"DO NOT add any delimiter between the response and the word 'normal' or 'visualization'.",
"Your response should be one word accordingly.",
],
show_tool_calls=True,
markdown=True,
debug_mode=True
)
return routing_agent
except Exception as e:
print("Error creating routing agent:", traceback.format_exc())
raise e
async def normal_and_reasoning_agent(self, tools, model_name) -> Agent:
try:
agent = Agent(
model=Nebius(
id=model_name, #meta-llama/Meta-Llama-3.1-405B-Instruct #Qwen/Qwen3-235B-A22B #Qwen/Qwen3-30B-A3B
temperature=0,
api_key=NEBIUS_API_KEY,
base_url="https://api.studio.nebius.com/v1/"
),
name=self.agent_config["name"],
description=self.agent_config["description"],
instructions=self.agent_config["instructions"],
tools=tools,
show_tool_calls=True,
markdown=True,
debug_mode=True,
knowledge=self.knowledge_base,
search_knowledge=True,
storage=storage,
memory=memory,
user_id=self.user_id,
add_history_to_messages=True,
session_id=self.thread_id,
num_history_responses=10
)
return agent
except Exception as e:
print("Error creating agent:", traceback.format_exc())
raise e |