90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
import uuid
|
|
import time
|
|
from typing import Dict, List, Optional
|
|
import logging
|
|
|
|
from app.config import settings
|
|
from app.models import AgentConfig, AgentDefinition
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class AgentService:
|
|
"""
|
|
Service for managing Agent Definitions.
|
|
Currently uses in-memory storage.
|
|
Can be extended to support Database storage.
|
|
"""
|
|
def __init__(self):
|
|
# In-memory store: {agent_id: AgentDefinition}
|
|
self._store: Dict[str, AgentDefinition] = {}
|
|
|
|
# Populate with a default agent for convenience
|
|
self._create_default_agent()
|
|
|
|
def _create_default_agent(self):
|
|
default_id = "default-assistant"
|
|
if default_id not in self._store:
|
|
self._store[default_id] = AgentDefinition(
|
|
id=default_id,
|
|
name="Assistant",
|
|
description="A helpful default AI assistant",
|
|
model=settings.LLM_DEFAULT_MODEL,
|
|
instruction="You are a helpful assistant.",
|
|
tools=[],
|
|
created_at=time.time(),
|
|
updated_at=time.time()
|
|
)
|
|
|
|
def create_agent(self, config: AgentConfig) -> AgentDefinition:
|
|
"""Create and store a new agent definition."""
|
|
agent_id = str(uuid.uuid4())
|
|
definition = AgentDefinition(
|
|
id=agent_id,
|
|
**config.model_dump(),
|
|
created_at=time.time(),
|
|
updated_at=time.time()
|
|
)
|
|
self._store[agent_id] = definition
|
|
logger.info(f"Created new agent definition: {agent_id}")
|
|
return definition
|
|
|
|
def get_agent(self, agent_id: str) -> Optional[AgentDefinition]:
|
|
"""Retrieve an agent definition by ID."""
|
|
return self._store.get(agent_id)
|
|
|
|
def update_agent(self, agent_id: str, config: AgentConfig) -> Optional[AgentDefinition]:
|
|
"""Update an existing agent definition."""
|
|
if agent_id not in self._store:
|
|
return None
|
|
|
|
# Preserve original creation time and ID
|
|
original = self._store[agent_id]
|
|
updated = AgentDefinition(
|
|
id=agent_id,
|
|
**config.model_dump(),
|
|
created_at=original.created_at,
|
|
updated_at=time.time()
|
|
)
|
|
self._store[agent_id] = updated
|
|
logger.info(f"Updated agent definition: {agent_id}")
|
|
return updated
|
|
|
|
def list_agents(self) -> List[AgentDefinition]:
|
|
"""List all stored agents."""
|
|
return list(self._store.values())
|
|
|
|
def delete_agent(self, agent_id: str) -> bool:
|
|
"""Delete an agent definition."""
|
|
if agent_id in self._store:
|
|
del self._store[agent_id]
|
|
logger.info(f"Deleted agent definition: {agent_id}")
|
|
return True
|
|
return False
|
|
|
|
|
|
# Singleton instance
|
|
_agent_service = AgentService()
|
|
|
|
def get_agent_service() -> AgentService:
|
|
return _agent_service
|