trinity.buffer.operators package

Subpackages

• trinity.buffer.operators.filters package
  • Submodules
    • trinity.buffer.operators.filters.reward_filter module
      • RewardFilter
      • RewardSTDFilter
      • InvalidRewardFilter
  • Module contents
• trinity.buffer.operators.mappers package
  • Submodules
    • trinity.buffer.operators.mappers.pass_rate_calculator module
      • PassRateCalculator
    • trinity.buffer.operators.mappers.reward_shaping_mapper module
      • RewardShapingMapper
  • Module contents

Submodules

• trinity.buffer.operators.data_juicer_operator module
  • DataJuicerOperator
    • DataJuicerOperator.__init__()
    • DataJuicerOperator.process()
    • DataJuicerOperator.close()
• trinity.buffer.operators.experience_operator module
  • ExperienceOperator
    • ExperienceOperator.process()
    • ExperienceOperator.close()
  • ExperienceOperatorV1
    • ExperienceOperatorV1.set_auxiliary_model()
    • ExperienceOperatorV1.prepare()
    • ExperienceOperatorV1.process()
    • ExperienceOperatorV1.close()
  • ExperienceOperatorV1Wrapper
    • ExperienceOperatorV1Wrapper.__init__()
    • ExperienceOperatorV1Wrapper.process()
    • ExperienceOperatorV1Wrapper.close()
  • ensure_v1_operator()
  • create_operators()

Module contents

class trinity.buffer.operators.ExperienceOperator

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.

close()

Close the operator if it has any resources to release.

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

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]

class trinity.buffer.operators.ExperienceOperatorV1

Bases: ABC

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

async close()

Close the operator if it has any resources to release.

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.

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]

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

Set the auxiliary models for the operator.

trinity.buffer.operators.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.

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]