`fragile.core.api_classes`#

Module Contents#

Classes#

`SwarmComponent`	A component of a swarm simulation.
`EnvironmentAPI`	The Environment is in charge of stepping the walkers, acting as a state transition function.
`PolicyAPI`	The policy is in charge of calculating the interactions with the `Environment`.
`Callback`	The `Walkers` is a data structure that takes care of all the data involved in making a Swarm evolve.
`WalkersMetric`	A component of a swarm simulation.
`WalkersAPI`	The WalkersAPI class defines the base functionality for managing walkers in a swarm simulation.
`SwarmAPI`	The Swarm implements the iteration logic to make the `Walkers` evolve.

class fragile.core.api_classes.SwarmComponent(swarm=None, param_dict=None, inputs=None, outputs=None)[source]#

A component of a swarm simulation.

Every class that stores its data in SwarmState inherits from this class.

Parameters

swarm (Optional[SwarmAPI]) – Reference to the Swarm that incorporates this SwarmComponent. Defaults to None.
param_dict (Optional[fragile.core.typing.StateDict]) – Describes the multi-dimensional arrays that define the internal state of the component. Defaults to None.
inputs (Optional[fragile.core.typing.InputDict]) – Describes the data that this components processes as inputs. This data is present in the internal state of the Swarm, and generated by other components. Defaults to None.
outputs (Optional[Tuple[str]]) – Name of the data attributes that this component makes available as part of its internal state. Defaults to None.

swarm#

The parent Swarm instance.

Type: Swarm
Return type: SwarmAPI

n_walkers#

Number of walkers in the Swarm.

Type: int
Return type: int

inputs#

A dictionary mapping input names to their corresponding type and default values.

Type: Dict[str, Dict[str, Any]]
Return type: fragile.core.typing.InputDict

outputs#

A tuple containing the names of the data attributes.

Type: Tuple[str]
Return type: Tuple[str, Ellipsis]

param_dict#

A dictionary mapping parameter names.

Type: Dict[str, Dict[str, Any]]
Return type: fragile.core.typing.StateDict

get(...)[source]#

Parameters

name (str) –
default (Any) –
raise_error (bool) –
inactives (bool) –

Return type

Any

get_input_data(...)[source]#

Return type: fragile.core.typing.StateData

update(...)[source]#

Parameters

other (Union[fragile.core.state.SwarmState, Dict[str, fragile.core.typing.Tensor]]) –
inactives (bool) –

Return type

None

reset(...)[source]#

Parameters

inplace (bool) –
root_walker (Optional[fragile.core.typing.StateData]) –
states (Optional[fragile.core.typing.StateData]) –

_prepare_tensors(...)[source]#

default_inputs#

default_outputs#

default_param_dict#

property swarm#

Return a reference to the Swarm that includes the current component.

Return type: SwarmAPI

property n_walkers#

Return the number of walkers of the Swarm.

Return type: int

property inputs#

Return a dictionary containing the data that this component needs to function.

Return type: fragile.core.typing.InputDict

property outputs#

Return a tuple containing the names of the data attribute that the component outputs.

Return type: Tuple[str, Ellipsis]

property param_dict#

Return the dictionary defining all the data attributes that the component requires.

Return type: fragile.core.typing.StateDict

setup(swarm)[source]#: Prepare the component during the setup phase of the Swarm.

get(name, default=None, raise_error=False, inactives=False)[source]#

Access attributes of the Swarm and its children.

Parameters

name (str) –
default (Any) –
raise_error (bool) –
inactives (bool) –

Return type

Any

get_input_data()[source]#

Return a dictionary with all the data that this component requires as inputs.

Returns: Dictionary containing all the data required by the SwarmComponent, where each key is the name of a data attribute and its value a multi-dimensional array.
Return type: fragile.core.typing.StateData

update(other=None, inactives=False, **kwargs)[source]#

Modify the data stored in the SwarmState instance.

Existing attributes will be updated, and new attributes will be created if needed.

Parameters

other (Union[fragile.core.state.SwarmState, Dict[str, fragile.core.typing.Tensor]]) – State class that will be copied upon update.
inactives (bool) – Whether to update the walkers marked as inactive.
**kwargs – It is possible to specify the update as name value attributes, where name is the name of the attribute to be updated, and value is the new value for the attribute.

Returns

None

Return type

None

_prepare_tensors(**kwargs)[source]#

reset(inplace=True, root_walker=None, states=None, **kwargs)[source]#

Reset the internal state of the SwarmComponent.

Parameters

inplace (bool, optional) – Unused. Defaults to True.
root_walker (Optional[StateData], optional) – Set the internal state of the SwarmComponent to this value. Defaults to None.
states (Optional[StateData], optional) – Set the internal state of the SwarmComponent to this value. Defaults to None.
kwargs – Other parameters required to reset the component.

class fragile.core.api_classes.EnvironmentAPI(action_shape, action_dtype, observs_shape, observs_dtype, swarm=None)[source]#

Bases: SwarmComponent

The Environment is in charge of stepping the walkers, acting as a state transition function.

For every different problem, a new Environment needs to be implemented following the EnvironmentAPI interface.

Parameters

action_shape (tuple) –
observs_shape (tuple) –
swarm (SwarmAPI) –

default_inputs#

default_outputs = ['observs', 'rewards', 'oobs']#

property observs_shape#

Return the shape of the observations.

Return type: tuple

property observs_dtype#: Return the dtype of the observations.

property action_shape#

Return the shape of the actions.

Return type: tuple

property action_dtype#: Return the dtype of the actions.

property param_dict#

Return the dictionary defining all the data attributes that the component requires.

Return type: fragile.core.typing.StateDict

abstract step(**kwargs)[source]#

Return the data corresponding to the new state of the environment after using the input data to make the corresponding state transition.

Return type: fragile.core.typing.StateData

make_transitions(inplace=True, inactives=False, **kwargs)[source]#

Return the data corresponding to the new state of the environment after using the input data to make the corresponding state transition.

Parameters

inplace (bool) – If False return the new data. If True, update the state of the Swarm.
inactives (bool) – Whether to update the walkers marked as inactive.
**kwargs – Keyword arguments passed if the returned value from the states_to_data function of the class was a dictionary.

Returns

Dictionary containing the data representing the state of the environment after the state transition. The keys of the dictionary are the names of the data attributes and its values are arrays representing a batch of new values for that attribute.

The StatesEnv returned by step will contain the returned data.

Return type

Union[None, fragile.core.typing.StateData]

reset(inplace=True, root_walker=None, states=None, inactives=True, **kwargs)[source]#

Reset the internal state of the SwarmComponent.

Parameters

inplace (bool, optional) – If False return the new data. If True, update the state of the Swarm. Defaults to True.
root_walker (Optional[StateData], optional) – _description_. Defaults to None.
states (Optional[StateData], optional) – _description_. Defaults to None.
inactives (bool, optional) – Whether to update the walkers marked as inactive. Defaults to True.
kwargs – Other arguments passed to make_transitions.

Returns

The StatesEnv returned by step will contain the returned data.

Return type

Union[None, StateData]

class fragile.core.api_classes.PolicyAPI(swarm=None, param_dict=None, inputs=None, outputs=None)[source]#

Bases: SwarmComponent

The policy is in charge of calculating the interactions with the Environment.

The PolicyAPI class is responsible for defining the policy that determines the actions for interacting with the Environment in a swarm simulation. This is an abstract base class, and specific policy implementations should inherit from this class and implement the ‘select_actions’ method.

Parameters

swarm (Optional[SwarmAPI]) –
param_dict (Optional[fragile.core.typing.StateDict]) –
inputs (Optional[fragile.core.typing.InputDict]) –
outputs (Optional[Tuple[str]]) –

default_outputs#

abstract select_actions(**kwargs)[source]#

Select actions for each walker in the swarm based on the current state.

This method must be implemented by subclasses.

Parameters: **kwargs – Additional keyword arguments required for selecting actions.
Returns: The selected actions as a Tensor or a StateData dictionary.
Return type: Union[Tensor, StateData]

act(inplace=True, **kwargs)[source]#

Calculate SwarmState containing the data needed to interact with the environment.

Parameters

inplace (bool, optional) – If True, updates the swarm state with the selected actions. If False, returns the selected actions. Defaults to True.
**kwargs – Additional keyword arguments required for selecting actions.

Returns

None if inplace is True. Otherwise, a StateData dictionary: containing the selected actions.

Return type

Union[None, StateData]

reset(inplace=True, root_walker=None, states=None, **kwargs)[source]#

Reset the internal state of the PolicyAPI.

Parameters

inplace (bool, optional) – If True, updates the swarm state with the selected actions. If False, returns the selected actions. Defaults to True.
root_walker (Optional[StateData], optional) – Set the internal state of the PolicyAPI to this value. Defaults to None.
states (Optional[StateData], optional) – Set the internal state of the PolicyAPI to this value. Defaults to None.
**kwargs – Other parameters required to reset the component.

Returns

None if inplace is True. Otherwise, a StateData dictionary: containing the selected actions.

Return type

Union[None, StateData]

class fragile.core.api_classes.Callback(swarm=None, param_dict=None, inputs=None, outputs=None)[source]#

Bases: SwarmComponent

The Walkers is a data structure that takes care of all the data involved in making a Swarm evolve.

Parameters

swarm (Optional[SwarmAPI]) –
param_dict (Optional[fragile.core.typing.StateDict]) –
inputs (Optional[fragile.core.typing.InputDict]) –
outputs (Optional[Tuple[str]]) –

name#

before_reset()[source]#

after_reset()[source]#

before_evolve()[source]#

after_evolve()[source]#

before_policy()[source]#

after_policy()[source]#

before_env()[source]#

after_env()[source]#

before_walkers()[source]#

after_walkers()[source]#

reset(inplace=True, root_walker=None, states=None, **kwargs)[source]#

Reset the internal state of the SwarmComponent.

Parameters

inplace (bool, optional) – Unused. Defaults to True.
root_walker (Optional[StateData], optional) – Set the internal state of the SwarmComponent to this value. Defaults to None.
states (Optional[StateData], optional) – Set the internal state of the SwarmComponent to this value. Defaults to None.
kwargs – Other parameters required to reset the component.

evolution_end()[source]#

Return type: bool

run_end()[source]#

class fragile.core.api_classes.WalkersMetric(swarm=None, param_dict=None, inputs=None, outputs=None)[source]#

Bases: SwarmComponent

A component of a swarm simulation.

Every class that stores its data in SwarmState inherits from this class.

Parameters

swarm (Optional[SwarmAPI]) – Reference to the Swarm that incorporates this SwarmComponent. Defaults to None.
param_dict (Optional[fragile.core.typing.StateDict]) – Describes the multi-dimensional arrays that define the internal state of the component. Defaults to None.
inputs (Optional[fragile.core.typing.InputDict]) – Describes the data that this components processes as inputs. This data is present in the internal state of the Swarm, and generated by other components. Defaults to None.
outputs (Optional[Tuple[str]]) – Name of the data attributes that this component makes available as part of its internal state. Defaults to None.

swarm#

The parent Swarm instance.

Type: Swarm
Return type: SwarmAPI

n_walkers#

Number of walkers in the Swarm.

Type: int
Return type: int

inputs#

A dictionary mapping input names to their corresponding type and default values.

Type: Dict[str, Dict[str, Any]]
Return type: fragile.core.typing.InputDict

outputs#

A tuple containing the names of the data attributes.

Type: Tuple[str]
Return type: Tuple[str, Ellipsis]

param_dict#

A dictionary mapping parameter names.

Type: Dict[str, Dict[str, Any]]
Return type: fragile.core.typing.StateDict

get(...)#

Parameters

name (str) –
default (Any) –
raise_error (bool) –
inactives (bool) –

Return type

Any

get_input_data(...)#

Return type: fragile.core.typing.StateData

update(...)#

Parameters

other (Union[fragile.core.state.SwarmState, Dict[str, fragile.core.typing.Tensor]]) –
inactives (bool) –

Return type

None

reset(...)[source]#

Parameters

inplace (bool) –
root_walker (Optional[fragile.core.typing.StateData]) –
states (Optional[fragile.core.typing.StateData]) –

_prepare_tensors(...)#

__call__(inplace=True, **kwargs)[source]#

Parameters: inplace (bool) –
Return type: fragile.core.typing.Tensor

abstract calculate(**kwargs)[source]#

reset(inplace=True, root_walker=None, states=None, **kwargs)[source]#

Reset the internal state of the SwarmComponent.

Parameters

inplace (bool, optional) – Unused. Defaults to True.
root_walker (Optional[StateData], optional) – Set the internal state of the SwarmComponent to this value. Defaults to None.
states (Optional[StateData], optional) – Set the internal state of the SwarmComponent to this value. Defaults to None.
kwargs – Other parameters required to reset the component.

class fragile.core.api_classes.WalkersAPI(swarm=None, param_dict=None, inputs=None, outputs=None)[source]#

Bases: SwarmComponent

The WalkersAPI class defines the base functionality for managing walkers in a swarm simulation. This class inherits from the SwarmComponent class.

Parameters

swarm (Optional[SwarmAPI]) –
param_dict (Optional[fragile.core.typing.StateDict]) –
inputs (Optional[fragile.core.typing.InputDict]) –
outputs (Optional[Tuple[str]]) –

get_input_data()[source]#

Return a dictionary with all the data that this component requires as inputs, including the data for inactive walkers.

Returns

A dictionary containing all the required data for the WalkersAPI component,: where each key is the name of a data attribute and its value is a multi-dimensional array.

Return type

StateData

update(other=None, inactives=True, **kwargs)[source]#

Modify the data stored in the SwarmState instance.

Existing attributes will be updated, and new attributes will be created if needed.

Parameters

other (Union[fragile.core.state.SwarmState, Dict[str, fragile.core.typing.Tensor]]) – State class that will be copied upon update.
inactives (bool) – Whether to update the walkers marked as inactive.
**kwargs – It is possible to specify the update as name value attributes, where name is the name of the attribute to be updated, and value is the new value for the attribute.

Returns

None

Return type

None

balance(inplace=True, **kwargs)[source]#

Perform a balance operation on the swarm state.

Parameters

inplace (bool, optional) – If True, updates the swarm state with the balanced data. If False, returns the balanced data. Defaults to True.
**kwargs – Additional keyword arguments required for running the balance operation.

Returns

None if inplace is True. Otherwise, a StateData dictionary: containing the balanced data.

Return type

Union[None, StateData]

abstract run_epoch(inplace=True, **kwargs)[source]#

Implement the functionality for running an epoch in the derived class. This method is called during the balance operation.

Parameters

inplace (bool, optional) – If True, updates the swarm state with the data generated during the epoch. If False, returns the data. Defaults to True.
**kwargs – Additional keyword arguments required for running the epoch.

Returns

A dictionary containing the data generated during the epoch if inplace is: False. Otherwise, returns None.

Return type

StateData

reset(inplace=True, **kwargs)[source]#

Reset the internal state of the Walkers class. This method should be implemented in the derived class.

Parameters

inplace (bool, optional) – Unused. Defaults to True.
**kwargs – Additional keyword arguments required for resetting the component.

Returns

None

get_in_bounds_compas(oobs=None)[source]#

Return an array of indexes corresponding to alive walkers chosen at random.

Parameters: oobs (Optional[np.ndarray], optional) – An optional boolean array indicating out-of-bounds walkers. If not provided, all walkers are considered alive. Defaults to None.
Returns: An array of indexes corresponding to randomly chosen alive walkers.
Return type: Tensor

clone_walkers(will_clone=None, compas_clone=None, **kwargs)[source]#

Sample the clone probability distribution and clone the walkers accordingly.

Parameters

will_clone (Optional[np.ndarray], optional) – A boolean array indicating which walkers will be cloned. If not provided, the clone operation will be performed on all walkers. Defaults to None.
compas_clone (Optional[np.ndarray], optional) – An array of indexes indicating the walkers to be cloned. If not provided, random alive walkers will be chosen. Defaults to None.
**kwargs – Additional keyword arguments required for cloning the walkers.

Returns

None

Return type

None

class fragile.core.api_classes.SwarmAPI(n_walkers, env, policy, walkers, callbacks=None, minimize=False, max_epochs=1e+100)[source]#

The Swarm implements the iteration logic to make the Walkers evolve.

It contains the necessary logic to use an Environment, a Model, and a Walkers instance to create the algorithm execution loop.

This class defines a method called run() that receives two optional arguments, root_walker and state, and has no return value. This method runs the fractal AI Swarm evolution process until a stop condition is met.

In its implementation, it calls several other methods: (before_reset(), reset(), after_reset(), evolve(), before_evolve(), after_evolve(), evolution_end(), before_env(), after_env(), before_policy(), after_policy(), before_walkers(), after_walkers(), run_end()) defined within the same class, which are mainly used to manage different aspects of the search process or to invoke user-defined callbacks.

The evolve() method updates the states of the search environment and model, makes the walkers undergo a perturbation process, and balances them. It also invokes several other methods that trigger user-defined callbacks.

The evolution_end() method returns True if any of the following conditions is met: 1. The current epoch exceeds the maximum allowed epochs. 2. All walkers are out of the problem domain. 3. Any callback of the class has set the evolution_end flag to True.

walkers_last#

If True indicates that the Walkers class runs after acting on the environment. If Fase, the walkers run before acting on the environment.

Type: bool

Parameters

n_walkers (int) – The number of walkers in the swarm.
env (EnvironmentAPI) – An environment that simulates the objective function.
policy (PolicyAPI) – A policy that defines how the individuals evolve.
walkers (WalkersAPI) – A set of motion rules to control a population of walkers.
callbacks (Optional[Iterable[Callback]]) – A list of functions to call at each iteration.
minimize (bool) – If True, take the minimum value of fitness, else take the maximum.
max_epochs (int) – Maximum number of epochs allowed before the swarm search is stopped.

walkers_last = True#

property n_walkers#

Return the number of walkers in the swarm.

Return type: int

property n_actives#

Returns the number of active walkers in the swarm.

Return type: int

property epoch#

Return the current epoch of the search algorithm.

Return type: int

property state#

Returns the state instance describing the walkers of the swarm.

Return type: fragile.core.state.SwarmState

property env#

All the simulation code (problem specific) will be handled here.

Return type: EnvironmentAPI

property policy#

All the policy and random perturbation code (problem specific) will be handled here.

Return type: PolicyAPI

property walkers#

Access the Walkers in charge of implementing the FAI evolution process.

Return type: WalkersAPI

property callbacks#

Return the dictionary containing all the user-defined callbacks.

Return type: Dict[str, Callback]

property param_dict#

Return the copy of parameters dictionary describing the attributes of the walkers.

Return type: fragile.core.typing.StateDict

property clone_names#

Return the set of all the attributes that are cloned when iterating the Swarm.

Return type: Set[str]

property inputs#

Return the dictionary containing all the inputs of the search algorithm.

Return type: dict

__len__()[source]#

Return the the number of walkers in the swarm.

Return type: int

__getattr__(item)[source]#

setup_state(param_dict, n_walkers=None)[source]#

Set up SwarmState instance for the search.

Parameters

param_dict (fragile.core.typing.StateDict) – Initial dictionary of parameters.
n_walkers (Optional[int]) – Number of Walker instances. If not set, uses the previous number.

Returns

None.

setup()[source]#

Prepare Swarm and internal components.

Returns: None.
Return type: None

register_callback(callback, setup=True)[source]#

When setup=True, calls setup() on the callback. Adds the Callback’s attribute dictionary to the instance’s State.

This allows the state to manage all the data the callback needs.

Parameters

callback (Callback) – An instance of Callback.
setup (bool) – Indicates whether we should call setup() on the given callback.

Returns

None.

Return type

None

get(name, default=None, raise_error=False)[source]#

Access attributes of the Swarm and its children.

Parameters

name (str) – Name of the attribute
default (Any) – Default value to return if attribute doesn’t exist.
raise_error (bool) – Raise an error if the named attribute cannot be found.

Returns

The attribute if it exists, else the default.

Return type

Any

reset(root_walker=None, state=None)[source]#

Reset a Swarm and clear the internal data to start a new search process.

Upon being called, reset clears all internal data from previous runs and sets up the Swarm to begin a new search process.

Parameters

root_walker (OneWalker, optional) – A walker representing the initial state of the search. The walkers will be reset to thiswalker and it will be added to the root of the StateTree.
state (SwarmState) – Defines the initial state of the Swarm. Defaults are loaded from the current setup given during instantiation of the Swarm() object.

Return type

None

run(root_walker=None, state=None)[source]#

Run a new search process until the stop condition is met.

This method runs a new search process by resetting the walkers to the initial state given in the root_walker argument. If not provided, it will use the previously set root walker. The initial swarm state can be defined using the state dictionary.

During each epoch of the search process, the swarm undergoes an evolution process in the environment, and the search continues until the stop condition is met. The stop condition is defined by either of the following conditions:

Maximum number of epochs are exceeded (defined in max_epochs attribute).

All the walkers are out of bounds (defined in oobs attribute of the walkers).

Any user-defined callback returns True for its evolution_end method.

After the stop condition is met, the run_end method is called and all callbacks’ run_end methods are executed.

Parameters

root_walker (Optional[StateData]) – Walker representing the initial state of the search. The walkers will be reset to this walker, and it will be added to the root of the StateTree if any.
state (Optional[StateData]) – StateData dictionary that define the initial state of the Swarm.

Returns

None.

Return type

None

evolution_end()[source]#

Check if the Swarm’s evolution process should stop.

Checks whether the maximum number of epochs has been reached, all walkers are out of bounds or any of the callbacks have ended their evolution.

Returns: A boolean value indicating whether the evolution process has to stop.
Return type: bool

evolve()[source]#

Make the walkers undergo a perturbation process in the swarm Environment.

The evolution algorithm function undergoes the following steps:

If the self.walkers_last is False then it calls the callback functions declared as ‘before_walkers’.
If the self.walkers_last is False It balances the walkers according to the clone probabilities by calling the balance function on the Walkers instance that is a component of the Swarm class.
If the self.walkers_last is False After the balancing is complete, the function calls more callback functions declared as ‘after_walkers’.
It then calls more callback functions defined under the ‘before_policy’ tag.
Makes each walker select an action by calling the act() method of the policy object.
Performs transition based on the new state created after taking the given action.
Calls more callback functions defined under the ‘after_env’ tag.
If the self.walkers_last is True runs the walkers balance process and related callbacks.

This function updates the StatesEnv and the StatesModel after each step.

Return type: None

before_reset()[source]#

“Called before resetting the search process back to its initial state.

Return type: None

after_reset()[source]#

Called after resetting the search process back to its initial state.

Return type: None

before_policy()[source]#

Called before using the policy to sample an action.

Return type: None

after_policy()[source]#

Called after using the policy to sample an action.

Return type: None

before_env()[source]#

Called before the environment has been updated.

Return type: None

after_env()[source]#

Called after the environment has been updated.

Return type: None

before_walkers()[source]#

Called before the walkers have been balanced.

Return type: None

after_walkers()[source]#

Called after the walkers have been balanced.

Return type: None

before_evolve()[source]#

Called before the evolve step has started.

Return type: None

after_evolve()[source]#

Called after the evolve step has ended.

Return type: None

run_end()[source]#

Called after the evolution process has ended.

Return type: None

fragile.core.api_classes

Contents

fragile.core.api_classes#

Module Contents#

Classes#

`fragile.core.api_classes`#