126 lines
4.3 KiB
Python
126 lines
4.3 KiB
Python
"""Agent Manager Service"""
|
|
import uuid
|
|
import json
|
|
import logging
|
|
from typing import List, Dict, Optional
|
|
|
|
from luxx.core.database import SessionLocal
|
|
from luxx.models.room import Agent
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AgentManager:
|
|
"""Service for managing agents"""
|
|
|
|
def get_agent(self, agent_id: str) -> Optional[Dict]:
|
|
"""Get an agent by ID"""
|
|
db = SessionLocal()
|
|
try:
|
|
agent = db.query(Agent).filter(Agent.id == agent_id).first()
|
|
return agent.to_dict() if agent else None
|
|
finally:
|
|
db.close()
|
|
|
|
def list_agents(self, include_inactive: bool = False) -> List[Dict]:
|
|
"""List all agents"""
|
|
db = SessionLocal()
|
|
try:
|
|
query = db.query(Agent)
|
|
if not include_inactive:
|
|
query = query.filter(Agent.is_active == True)
|
|
agents = query.order_by(Agent.priority).all()
|
|
return [a.to_dict() for a in agents]
|
|
finally:
|
|
db.close()
|
|
|
|
def create_agent(self, name: str, role: str, system_prompt: str, provider_id: int = None, model: str = None,
|
|
tools: List[str] = None, priority: int = 5, auto_response: bool = True, mention_trigger: bool = False,
|
|
temperature: float = 0.7, max_tokens: int = 2048, avatar: str = None) -> Dict:
|
|
"""Create a new agent"""
|
|
if not system_prompt:
|
|
system_prompt = "You are a helpful assistant."
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
agent = Agent(
|
|
id=str(uuid.uuid4()),
|
|
name=name,
|
|
role=role,
|
|
system_prompt=system_prompt,
|
|
provider_id=provider_id,
|
|
model=model,
|
|
tools=json.dumps(tools) if tools else None,
|
|
priority=priority,
|
|
auto_response=auto_response,
|
|
mention_trigger=mention_trigger,
|
|
temperature=str(temperature),
|
|
max_tokens=max_tokens,
|
|
avatar=avatar
|
|
)
|
|
db.add(agent)
|
|
db.commit()
|
|
return agent.to_dict()
|
|
finally:
|
|
db.close()
|
|
|
|
def update_agent(self, agent_id: str, name: str = None, role: str = None, system_prompt: str = None,
|
|
provider_id: int = None, model: str = None, tools: List[str] = None, priority: int = None,
|
|
auto_response: bool = None, mention_trigger: bool = None, temperature: float = None,
|
|
max_tokens: int = None, is_active: bool = None, avatar: str = None) -> Optional[Dict]:
|
|
"""Update an agent"""
|
|
db = SessionLocal()
|
|
try:
|
|
agent = db.query(Agent).filter(Agent.id == agent_id).first()
|
|
if not agent:
|
|
return None
|
|
|
|
if name is not None:
|
|
agent.name = name
|
|
if role is not None:
|
|
agent.role = role
|
|
if system_prompt is not None:
|
|
agent.system_prompt = system_prompt
|
|
if provider_id is not None:
|
|
agent.provider_id = provider_id
|
|
if model is not None:
|
|
agent.model = model
|
|
if tools is not None:
|
|
agent.tools = json.dumps(tools)
|
|
if priority is not None:
|
|
agent.priority = priority
|
|
if auto_response is not None:
|
|
agent.auto_response = auto_response
|
|
if mention_trigger is not None:
|
|
agent.mention_trigger = mention_trigger
|
|
if temperature is not None:
|
|
agent.temperature = str(temperature)
|
|
if max_tokens is not None:
|
|
agent.max_tokens = max_tokens
|
|
if is_active is not None:
|
|
agent.is_active = is_active
|
|
if avatar is not None:
|
|
agent.avatar = avatar
|
|
|
|
db.commit()
|
|
return agent.to_dict()
|
|
finally:
|
|
db.close()
|
|
|
|
def delete_agent(self, agent_id: str) -> bool:
|
|
"""Delete an agent"""
|
|
db = SessionLocal()
|
|
try:
|
|
agent = db.query(Agent).filter(Agent.id == agent_id).first()
|
|
if not agent:
|
|
return False
|
|
db.delete(agent)
|
|
db.commit()
|
|
return True
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# Global manager instance
|
|
agent_manager = AgentManager()
|