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.
- format_args: FormatConfig
- rollout_args: GenerationConfig
- is_eval: bool = False
- 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:
- 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
- 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
- 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]
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.
- format_args: FormatConfig
- rollout_args: GenerationConfig
- is_eval: bool = False
- 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:
- 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
- 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]
- 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
- 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]
- 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]
- generate_env_inference_samples(env, rollout_num) List[Experience] [source]
- run() List[Experience] [source]
Run workflow and return a list of experiences.