from abc import ABC, abstractmethod
from typing import Any, Callable, Dict, List, NamedTuple, Optional, Type, Union
from swarmauri.core.swarms.ISwarm import ISwarm
from swarmauri.core.chains.ICallableChain import ICallableChain
from swarmauri.core.agents.IAgent import IAgent
[docs]
class Step(NamedTuple):
description: str
callable: Callable # Reference to the function to execute
args: Optional[List[Any]] = None
kwargs: Optional[Dict[str, Any]] = None
[docs]
class CallableChainItem(NamedTuple):
key: str # Unique identifier for the item within the chain
execution_context: Dict[str, Any] # Execution context and metadata
steps: List[Step]
[docs]
class AgentDefinition(NamedTuple):
type: str
configuration: Dict[str, Any]
capabilities: List[str]
dependencies: List[str]
execution_context: Dict[str, Any]
[docs]
class FunctionParameter(NamedTuple):
name: str
type: Type
default: Optional[Any] = None
required: bool = True
[docs]
class FunctionDefinition(NamedTuple):
identifier: str
parameters: List[FunctionParameter]
return_type: Type
execution_context: Dict[str, Any]
callable_source: Callable
[docs]
class ISwarmFactory(ABC):
[docs]
@abstractmethod
def create_swarm(self, *args, **kwargs) -> ISwarm:
"""
Creates and returns a new swarm instance configured with the provided arguments.
"""
pass
[docs]
@abstractmethod
def create_agent(self, agent_definition: AgentDefinition) -> IAgent:
"""
Creates a new agent based on the provided enhanced agent definition.
Args:
agent_definition: An instance of AgentDefinition detailing the agent's setup.
Returns:
An instance or identifier of the newly created agent.
"""
pass
[docs]
@abstractmethod
def create_callable_chain(self, chain_definition: List[CallableChainItem]) -> ICallableChain:
"""
Creates a new callable chain based on the provided definition.
Args:
chain_definition: Details required to build the chain, such as sequence of functions and arguments.
Returns:
ICallableChain: The constructed callable chain instance.
"""
pass
[docs]
@abstractmethod
def register_function(self, function_definition: FunctionDefinition) -> None:
"""
Registers a function within the factory ecosystem, making it available for callable chains and agents.
Args:
function_definition: An instance of FunctionDefinition detailing the function's specification.
"""
pass
[docs]
@abstractmethod
def export_callable_chains(self, format_type: str = 'json') -> Union[dict, str, bytes]:
"""
Exports configurations of all callable chains in the specified format.
Supported formats: 'json', 'pickle'.
Args:
format_type (str): The format for exporting the configurations.
Returns:
Union[dict, str, bytes]: The callable chain configurations in the specified format.
"""
pass
[docs]
@abstractmethod
def load_callable_chains(self, chains_data, format_type: str = 'json'):
"""
Loads callable chain configurations from given data.
Args:
chains_data (Union[dict, str, bytes]): Data containing callable chain configurations.
format_type (str): The format of the provided chains data.
"""
pass
[docs]
@abstractmethod
def export_configuration(self, format_type: str = 'json') -> Union[dict, str, bytes]:
"""
Exports the swarm's and agents' configurations in the specified format.
Supported formats: 'json', 'pickle'. Default is 'json'.
Args:
format_type (str): The format for exporting the configurations.
Returns:
Union[dict, str, bytes]: The configurations in the specified format.
"""
pass