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)

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

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)

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)

property resettable

reset(task: Task)

Reset the workflow.

abstract run() → List[Experience]

Run workflow and return a list of experiences.

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

Bases: Workflow

The base workflow class for multi-turn tasks.

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

abstract run() → List[Experience]

Run workflow and return a list of experiences.

process_messages_to_experience(messages, reward, info={}) → Experience

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

Bases: Workflow

A workflow for simple single-round task.

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

property resettable

reset(task: Task)

Reset the workflow.

format_messages()

run() → List[Experience]

Run workflow and return a list of experiences.

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

Bases: SimpleWorkflow

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

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

reset(task: Task)

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)

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

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)

Bases: Workflow

A workflow for simple single-round task.

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

property resettable

reset(task: Task)

Reset the workflow.

format_messages()

run() → List[Experience]

Run workflow and return a list of experiences.

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

Bases: SimpleWorkflow

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

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

reset(task: Task)

Reset the workflow.

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

Bases: MultiTurnWorkflow

A workflow for webshop task.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List | None = None)

property resettable

reset(task: Task)

Reset the workflow.

get_model_response(messages)

get_model_response_text(messages)

generate_env_inference_samples(env, session_id, rollout_num) → List[Experience]

run() → List[Experience]

Run workflow and return a list of experiences.

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

Bases: MultiTurnWorkflow

A workflow for alfworld task.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List | None = None)

get_model_response(messages)

get_model_response_text(messages)

generate_env_inference_samples(env, rollout_num) → List[Experience]

run() → List[Experience]

Run workflow and return a list of experiences.

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

Bases: MultiTurnWorkflow

A workflow for sciworld task.

__init__(model: ModelWrapper, task: Task, auxiliary_models: List | None = None)

get_model_response(messages)

get_model_response_text(messages)

generate_env_inference_samples(env, rollout_num) → List[Experience]

run() → List[Experience]

Run workflow and return a list of experiences.