Class swarmauri_core.chains.IChain.IChain

swarmauri_core.chains.IChain.IChain

Bases: ABC

Defines the interface for a Chain within a system, facilitating the organized execution of a sequence of tasks or operations. This interface is at the core of orchestrating operations that require coordination between multiple steps, potentially involving decision-making, branching, and conditional execution based on the outcomes of previous steps or external data.

A chain can be thought of as a workflow or pipeline, where each step in the chain can perform an operation, transform data, or make decisions that influence the flow of execution.

Implementors of this interface are responsible for managing the execution order, data flow between steps, and any dynamic adjustments to the execution based on runtime conditions.

METHOD	DESCRIPTION
add_step	Adds a step to the chain.
remove_step	Removes a step from the chain.
execute	Executes the chain, potentially returning a result.

add_step abstractmethod

add_step(step, **kwargs)

Adds a new step to the chain. Steps are executed in the order they are added. Each step is represented by a Callable, which can be a function or method, with optional keyword arguments that specify execution aspects or data needed by the step.

PARAMETER	DESCRIPTION
step	The Callable representing the step to add to the chain. TYPE: IChainStep
**kwargs	Optional keyword arguments that provide additional data or configuration for the step when it is executed. TYPE: dict[str, Any] DEFAULT: {}

Source code in swarmauri_core/chains/IChain.py

@abstractmethod
def add_step(self, step: IChainStep, **kwargs: dict[str, Any]) -> None:
    """
    Adds a new step to the chain. Steps are executed in the order they are added.
    Each step is represented by a Callable, which can be a function or method, with
    optional keyword arguments that specify execution aspects or data needed by the step.

    Parameters:
        step (IChainStep): The Callable representing the step to add to the chain.
        **kwargs: Optional keyword arguments that provide additional data or configuration
                  for the step when it is executed.
    """
    pass

remove_step abstractmethod

remove_step(step)

Removes an existing step from the chain. This alters the chain's execution sequence by excluding the specified step from subsequent executions of the chain.

PARAMETER	DESCRIPTION
step	The Callable representing the step to remove from the chain. TYPE: IChainStep

Source code in swarmauri_core/chains/IChain.py

@abstractmethod
def remove_step(self, step: IChainStep) -> None:
    """
    Removes an existing step from the chain. This alters the chain's execution sequence
    by excluding the specified step from subsequent executions of the chain.

    Parameters:
        step (IChainStep): The Callable representing the step to remove from the chain.
    """
    pass

execute abstractmethod

execute(*args, **kwargs)

Initiates the execution of the chain. This involves invoking each step in the order they have been added to the chain, passing control from one step to the next, and optionally aggregating or transforming results along the way.

The execution process can incorporate branching, looping, or conditional logic based on the implementation, allowing for complex workflows to be represented and managed within the chain.

PARAMETER	DESCRIPTION
*args	Positional arguments passed to the first step in the chain. These can be data inputs or other values required for the chain's execution. TYPE: tuple[Any, ...] DEFAULT: ()
**kwargs	Keyword arguments that provide additional context, data inputs, or configuration for the chain's execution. These can be passed to individual steps or influence the execution flow of the chain. TYPE: dict[str, Any] DEFAULT: {}

RETURNS	DESCRIPTION
Any	The outcome of executing the chain. This could be a value produced by the final step, a collection of outputs from multiple steps, or any other result type as determined by the specific chain implementation. TYPE: Dict[str, Any]

Source code in swarmauri_core/chains/IChain.py

@abstractmethod
def execute(
    self, *args: tuple[Any, ...], **kwargs: dict[str, Any]
) -> Dict[str, Any]:
    """
    Initiates the execution of the chain. This involves invoking each step in the order
    they have been added to the chain, passing control from one step to the next, and optionally
    aggregating or transforming results along the way.

    The execution process can incorporate branching, looping, or conditional logic based on the
    implementation, allowing for complex workflows to be represented and managed within the chain.

    Parameters:
        *args: Positional arguments passed to the first step in the chain. These can be data inputs
               or other values required for the chain's execution.
        **kwargs: Keyword arguments that provide additional context, data inputs, or configuration
                  for the chain's execution. These can be passed to individual steps or influence
                  the execution flow of the chain.

    Returns:
        Any: The outcome of executing the chain. This could be a value produced by the final
             step, a collection of outputs from multiple steps, or any other result type as
             determined by the specific chain implementation.
    """
    pass