trinity.common.workflows

Submodules

trinity.common.workflows.workflow module

Base Workflow Class

class trinity.common.workflows.workflow.Task(workflow: ~typing.Type[~trinity.common.workflows.workflow.Workflow], format_args: ~trinity.common.config.FormatConfig, rollout_args: ~trinity.common.config.GenerationConfig = <factory>, is_eval: bool = False, reward_fn: ~typing.Type[~trinity.common.rewards.reward_fn.RewardFn] | None = None, raw_task: dict | None = None)[source]

Bases: object

A Task class that defines a task and its associated reward function / workflow.

workflow: Type[Workflow]
format_args: FormatConfig
rollout_args: GenerationConfig
is_eval: bool = False
reward_fn: Type[RewardFn] | None = None
raw_task: dict | None = None
to_workflow(model: Any, auxiliary_models: List[OpenAI] | None = None) Workflow[source]

Convert the task to a workflow.

Parameters:
  • model (ModelWrapper) – The rollout model for the workflow.

  • auxiliary_models (List[openai.OpenAI]) – The auxiliary models for the workflow.

Note

model_path attribute is added to the auxiliary_models for use within the workflow.

Returns:

The generated workflow object.

Return type:

Workflow

property task_desc: str | None
property truth: str | None
__init__(workflow: ~typing.Type[~trinity.common.workflows.workflow.Workflow], format_args: ~trinity.common.config.FormatConfig, rollout_args: ~trinity.common.config.GenerationConfig = <factory>, is_eval: bool = False, reward_fn: ~typing.Type[~trinity.common.rewards.reward_fn.RewardFn] | None = None, raw_task: dict | None = None) None
class trinity.common.workflows.workflow.Workflow(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]

Bases: ABC

The base workflow class.

A workflow is a runnable object which generates a list of experiences.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]
property resettable
reset(task: Task)[source]

Reset the workflow.

abstract run() List[Experience][source]

Run workflow and return a list of experiences.

class trinity.common.workflows.workflow.MultiTurnWorkflow(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]

Bases: Workflow

The base workflow class for multi-turn tasks.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]
abstract run() List[Experience][source]

Run workflow and return a list of experiences.

process_messages_to_experience(messages, reward, info={}) Experience[source]
class trinity.common.workflows.workflow.SimpleWorkflow(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]

Bases: Workflow

A workflow for simple single-round task.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]
property resettable
reset(task: Task)[source]

Reset the workflow.

format_messages()[source]
run() List[Experience][source]

Run workflow and return a list of experiences.

class trinity.common.workflows.workflow.MathWorkflow(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]

Bases: SimpleWorkflow

A workflow for math tasks as introduced in DeepSeek-R1.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]
reset(task: Task)[source]

Reset the workflow.

Module contents

Workflow module

class trinity.common.workflows.Task(workflow: ~typing.Type[~trinity.common.workflows.workflow.Workflow], format_args: ~trinity.common.config.FormatConfig, rollout_args: ~trinity.common.config.GenerationConfig = <factory>, is_eval: bool = False, reward_fn: ~typing.Type[~trinity.common.rewards.reward_fn.RewardFn] | None = None, raw_task: dict | None = None)[source]

Bases: object

A Task class that defines a task and its associated reward function / workflow.

workflow: Type[Workflow]
format_args: FormatConfig
rollout_args: GenerationConfig
is_eval: bool = False
reward_fn: Type[RewardFn] | None = None
raw_task: dict | None = None
to_workflow(model: Any, auxiliary_models: List[OpenAI] | None = None) Workflow[source]

Convert the task to a workflow.

Parameters:
  • model (ModelWrapper) – The rollout model for the workflow.

  • auxiliary_models (List[openai.OpenAI]) – The auxiliary models for the workflow.

Note

model_path attribute is added to the auxiliary_models for use within the workflow.

Returns:

The generated workflow object.

Return type:

Workflow

property task_desc: str | None
property truth: str | None
__init__(workflow: ~typing.Type[~trinity.common.workflows.workflow.Workflow], format_args: ~trinity.common.config.FormatConfig, rollout_args: ~trinity.common.config.GenerationConfig = <factory>, is_eval: bool = False, reward_fn: ~typing.Type[~trinity.common.rewards.reward_fn.RewardFn] | None = None, raw_task: dict | None = None) None
class trinity.common.workflows.SimpleWorkflow(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]

Bases: Workflow

A workflow for simple single-round task.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]
property resettable
reset(task: Task)[source]

Reset the workflow.

format_messages()[source]
run() List[Experience][source]

Run workflow and return a list of experiences.

class trinity.common.workflows.MathWorkflow(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]

Bases: SimpleWorkflow

A workflow for math tasks as introduced in DeepSeek-R1.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List[OpenAI] | None = None)[source]
reset(task: Task)[source]

Reset the workflow.

class trinity.common.workflows.WebShopWorkflow(model: ModelWrapper, task: Task, auxiliary_models: List | None = None)[source]

Bases: MultiTurnWorkflow

A workflow for webshop task.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List | None = None)[source]
property resettable
reset(task: Task)[source]

Reset the workflow.

get_model_response(messages)[source]
get_model_response_text(messages)[source]
generate_env_inference_samples(env, session_id, rollout_num) List[Experience][source]
run() List[Experience][source]

Run workflow and return a list of experiences.

class trinity.common.workflows.AlfworldWorkflow(model: ModelWrapper, task: Task, auxiliary_models: List | None = None)[source]

Bases: MultiTurnWorkflow

A workflow for alfworld task.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List | None = None)[source]
get_model_response(messages)[source]
get_model_response_text(messages)[source]
generate_env_inference_samples(env, rollout_num) List[Experience][source]
run() List[Experience][source]

Run workflow and return a list of experiences.

class trinity.common.workflows.SciWorldWorkflow(model: ModelWrapper, task: Task, auxiliary_models: List | None = None)[source]

Bases: MultiTurnWorkflow

A workflow for sciworld task.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List | None = None)[source]
get_model_response(messages)[source]
get_model_response_text(messages)[source]
generate_env_inference_samples(env, rollout_num) List[Experience][source]
run() List[Experience][source]

Run workflow and return a list of experiences.