分支和循环

工作流的一个关键特性是它们能够实现分支和循环逻辑，比基于图的方法更简单、更灵活。

工作流中的循环

要创建一个循环，我们将使用前一个教程中的示例 `MyWorkflow` 并添加一个新的自定义事件类型。我们将它命名为 `LoopEvent`，但同样它可以是任何任意名称。

class LoopEvent(Event):
    loop_output: str

现在我们将 `import random` 并修改 `step_one` 函数，使其随机决定是循环还是继续。

@step
async def step_one(self, ev: StartEvent | LoopEvent) -> FirstEvent | LoopEvent:
    if random.randint(0, 1) == 0:
        print("Bad thing happened")
        return LoopEvent(loop_output="Back to step one.")
    else:
        print("Good thing happened")
        return FirstEvent(first_output="First step complete.")

让我们可视化这个

通过定义适当的事件类型和返回类型，您可以从任何一个步骤创建到任何其他步骤的循环。

工作流中的分支

与循环密切相关的是分支。正如您已经看到的，您可以有条件地返回不同的事件。让我们看一个分支到两个不同路径的工作流。

class BranchA1Event(Event):
    payload: str


class BranchA2Event(Event):
    payload: str


class BranchB1Event(Event):
    payload: str


class BranchB2Event(Event):
    payload: str


class BranchWorkflow(Workflow):
    @step
    async def start(self, ev: StartEvent) -> BranchA1Event | BranchB1Event:
        if random.randint(0, 1) == 0:
            print("Go to branch A")
            return BranchA1Event(payload="Branch A")
        else:
            print("Go to branch B")
            return BranchB1Event(payload="Branch B")

    @step
    async def step_a1(self, ev: BranchA1Event) -> BranchA2Event:
        print(ev.payload)
        return BranchA2Event(payload=ev.payload)

    @step
    async def step_b1(self, ev: BranchB1Event) -> BranchB2Event:
        print(ev.payload)
        return BranchB2Event(payload=ev.payload)

    @step
    async def step_a2(self, ev: BranchA2Event) -> StopEvent:
        print(ev.payload)
        return StopEvent(result="Branch A complete.")

    @step
    async def step_b2(self, ev: BranchB2Event) -> StopEvent:
        print(ev.payload)
        return StopEvent(result="Branch B complete.")

我们的导入与之前相同，但我们创建了 4 个新的事件类型。`start` 随机决定选择一个分支或另一个，然后每个分支中的多个步骤完成工作流。让我们可视化这个。

您当然可以按任意顺序组合分支和循环，以满足您的应用程序需求。在本教程的后面部分，您将学习如何使用 `send_event` 并行运行多个分支，并使用 `collect_events` 同步它们。

接下来我们将学习使用 Context 维护状态。