trinity.common.workflows

Submodules

trinity.common.workflows.customized_math_workflows module

We include the customized math workflows in this file.

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

Bases: SimpleWorkflow

A workflow for math tasks that give answers in boxed format.

reset(task: Task)[source]: Reset the workflow.

format_prompt()[source]

run() → List[Experience][source]: Run workflow and return a list of experiences.

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 = <factory>, rollout_args: ~trinity.common.config.GenerationConfig = <factory>, workflow_args: dict = <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

workflow_args: dict

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

to_dict() → dict[source]

__init__(workflow: ~typing.Type[~trinity.common.workflows.workflow.Workflow], format_args: ~trinity.common.config.FormatConfig = <factory>, rollout_args: ~trinity.common.config.GenerationConfig = <factory>, workflow_args: dict = <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 = <factory>, rollout_args: ~trinity.common.config.GenerationConfig = <factory>, workflow_args: dict = <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

workflow_args: dict

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

to_dict() → dict[source]

__init__(workflow: ~typing.Type[~trinity.common.workflows.workflow.Workflow], format_args: ~trinity.common.config.FormatConfig = <factory>, rollout_args: ~trinity.common.config.GenerationConfig = <factory>, workflow_args: dict = <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(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.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.

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

Bases: SimpleWorkflow

A workflow for math tasks that give answers in boxed format.

reset(task: Task)[source]: Reset the workflow.

format_prompt()[source]

run() → List[Experience][source]: Run workflow and return a list of experiences.