trinity.common.workflows.step_wise_workflow module

class trinity.common.workflows.step_wise_workflow.StepWiseRewardWorkflow(*, task: Task, model: ModelWrapper, auxiliary_models=None, use_openai_client=True)

基类：Workflow

A workflow that implements step-wise rewards for tasks.

__init__(*, task: Task, model: ModelWrapper, auxiliary_models=None, use_openai_client=True)

run() → list[Experience]

Run the workflow and return a list of experiences with step-wise rewards.

step(step_num: int) → bool

Run a single step of your agent application.

参数:

step_num (int) -- The current step number.

返回:

Whether to continue running the agent application.

返回类型:

bool

Tips:

You can use the openai client (self.client) to migrate your existing applications at low cost.

reward(exps: list[Experience], step_num: int) → float

Calculate the reward for the given experiences at the specified step.

property max_step_num

Return the maximum number of steps in the task.

class trinity.common.workflows.step_wise_workflow.AsyncStepWiseRewardWorkflow(*, task: Task, model: ModelWrapper, auxiliary_models=None, use_openai_client=True)

基类：StepWiseRewardWorkflow

Async version of StepWiseRewardWorkflow.

is_async: bool = True

async run_async() → list[Experience]

Run the workflow and return a list of experiences with step-wise rewards asynchronously.

async step_async(step_num: int) → bool

Run a single step of your agent application asynchronously.

参数:

step_num (int) -- The current step number.

返回:

Whether to continue running the agent application.

返回类型:

bool

Tips:

You can use the openai client (self.client) to migrate your existing applications at low cost.

async reward_async(exps: list[Experience], step_num: int) → float

Calculate the reward for the given experiences at the specified step asynchronously.

class trinity.common.workflows.step_wise_workflow.RewardPropagationWorkflow(*, task: Task, model: ModelWrapper, auxiliary_models=None, use_openai_client=True)

基类：Workflow

A workflow that propagates rewards across multiple turns.

__init__(*, task: Task, model: ModelWrapper, auxiliary_models=None, use_openai_client=True)

run() → list[Experience]

Run the workflow and return a list of experiences with step-wise rewards.

step(step_num: int) → bool

Run a single step of your agent application.

参数:

step_num (int) -- The current step number.

返回:

Whether to continue running the agent application.

返回类型:

bool

Tips:

You can use the openai client (self.client) to migrate your existing applications at low cost.

reward(exps: list[Experience]) → float

Calculate the reward for the given experiences of the entire run.

property max_step_num

Return the maximum number of steps in the task.

class trinity.common.workflows.step_wise_workflow.AsyncRewardPropagationWorkflow(*, task: Task, model: ModelWrapper, auxiliary_models=None, use_openai_client=True)

基类：RewardPropagationWorkflow

Async version of RewardPropagationWorkflow.

is_async: bool = True

async run_async() → list[Experience]

Run the workflow and return a list of experiences with step-wise rewards asynchronously.

async step_async(step_num: int) → bool

Run a single step of your agent application asynchronously.

参数:

step_num (int) -- The current step number.

返回:

Whether to continue running the agent application.

返回类型:

bool

Tips:

You can use the openai client (self.client) to migrate your existing applications at low cost.

async reward_async(exps: list[Experience]) → float

Calculate the reward for the given experiences of the entire run asynchronously.