Workflows 的一个关键能力，是让我们能实现分支与循环逻辑，而且比基于图的做法更简单、更灵活。

工作流中的循环

为了构造一个循环，我们写一个会随机循环的 LoopingWorkflow。它只有一个事件，我们叫它 LoopEvent（其实可以取任意名字）。

from workflows.events import Event

class LoopEvent(Event):
    num_loops: int

现在我们 import random，并修改 step_one 函数，让它随机决定是继续循环还是往下走：

import random
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent

class LoopingWorkflow(Workflow):
    @step
    async def prepare_input(self, ev: StartEvent) -> LoopEvent:
        num_loops = random.randint(0, 10)
        return LoopEvent(num_loops=num_loops)

    @step
    async def loop_step(self, ev: LoopEvent) -> LoopEvent | StopEvent:
        if ev.num_loops <= 0:
            return StopEvent(result="Done looping!")

        return LoopEvent(num_loops=ev.num_loops-1)

把它可视化一下：

通过定义合适的事件类型与返回类型，你可以从任意步骤向任意其他步骤建立循环。

工作流中的分支

与循环紧密相关的是分支。如你所见，可以按条件返回不同的事件。我们看一个会分岔成两条路径的工作流：

import random
from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent

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 把它们同步起来。

事件子类路由

事件路由默认是精确匹配。标注为 ParentEvent 的步骤不会自动收到 ChildEvent，即使 ChildEvent 是它的子类。这样在事件模型不断增大时，不会因为意外的宽泛匹配而改变工作流。

当父事件确实就是路由约定时，按步骤显式开启：

from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent

class ToolEvent(Event):
    tool_name: str

class SearchEvent(ToolEvent):
    query: str

class ToolWorkflow(Workflow):
    @step
    async def start(self, ev: StartEvent) -> SearchEvent:
        return SearchEvent(tool_name="search", query=ev.query)

    @step(accept_event_subclasses=True)
    async def handle_tool(self, ev: ToolEvent) -> StopEvent:
        return StopEvent(result=ev.tool_name)

请谨慎使用。在多数工作流里，具体的事件类型能让分支结构更清晰，也能给出更好的校验错误。