|
|
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, |
|
|
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 |
