某个步骤执行失败可能导致整个工作流失败，但很多时候错误是预期之中的，执行可以安全重试。想想因网络瞬时拥塞而超时的 HTTP 请求，或触碰到限流的对外 API 调用。

在你希望步骤再试一次的所有这些情形里，可以使用重试策略。重试策略指示工作流把一个步骤执行多次，控制每次重试前等待多久、哪些错误可以重试、以及何时放弃。

重试会针对同一个输入事件重新执行整个步骤体。请把失败调用之前的副作用保持幂等，或者把它们挪到可重试的工作之后。如果步骤在抛出异常前已写过状态或调用过外部服务，那些操作可能会被再做一遍。

重试模块由三类可组合的构件组成：

• 重试条件决定某个异常是否可重试。
• 等待策略决定下一次尝试前休眠多久。
• 停止条件决定工作流何时放弃。

用 retry_policy(retry=..., wait=..., stop=...) 把它们组合成策略，再把结果交给 @step 装饰器。重试条件支持 | 与 &，等待策略支持 +，停止条件支持 | 与 &。

基本用法

from workflows import Workflow, Context, step
from workflows.events import StartEvent, StopEvent
from workflows.retry_policy import (
    retry_policy,
    stop_after_attempt,
    wait_fixed,
)

class MyWorkflow(Workflow):
    @step(
        retry_policy=retry_policy(
            wait=wait_fixed(5),
            stop=stop_after_attempt(10),
        )
    )
    async def flaky_step(self, ctx: Context, ev: StartEvent) -> StopEvent:
        result = flaky_call()  # this might raise
        return StopEvent(result=result)

不带任何参数时，retry_policy() 会对所有异常重试、最多 3 次尝试，每次之间固定等待 5 秒。传入 retry=、wait= 或 stop= 可以定制任意一个部件。

过滤异常

用重试条件控制哪些异常会被重试：

from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent
from workflows.retry_policy import (
    retry_policy,
    retry_if_exception_message,
    retry_if_exception_type,
    stop_after_attempt,
    stop_before_delay,
    wait_fixed,
    wait_random,
)

class MyWorkflow(Workflow):
    @step(
        retry_policy=retry_policy(
            retry=retry_if_exception_type((TimeoutError, ConnectionError))
            | retry_if_exception_message(match="rate limit|temporarily unavailable"),
            wait=wait_fixed(1) + wait_random(0, 1),
            stop=stop_after_attempt(5) | stop_before_delay(30),
        )
    )
    async def call_provider(self, ev: StartEvent) -> StopEvent:
        result = await flaky_call()
        return StopEvent(result=result)

如果你更喜欢显式写法，那些具名组合器等价于：

from workflows.retry_policy import (
    retry_policy,
    retry_any,
    retry_if_exception_message,
    retry_if_exception_type,
    stop_any,
    stop_after_attempt,
    stop_before_delay,
    wait_combine,
    wait_fixed,
    wait_random,
)

policy = retry_policy(
    retry=retry_any(
        retry_if_exception_type((TimeoutError, ConnectionError)),
        retry_if_exception_message(match="rate limit|temporarily unavailable"),
    ),
    wait=wait_combine(wait_fixed(1), wait_random(0, 1)),
    stop=stop_any(stop_after_attempt(5), stop_before_delay(30)),
)

指数退避

对于调用 LLM 服务商或受限流 API 的步骤，指数退避可以避免惊群效应：

from workflows import Workflow, Context, step
from workflows.events import StartEvent, StopEvent
from workflows.retry_policy import (
    retry_policy,
    stop_after_attempt,
    wait_exponential_jitter,
)

class MyWorkflow(Workflow):
    @step(
        retry_policy=retry_policy(
            wait=wait_exponential_jitter(initial=1, exp_base=2, max=30, jitter=1),
            stop=stop_after_attempt(5),
        )
    )
    async def call_llm(self, ctx: Context, ev: StartEvent) -> StopEvent:
        result = await llm_call()  # this might raise on rate-limit
        return StopEvent(result=result)

完整 API 参考

该模块提供的重试条件、等待策略与停止条件远多于此处示例。完整的构件清单及其参数见 API 参考。

自定义重试策略

如果可组合 API 覆盖不了你的场景，你可以自己写一个策略。唯一要求是一个类，其 next 方法符合 RetryPolicy 协议：

def next(
    self,
    elapsed_time: float,
    attempts: int,
    error: Exception,
    *,
    seed: int | None = None,
) -> float | None:
    ...

返回重试前应等待的秒数，或返回 None 表示停止。attempts 统计迄今为止的失败次数，第一次失败时为 1。可选的 seed 供持久化运行时使用，以便在重放时让抖动是确定性的。

例如，下面这个策略只在周五重试：

from datetime import datetime

class RetryOnFridayPolicy:
    def next(
        self,
        elapsed_time: float,
        attempts: int,
        error: Exception,
        *,
        seed: int | None = None,
    ) -> float | None:
        if datetime.today().strftime("%A") == "Friday":
            return 5  # retry in 5 seconds
        return None  # don't retry

已弃用的便捷构造器

注意

ConstantDelayRetryPolicy 与 ExponentialBackoffRetryPolicy 早于可组合 API，仅为向后兼容而保留。请优先使用 retry_policy(...) 并显式给出 retry、wait 与 stop 参数。

from workflows.retry_policy import ConstantDelayRetryPolicy

# Deprecated, equivalent to:
#   retry_policy(wait=wait_fixed(5), stop=stop_after_attempt(10))
policy = ConstantDelayRetryPolicy(delay=5, maximum_attempts=10)

from workflows.retry_policy import ExponentialBackoffRetryPolicy

# Deprecated, equivalent to:
#   retry_policy(wait=wait_random_exponential(multiplier=1, exp_base=2, max=30),
#                stop=stop_after_attempt(5))
policy = ExponentialBackoffRetryPolicy(
    initial_delay=1, multiplier=2, max_delay=30, maximum_attempts=5,
)

在步骤内查看重试状态

Context.retry_info() 返回一个描述当前尝试的 RetryInfo。可以用它在不同重试之间调整行为：放宽检索范围、缩短超时、记录一条警告：

from workflows import Context, Workflow, step
from workflows.events import StartEvent, StopEvent
from workflows.retry_policy import retry_policy, stop_after_attempt

class MyWorkflow(Workflow):
    @step(retry_policy=retry_policy(stop=stop_after_attempt(3)))
    async def flaky(self, ctx: Context, ev: StartEvent) -> StopEvent:
        info = ctx.retry_info()
        if info.last_exception is not None:
            logger.warning(
                "retry %d after %s", info.retry_number, str(info.last_exception)
            )
        ...

retry_number 在首次运行时为 0。last_exception 与 last_failed_at 在第一次失败前为 None，之后描述最近一次失败。

重试耗尽后的恢复

当策略放弃时，异常会向外传播并使工作流失败，除非声明了 @catch_error 处理器。处理器是一个接收 StepFailedEvent 的步骤：它可以返回一个 StopEvent 优雅结束运行，也可以返回其他事件以重新路由工作流，或者抛出异常以新的错误结束运行。请检查 ev.step_name 与 ev.exception 来决定怎么做；字段全集见 StepFailedEvent API 参考。

from workflows import Context, Workflow, catch_error, step
from workflows.events import StartEvent, StepFailedEvent, StopEvent
from workflows.retry_policy import retry_policy, stop_after_attempt, wait_fixed

class Pipeline(Workflow):
    @step(retry_policy=retry_policy(wait=wait_fixed(1), stop=stop_after_attempt(3)))
    async def fetch(self, ev: StartEvent) -> StopEvent:
        return StopEvent(result=await call_api())

    @catch_error
    async def on_failure(
        self, ctx: Context, ev: StepFailedEvent
    ) -> StopEvent:
        return StopEvent(
            result={"failed_step": ev.step_name, "error": str(ev.exception)}
        )

不带参数的 @catch_error 是一个通配符。它会捕获任何重试已耗尽、且未被更具体处理器接管步骤。每个工作流只允许一个通配符。

限定到具体步骤

传入 for_steps=[...] 可以限定处理器覆盖哪些步骤。一个步骤名最多只能出现在一个限定处理器中；不存在的名字会在构造时被拒绝。

class Pipeline(Workflow):
    @step(retry_policy=...)
    async def fetch(self, ev: StartEvent) -> FetchedEvent: ...

    @step(retry_policy=...)
    async def parse(self, ev: FetchedEvent) -> StopEvent: ...

    @catch_error(for_steps=["fetch"])
    async def on_fetch_failure(
        self, ctx: Context, ev: StepFailedEvent
    ) -> StopEvent:
        return StopEvent(result={"fallback": True})

    @catch_error  # wildcard; covers `parse` and anything else
    async def on_any_failure(
        self, ctx: Context, ev: StepFailedEvent
    ) -> StopEvent:
        return StopEvent(result={"aborted": ev.step_name})

恢复预算

处理器自身也可能发出重新流入图中事件，因此同一条血统可能再次进入同一个处理器。max_recoveries（默认 1）限制一条事件血统最多可以被再次进入几次，超出就直接让工作流失败。对于参与有界重试循环的处理器可以调高它；对于终态处理器请保持 1。

@catch_error(for_steps=["fetch", "retry_fetch"], max_recoveries=2)
async def on_failure(self, ctx: Context, ev: StepFailedEvent) -> RetryFetch:
    return RetryFetch()