trinity.buffer.operators.experience_operator module

trinity.buffer.operators.experience_operator module#

class trinity.buffer.operators.experience_operator.ExperienceOperator[源代码]#

基类：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][源代码]#

Process a list of experiences and return a transformed list.

参数:

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

返回:

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

返回类型:

Tuple[List[Experience], Dict]

close()[源代码]#

Close the operator if it has any resources to release.

class trinity.buffer.operators.experience_operator.ExperienceOperatorV1[源代码]#

基类：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[源代码]#

Set the auxiliary models for the operator.

async prepare() → None[源代码]#

Prepare the operator if it has any asynchronous initialization.

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

Process a list of experiences and return a transformed list.

参数:

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

返回:

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

返回类型:

Tuple[List[Experience], Dict]

async close()[源代码]#

Close the operator if it has any resources to release.

class trinity.buffer.operators.experience_operator.ExperienceOperatorV1Wrapper(operator: ExperienceOperator)[源代码]#

基类：ExperienceOperatorV1

Adapt a legacy ExperienceOperator to the ExperienceOperatorV1 async interface.

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

Process a list of experiences and return a transformed list.

参数:

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

返回:

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

返回类型:

Tuple[List[Experience], Dict]

async close()[源代码]#

Close the operator if it has any resources to release.

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

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][源代码]#

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

参数:

  • 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.

返回:

List of instantiated ExperienceOperatorV1 objects.

返回类型:

List[ExperienceOperatorV1]

目录