每次工作流运行都有一个 Context，每个 context 都有一个状态存储。用它来存放步骤在一次运行中需要共享的值，或者存放你有意通过复用/恢复同一 context 带到后续运行中的值。

状态存储不是放重量级客户端、索引、文件句柄等运行时依赖的地方。那些请放进 resources。状态应当是你愿意在对工作流做快照或恢复时序列化的数据。

默认情况下，工作流初始化的是一个无类型的状态存储。你可以通过 ctx.store 读写它：

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

class MyWorkflow(Workflow):

    @step
    async def my_step(self, ctx: Context, ev: StartEvent) -> StopEvent:
        current_count = await ctx.store.get("count", default=0)
        current_count += 1
        await ctx.store.set("count", current_count)
        return StopEvent(result=current_count)

给状态加锁

有些情况下，状态会被同时运行的多个步骤修改。这时请用 edit_state() 锁定状态，使这次更新是原子的：

@step
async def my_step(self, ctx: Context, ev: StartEvent) -> StopEvent:
    async with ctx.store.edit_state() as state:
        current_count = state.get("count", 0)
        state["count"] = current_count + 1
        result = state["count"]
    return StopEvent(result=result)

代码块执行期间，其他步骤无法编辑状态。请让这个块尽量小：把读取-修改-写入的工作放在块内，把耗时的 LLM 或网络调用放在块外。

加入带类型的状态

通常你的工作流状态有已知的形状。这时用 Pydantic 模型来承载，你会得到：

• 获得状态的类型提示
• 获得状态的自动校验
• （可选）借助校验器与序列化器完全掌控状态的序列化与反序列化

**注意：**Pydantic 模型要为所有字段提供默认值。这样 Context 就能自动初始化状态。

下面是一个简短示例，展示如何结合 workflows + pydantic 用上这些能力：

from pydantic import BaseModel, Field

class CounterState(BaseModel):
    count: int = Field(default=0)

然后，只需用这个状态模型标注你的工作流状态：

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

class MyWorkflow(Workflow):
    @step
    async def start(
        self,
        ctx: Context[CounterState], ev: StartEvent
    ) -> StopEvent:
        # Allows for atomic state updates
        async with ctx.store.edit_state() as state:
            state.count += 1

        return StopEvent(result="Done!")

你也可以一次只处理带类型状态中的一个字段：

@step
async def start(self, ctx: Context[CounterState], ev: StartEvent) -> StopEvent:
    current = await ctx.store.get("count")
    await ctx.store.set("count", current + 1)
    state = await ctx.store.get_state()
    return StopEvent(result=state.count)

跨运行保持上下文

如果你想在同一工作流的多次运行之间保持状态，就创建一个 context 并把同一个传给 .run()：

workflow = MyWorkflow()
ctx = Context(workflow)

handler = workflow.run(ctx=ctx)
result = await handler

# Optional: save the ctx somewhere and restore
# ctx_dict = ctx.to_dict()
# ctx = Context.from_dict(workflow, ctx_dict)

# continue with next run
handler = workflow.run(ctx=ctx)
result = await handler

如果该 context 仍在运行，run(ctx=ctx) 会恢复那次运行、不会发送新的 StartEvent。如果上一次运行已经完成，run(ctx=ctx) 会用相同的存储状态开始一次新的运行。

可序列化的状态

存放在这里的状态会在对运行做快照时被序列化，以实现持久化，所以只放 JSON 序列化器能编码的值。客户端及其他不可序列化对象请放进 resources。

对于自定义值，要么让它们可被 Pydantic 序列化，要么在调用 Context.to_dict() 与 Context.from_dict() 时提供自定义序列化器。

在工作流上设置 serializer= 可选择其内部的 context 与检查点编码，或在 WorkflowServer 上设置服务端默认值。工作流级别的覆盖优先级更高。Appserver 在托管工作流时会保留已配置的序列化器。该设置不会改变 HTTP API 的 JSON 格式。