trinity.buffer.operators.experience_operator module

trinity.buffer.operators.experience_operator module#

class trinity.buffer.operators.experience_operator.ExperienceOperator[source]#

Bases: ABC

Base class for all experience operators in the Trinity framework. Operators are used to process experiences and perform some transformations based on them.

This interface will be deprecated in the future in favor of ExperienceOperatorV1, which supports asynchronous processing and access to auxiliary models. Do not implement new operators based on this interface. Please use ExperienceOperatorV1 instead.

abstractmethod process(exps: List[Experience]) → Tuple[List[Experience], Dict][source]#

Process a list of experiences and return a transformed list.

Parameters:

exps (List[Experience]) – List of experiences to process, which contains all experiences generated by the Explorer in one explore step.

Returns:

A tuple containing the processed list of experiences and a dictionary of metrics.

Return type:

Tuple[List[Experience], Dict]

close()[source]#

Close the operator if it has any resources to release.

class trinity.buffer.operators.experience_operator.ExperienceOperatorV1[source]#

Bases: ABC

An enhanced version of ExperienceOperator that runs asynchronously and has access to auxiliary models.

set_auxiliary_model(auxiliary_models: Dict[str | int, List['AsyncOpenAI']] | None = None) → None[source]#

Set the auxiliary models for the operator.

async prepare() → None[source]#

Prepare the operator if it has any asynchronous initialization.

abstractmethod async process(exps: List[Experience]) → Tuple[List[Experience], Dict][source]#

Process a list of experiences and return a transformed list.

Parameters:

exps (List[Experience]) – List of experiences to process, which contains all experiences generated by the Explorer in one explore step.

Returns:

A tuple containing the processed list of experiences and a dictionary of metrics.

Return type:

Tuple[List[Experience], Dict]

async close()[source]#

Close the operator if it has any resources to release.

class trinity.buffer.operators.experience_operator.ExperienceOperatorV1Wrapper(operator: ExperienceOperator)[source]#

Bases: ExperienceOperatorV1

Adapt a legacy ExperienceOperator to the ExperienceOperatorV1 async interface.

__init__(operator: ExperienceOperator)[source]#
async process(exps: List[Experience]) → Tuple[List[Experience], Dict][source]#

Process a list of experiences and return a transformed list.

Parameters:

exps (List[Experience]) – List of experiences to process, which contains all experiences generated by the Explorer in one explore step.

Returns:

A tuple containing the processed list of experiences and a dictionary of metrics.

Return type:

Tuple[List[Experience], Dict]

async close()[source]#

Close the operator if it has any resources to release.

trinity.buffer.operators.experience_operator.ensure_v1_operator(operator: ExperienceOperator | ExperienceOperatorV1) → ExperienceOperatorV1[source]#

Ensure the operator exposes ExperienceOperatorV1 interface.

trinity.buffer.operators.experience_operator.create_operators(operator_configs: List[OperatorConfig], auxiliary_models: Dict[str | int, List['AsyncOpenAI']] | None = None) → List[ExperienceOperatorV1][source]#

Create a list of ExperienceOperatorV1 instances based on the provided operator configurations.

Parameters:

  • operator_configs (List[OperatorConfig]) – List of operator configurations.

  • auxiliary_models (Dict[str | int, List["AsyncOpenAI"]], optional) – A dictionary of auxiliary models that can be used by the operators. The keys are model identifiers and the values are lists of openai.AsyncOpenAI instances. Defaults to None.

Returns:

List of instantiated ExperienceOperatorV1 objects.

Return type:

List[ExperienceOperatorV1]

Contents