Workflows 可以让多个步骤同时运行。当若干步骤彼此独立、且每个都在等待某些慢操作时，并行会更快。

常见模式是扇出与扇入。你把工作拆成若干份、同时运行，再把结果汇合回来。这些直接写在步骤签名里：步骤返回一个 list 就是扇出，每个元素一个事件；步骤接收一个 list 参数就是扇入，整批只触发一次。@step 装饰器会读这些类型，校验器与可视化随后就能把每个生产者步骤和消费其事件的步骤连起来，无需你额外做什么。当你需要发出签名推不出的事件时，可以用 ctx.send_event 自己发送，本页末尾的动态 API 会讲到。

限制同时运行的工作流数量

整轮运行的并发与步骤的并发是两回事。在工作流上设置 num_concurrent_runs，可以限制同时活跃的 run() 调用数量：

workflow = ParallelFlow(num_concurrent_runs=4)

该取值必须是正整数或 None。默认为 None，即不限数量。该限制约束的是进程内的并发：额外的调用会等待，直到某次已开始的运行结束。由 DBOS 支撑的工作流也支持它，并按 worker 逐个生效。

要用 @step(num_workers=...) 来控制单次运行内部的并发。它决定该步骤最多有多少份副本可以同时处理事件，并不限制工作流运行的数量。

限制跨运行的活跃步骤数

要限制某个步骤在所有运行中最多同时跑几份（例如给外部 API 的并发调用设上限），就在它们之间共享一个 asyncio.Semaphore。步骤就是普通的 async 函数，因此不需要库的额外支持：

import asyncio

API_SLOTS = asyncio.Semaphore(3)

class Fetcher(Workflow):
    @step
    async def fetch(self, ev: FetchRequest) -> FetchResult:
        async with API_SLOTS:
            data = await call_external_api(ev.url)
        return FetchResult(data=data)

所有运行共用同一个信号量，所以任意时刻最多只有三个 fetch 步骤正处在 API 调用中，不论它们属于哪次运行。信号量存在于单个 Python 进程内；如果你部署多个进程，每个进程各有一份。

扇出：返回一个 list

步骤返回一个 list，其中每个元素都会各自触发为一个事件。这里五个 Task 会在 work 之下并发运行：

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

class Task(Event):
    n: int

class Done(Event):
    n: int

class ParallelFlow(Workflow):
    @step
    async def fan_out(self, ev: StartEvent) -> list[Task]:
        return [Task(n=i) for i in range(5)]

    @step(num_workers=5)
    async def work(self, ev: Task) -> Done:
        await asyncio.sleep(random.randint(0, 5))
        return Done(n=ev.n)

整个 list 是一个批次。num_workers=5 允许最多五份 work 同时运行。返回 [] 时什么都不触发，但步骤仍会完成、批次随即关闭。

扇入：接收一个 list

让步骤接收一个 list 参数而非单个事件，它就会收集整个批次，然后带着全部内容只触发一次：

class ConcurrentFlow(Workflow):
    @step
    async def fan_out(self, ev: StartEvent) -> list[Task]:
        return [Task(n=i) for i in range(5)]

    @step(num_workers=5)
    async def work(self, ev: Task) -> Done:
        await asyncio.sleep(random.randint(1, 5))
        return Done(n=ev.n)

    @step
    async def join(self, events: list[Done]) -> StopEvent:
        return StopEvent(result=sorted(e.n for e in events))

fan_out 返回 list[Task]，join 接收 list[Done]，因此框架仅凭类型就知道批次是什么，并在批次完成时触发 join。

list 中的事件按完成顺序排列，也就是每个 worker 先后完成的顺序，而不是 fan_out 发出它们的顺序。如果你需要固定顺序，请像这里的 join 那样自己排序（用 sorted）。

worker 可以通过返回 None 放弃自己那一支。框架会追踪哪些分支仍返回了值，因此 join 仍只触发一次，且只带上那些分支：

    @step(num_workers=5)
    async def work(self, ev: Task) -> Done | None:
        if ev.n % 2 == 0:
            return None  # drop this branch
        return Done(n=ev.n)

这里偶数编号的任务返回 None，所以 join 只收到奇数的 Done 事件。

提前释放

默认情况下，list 类型的汇聚要等待整个批次。如果想只针对第一个结果采取行动，就把参数包进 Collect。Take(n) 会在第 n 个事件到达时触发，带上最早的 n 个事件。当你只需要前几个结果、或只要最先完成的那个时，就该用它：

from typing import Annotated
from workflows.collect import Collect, Take

class FastestWins(Workflow):
    @step
    async def fan_out(self, ev: StartEvent) -> list[Task]:
        return [Task(n=i) for i in range(5)]

    @step(num_workers=5)
    async def work(self, ev: Task) -> Done:
        await asyncio.sleep(random.randint(1, 5))
        return Done(n=ev.n)

    @step
    async def first(
        self, events: Annotated[list[Done], Collect(Take(1))]
    ) -> StopEvent:
        return StopEvent(result=events[0].n)

Take(1) 在最先到来的那个事件之后即停止。其余任务仍在继续运行——没有任何东西会取消它们，只是它们再也到不了汇聚点。写成 list[Done] 的普通参数等价于 Annotated[list[Done], Collect(All())]。而不带参数的 Collect() 写的是同一个默认值，只是在代码里更容易被搜索到。

混合事件类型的扇入

批次不必只含一种事件类型。list[A | B] 会把两种类型收进一个扁平批次，步骤收到的是其中任意一种：

    @step
    async def join(
        self, events: list[StepACompleteEvent | StepBCompleteEvent]
    ) -> StopEvent:
        ...

如果你想等待的是「每种类型各一个」而不是一个 list，就给步骤每种事件一个参数。当每个参数都收到自己的事件后，步骤才会触发一次。每个参数按类型匹配：

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

class StepACompleteEvent(Event):
    result: str

class StepBCompleteEvent(Event):
    result: str

class StepCCompleteEvent(Event):
    result: str

class ConcurrentFlow(Workflow):
    @step
    async def step_a(self, ev: StartEvent) -> StepACompleteEvent:
        return StepACompleteEvent(result="Query 1")

    @step
    async def step_b(self, ev: StartEvent) -> StepBCompleteEvent:
        return StepBCompleteEvent(result="Query 2")

    @step
    async def step_c(self, ev: StartEvent) -> StepCCompleteEvent:
        return StepCCompleteEvent(result="Query 3")

    @step
    async def assemble(
        self,
        a: StepACompleteEvent,
        b: StepBCompleteEvent,
        c: StepCCompleteEvent,
    ) -> StopEvent:
        return StopEvent(result=[a.result, b.result, c.result])

step_a、step_b 与 step_c 都由同一个 StartEvent 触发，因此它们会同时运行。assemble 为每个都准备了一个参数，所以要在三个都到达后才触发。

嵌套

扇出可以嵌套。在扇出里再做扇出，就得到一个嵌套批次：内层汇聚对每个外层元素各触发一次，随后外层汇聚对所有内层结果触发一次：

class InnerTask(Event):
    outer: int
    inner: int

class InnerDone(Event):
    outer: int
    inner: int

class InnerSummary(Event):
    outer: int
    total: int

class Nested(Workflow):
    @step
    async def outer(self, ev: StartEvent) -> list[Task]:
        return [Task(n=o) for o in range(3)]

    @step
    async def inner(self, ev: Task) -> list[InnerTask]:
        return [InnerTask(outer=ev.n, inner=i) for i in range(2)]

    @step
    async def inner_work(self, ev: InnerTask) -> InnerDone:
        return InnerDone(outer=ev.outer, inner=ev.inner)

    @step
    async def per_inner(self, events: list[InnerDone]) -> InnerSummary:
        return InnerSummary(outer=events[0].outer, total=len(events))

    @step
    async def per_outer(self, events: list[InnerSummary]) -> StopEvent:
        return StopEvent(result=sorted((s.outer, s.total) for s in events))

每个汇聚都留在自己的层级上。per_inner 会运行三次（每个外层 Task 一次），per_outer 则带着三份摘要运行一次。

动态 API

当步骤需要在返回之前就发出事件时，用 ctx.send_event 自己发送，并用 ctx.collect_events 收集。适用情形包括：只按条件发出部分事件、事先无法确定要发多少个事件，或者希望下游工作在生产者仍在运行时就开始。

这里有一个取舍。list 返回是全有或全无：它要在步骤返回时才作为一个批次发出，所以步骤先抛错就什么都不会发出。而 ctx.send_event 在你调用它的那一刻就发出。下游步骤可以在生产者步骤尚未完成前就开始，但一旦已发出，即便步骤随后失败也收不回来。

ctx.send_event 一次发出一个事件：

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

class StepTwoEvent(Event):
    query: str

class ParallelFlow(Workflow):
    @step
    async def start(self, ctx: Context, ev: StartEvent) -> StepTwoEvent | None:
        ctx.send_event(StepTwoEvent(query="Query 1"))
        ctx.send_event(StepTwoEvent(query="Query 2"))
        ctx.send_event(StepTwoEvent(query="Query 3"))

    @step(num_workers=4)
    async def step_two(self, ev: StepTwoEvent) -> StopEvent:
        print("Running slow query ", ev.query)
        await asyncio.sleep(random.randint(0, 5))
        return StopEvent(result=ev.query)

start 用 ctx.send_event 发出事件，而不是把它们返回。返回类型注解里仍包含 StepTwoEvent，尽管函数返回的是 None——这样校验与图示才知道该步骤可能产生这个事件。如果你把发出的事件从签名里省掉，运行时仍然可以发送它，但静态校验与可视化就无法推断出这条边。

要在往下走之前等待多个手工发送的事件，用 ctx.collect_events：

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

class StepTwoEvent(Event):
    query: str

class StepThreeEvent(Event):
    result: str

class ConcurrentFlow(Workflow):
    @step
    async def start(self, ctx: Context, ev: StartEvent) -> StepTwoEvent | None:
        ctx.send_event(StepTwoEvent(query="Query 1"))
        ctx.send_event(StepTwoEvent(query="Query 2"))
        ctx.send_event(StepTwoEvent(query="Query 3"))

    @step(num_workers=4)
    async def step_two(self, ctx: Context, ev: StepTwoEvent) -> StepThreeEvent:
        print("Running query ", ev.query)
        await asyncio.sleep(random.randint(1, 5))
        return StepThreeEvent(result=ev.query)

    @step
    async def step_three(
        self, ctx: Context, ev: StepThreeEvent
    ) -> StopEvent | None:
        # wait until we receive 3 events
        result = ctx.collect_events(ev, [StepThreeEvent] * 3)
        if result is None:
            return None

        # do something with all 3 results together
        print(result)
        return StopEvent(result="Done")

ctx.collect_events 接收触发事件与一个待等待的类型列表。step_three 会在每个 StepThreeEvent 上运行，但 collect_events 在三个都到齐前返回 None；到齐后按到达顺序以 list 返回。要等待的事件数量得你自己记录，这里就是那个 3。

你可以等待任意类型组合，而不只是同一种类型反复出现。你传入的顺序就是它们返回的顺序，与各自何时到达无关：

    @step
    async def step_three(
        self,
        ctx: Context,
        ev: StepACompleteEvent | StepBCompleteEvent | StepCCompleteEvent,
    ) -> StopEvent | None:
        if (
            ctx.collect_events(
                ev,
                [StepCCompleteEvent, StepACompleteEvent, StepBCompleteEvent],
            )
            is None
        ):
            return None
        return StopEvent(result="Done")

让扇出具备持久性

长时间的扇出很适合做检查点：待处理事件与部分完成的扇入状态都可以被序列化，并在重启后恢复。检查点循环与完整示例见编写持久化工作流。