Spaces:
Sleeping
Sleeping
| from typing import Dict, Any, Optional | |
| class AgentManager: | |
| """Manages a collection of agents with dynamic registration and retrieval.""" | |
| def __init__(self): | |
| """Initialize an empty agent registry.""" | |
| self._agents: Dict[str, Any] = {} | |
| def register_agent(self, agent_id: str, agent: Any) -> None: | |
| """ | |
| Register a new agent with a unique ID. | |
| Args: | |
| agent_id (str): Unique identifier for the agent. | |
| agent (Any): Agent object, expected to have specific structure (e.g., llm/history/prompt or agent/history). | |
| Raises: | |
| ValueError: If agent_id is already registered. | |
| """ | |
| if agent_id in self._agents: | |
| raise ValueError(f"Agent with ID '{agent_id}' is already registered.") | |
| self._agents[agent_id] = agent | |
| print(f"[INFO] Registered agent: {agent_id}") | |
| def get_agent(self, agent_id: str) -> Optional[Any]: | |
| """ | |
| Retrieve an agent by its ID. | |
| Args: | |
| agent_id (str): ID of the agent to retrieve. | |
| Returns: | |
| Optional[Any]: The agent if found, None otherwise. | |
| """ | |
| return self._agents.get(agent_id) | |
| def list_agents(self) -> list[str]: | |
| """ | |
| List all registered agent IDs. | |
| Returns: | |
| list[str]: List of registered agent IDs. | |
| """ | |
| return list(self._agents.keys()) |