Spaces:
Sleeping
Sleeping
File size: 3,477 Bytes
e00e744 |
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 |
import importlib.metadata
from typing import Dict, Any, Optional, Type, Callable
from .utils import get_logger, get_model, validate_api_keys
logger = get_logger(__name__)
class AgentManager:
"""Manages plugin discovery and loading for the FrameworkAgent"""
def __init__(self):
self.agents: Dict[str, Dict[str, Any]] = {}
self.agent_factories: Dict[str, Callable] = {}
self._load_plugins()
# Validate API keys on initialization
validate_api_keys()
def _load_plugins(self):
"""Load all available agents plugins using entry points"""
try:
# Load plugins from the 'lineagentic.lf_algorithm.plugins' entry point group
for entry_point in importlib.metadata.entry_points(group='lineagentic.lf_algorithm.plugins'):
try:
agent_info = entry_point.load()
if callable(agent_info):
# If it's a function, assume it returns plugin info
agent_data = agent_info()
else:
# If it's already a dict/object
agent_data = agent_info
agent_name = agent_data.get('name', entry_point.name)
self.agents[agent_name] = agent_data
# Store the factory function if available
if 'factory_function' in agent_data:
self.agent_factories[agent_name] = agent_data['factory_function']
logger.info(f"Loaded plugin: {agent_name}")
except Exception as e:
logger.error(f"Failed to load plugin {entry_point.name}: {e}")
except Exception as e:
logger.error(f"Error loading plugins: {e}")
def get_agent(self, agent_name: str) -> Optional[Dict[str, Any]]:
"""Get agent information by name"""
return self.agents.get(agent_name)
def list_agents(self) -> Dict[str, Dict[str, Any]]:
"""List all available agents"""
return self.agents.copy()
def create_agent(self, agent_name: str, **kwargs) -> Any:
"""Create an agent instance using the agent's factory function"""
if agent_name not in self.agent_factories:
raise ValueError(f"Agent '{agent_name}' not found or has no factory function")
factory = self.agent_factories[agent_name]
# Pass the get_model function to the agent factory
kwargs['get_model_func'] = get_model
return factory(agent_name=agent_name, **kwargs)
def get_supported_operations(self) -> Dict[str, list]:
"""Get all supported operations from all agents"""
operations = {}
for agent_name, agent_info in self.agents.items():
supported_ops = agent_info.get('supported_operations', [])
for op in supported_ops:
if op not in operations:
operations[op] = []
operations[op].append(agent_name)
return operations
def get_agents_for_operation(self, operation: str) -> list:
"""Get all agents that support a specific operation"""
supported_ops = self.get_supported_operations()
return supported_ops.get(operation, [])
# Global agent manager instance
agent_manager = AgentManager() |