持久化工作流一页展示了如何靠手工的 context 快照让工作流熬过重启与错误。llama-agents-dbos 包则把一个由 DBOS 支撑的运行时插进你的工作流，省掉这些手工活。每次步骤流转都会被自动持久化，因此崩溃的工作流会正好从中断处恢复——不需要任何快照代码。

安装

终端窗口

pip install llama-agents-dbos

快速上手 —— 独立的持久化工作流

使用 DBOS 最简单的方式是配合 SQLite（零外部依赖）。照常定义好工作流，传入一个 DBOSRuntime，你的状态就会被自动持久化。

import asyncio

from dbos import DBOS
from llama_agents.dbos import DBOSRuntime
from pydantic import Field
from workflows import Context, Workflow, step
from workflows.events import Event, StartEvent, StopEvent

# 1. Configure DBOS — SQLite by default
DBOS(config={"name": "counter-example", "run_admin_server": False})

# 2. Define events and workflow (nothing DBOS-specific here)
class Tick(Event):
    count: int = Field(description="Current count")

class CounterResult(StopEvent):
    final_count: int = Field(description="Final counter value")

class CounterWorkflow(Workflow):
    @step
    async def start(self, ctx: Context, ev: StartEvent) -> Tick:
        await ctx.store.set("count", 0)
        print("[Start] Initializing counter to 0")
        return Tick(count=0)

    @step
    async def increment(self, ctx: Context, ev: Tick) -> Tick | CounterResult:
        count = ev.count + 1
        await ctx.store.set("count", count)
        print(f"[Tick {count:2d}] count = {count}")

        if count >= 20:
            return CounterResult(final_count=count)

        await asyncio.sleep(0.5)
        return Tick(count=count)

# 3. Create runtime, attach to workflow, and launch
runtime = DBOSRuntime()
workflow = CounterWorkflow(runtime=runtime)

async def main() -> None:
    await runtime.launch()
    result = await workflow.run(run_id="counter-run-1")
    print(f"Result: final_count = {result.final_count}")

asyncio.run(main())

如果你在运行中途杀掉进程（比如在第 8 拍按 Ctrl+C），再次调用 workflow.run(run_id="counter-run-1") 会从第 8 拍恢复，而不是从零重新开始。

跨 run 调用持久化	✅
跨进程重启持久化	✅
能在运行时错误中存活	✅

持久化工作流服务端

DBOSRuntime 与 WorkflowServer 集成，于是你对外提供的每个工作流都开箱即用地获得持久执行。该运行时同时提供持久化存储与服务端运行时：

import asyncio

from dbos import DBOS
from llama_agents.dbos import DBOSRuntime
from llama_agents.server import WorkflowServer
from pydantic import Field
from workflows import Context, Workflow, step
from workflows.events import Event, StartEvent, StopEvent

DBOS(config={"name": "quickstart", "run_admin_server": False})

class Tick(Event):
    count: int = Field(description="Current count")

class CounterResult(StopEvent):
    final_count: int = Field(description="Final counter value")

class CounterWorkflow(Workflow):
    """Counts to 5, emitting stream events along the way."""

    @step
    async def start(self, ctx: Context, ev: StartEvent) -> Tick:
        return Tick(count=0)

    @step
    async def tick(self, ctx: Context, ev: Tick) -> Tick | CounterResult:
        count = ev.count + 1
        ctx.write_event_to_stream(Tick(count=count))
        print(f"  tick {count}")
        await asyncio.sleep(0.5)
        if count >= 5:
            return CounterResult(final_count=count)
        return Tick(count=count)

async def main() -> None:
    runtime = DBOSRuntime()

    server = WorkflowServer(
        workflow_store=runtime.create_workflow_store(),
        runtime=runtime.build_server_runtime(),
    )
    server.add_workflow("counter", CounterWorkflow(runtime=runtime))

    print("Serving on http://localhost:8000")
    print("Try: curl -X POST http://localhost:8000/workflows/counter/run")
    await server.start()
    try:
        await server.serve(host="0.0.0.0", port=8000)
    finally:
        await server.stop()

asyncio.run(main())

http://localhost:8000/ 上的工作流调试器 UI 与默认运行时下完全一样 —— 在服务层看来 DBOS 是透明的。

空闲释放

长时间运行的工作流可能在等待外部输入（人在环路、webhook 等）期间在内存里空闲很久。idle_timeout 参数告诉 DBOS 运行时把空闲的工作流从内存中释放，并在有新事件到达时自动恢复它们：

import asyncio

from dbos import DBOS
from llama_agents.dbos import DBOSRuntime
from llama_agents.server import WorkflowServer
from pydantic import Field
from workflows import Context, Workflow, step
from workflows.events import (
    HumanResponseEvent,
    InputRequiredEvent,
    StartEvent,
    StopEvent,
)

DBOS(config={"name": "idle-release-demo", "run_admin_server": False})

class AskName(InputRequiredEvent):
    prompt: str = Field(default="What is your name?")

class UserInput(HumanResponseEvent):
    response: str = Field(default="")

class GreeterWorkflow(Workflow):
    @step
    async def ask(self, ctx: Context, ev: StartEvent) -> AskName:
        return AskName()

    @step
    async def greet(self, ctx: Context, ev: UserInput) -> StopEvent:
        return StopEvent(result={"greeting": f"Hello, {ev.response}!"})

async def main() -> None:
    runtime = DBOSRuntime()

    server = WorkflowServer(
        workflow_store=runtime.create_workflow_store(),
        # Release workflows after 30 seconds of inactivity
        runtime=runtime.build_server_runtime(idle_timeout=30.0),
    )
    server.add_workflow("greeter", GreeterWorkflow(runtime=runtime))

    await server.start()
    try:
        await server.serve(host="0.0.0.0", port=8000)
    finally:
        await server.stop()

asyncio.run(main())

当 greeter 工作流发出 AskName 且 30 秒内没有输入到达时，运行时就把它从内存中释放。一旦有 UserInput 事件被发送（通过 POST /events/{handler_id}），运行时便会透明地从数据库恢复该工作流并投递这个事件。调用方完全不会察觉工作流曾被释放。

用 Postgres 做多副本部署

SQLite 很适合单进程场景。若生产部署需要多个服务副本，请改用 Postgres。每个副本必须有唯一的 executor_id：

from dbos import DBOS

DBOS(config={
    "name": "my-app",
    "system_database_url": "postgresql://user:pass@localhost:5432/mydb",
    "run_admin_server": False,
    "executor_id": "replica-1",  # unique per replica
})

完整的多副本演示见 examples/dbos/server_replicas.py 示例。想更深入地了解副本之间如何协同，见 DBOS 架构概览。

对于生产级的多副本部署，DBOS Conductor 会在核心运行时之上加上自动扩缩容与监控面板。

执行模型

理解 DBOS 的执行模型，能帮你写出在重启与副本之间行为正确的工作流。

工作流标识

工作流的名字就是它的持久身份。日志条目、步骤注册以及准入队列都以它为 key，因此 DBOS 为该工作流记录的一切都归档在这个名字之下。默认是带模块限定的类名（例如 my_app.CounterWorkflow），这意味着移动或重命名类会在你毫无察觉的情况下改变身份。对于任何长生命周期的东西，请显式设置名字并把它当作不可变更：

wf = CounterWorkflow(runtime=runtime, workflow_name="counter-v1")

worker 只会在它注册的那些名字下查找已记录的工作。改名之后，以旧名字归档的进行中与排队中的运行对新部署是不可见的 —— 请让继续注册旧名字的 worker 一直运行到那批工作完成。

使用服务端时，传给 add_workflow 的名字是 HTTP 路由名，与工作流的持久身份互不影响：

server.add_workflow("counter", CounterWorkflow(runtime=runtime, workflow_name="counter-v1"))

副本归属

每个副本由其 executor_id 标识，并拥有它启动的每一个工作流。一个工作流及其全部步骤都在同一进程中运行 —— 不会把单个步骤分发到不同副本。这意味着你的步骤可以安全依赖本地状态，例如内存缓存、本地文件或进程级单例。代价是单个工作流的工作量无法摊到多个副本上。

日志与重放

步骤的完成与流式事件都会写入数据库日志。当工作流在崩溃或空闲释放后恢复时，运行时会重放日志以重建工作流的 Context 与 store，然后从最后记录的步骤继续。

由于恢复是基于重放的，步骤可能被执行多次（如果它在日志条目提交之前被打断）。请尽量把步骤设计成幂等的，或者用 context store 在步骤内部跟踪进度（如持久化工作流一页所示）。

扩缩容与排空

副本 ID 与副本数量必须保持稳定。如果你缩容并移除了某个副本，该副本所拥有的工作流会被搁置，直到那个 executor_id 重新出现。移除副本之前，请先排空它：让其在飞的运行完成，并不再向其路由新工作。

DBOS Conductor 会自动处理这件事 —— 它通过心跳检测已排空或超时的副本，并把它们在飞的运行重新分配给健康的副本。

代码变更与版本

由于恢复依赖日志重放，在历史运行仍在进行时修改工作流代码可能带来不确定性 —— 例如某个步骤现在接受的事件集合与该运行当初启动时不同。为避免这一点：

• 在部署代码变更之前先排空在飞的运行，或者
• 把更新后的工作流注册在新名字下（例如 workflow_name="counter-v2"），这样旧运行继续沿用原代码，新运行使用更新后的版本。这是一次有意的身份变更 —— 名字到底为哪些东西做 key，见工作流标识。

运行并发上限

Workflow(num_concurrent_runs=N) 会把该工作流的活跃运行限制为每个副本 N 个，因此部署容量大致是 N 乘以副本数。超出上限的运行会进入 DBOS 队列等待，并在名额释放后约一秒内启动。该队列跨副本共享：排队中的运行与提交它的副本没有绑定关系，任何有空闲名额的副本都能接手。不设该值时，运行仍会直接启动，路径中没有队列。

wf = CounterWorkflow(runtime=runtime, num_concurrent_runs=4)

队列以工作流的持久身份为 key（见工作流标识），所以改名之后，以旧名字排队的运行就看不见了。修改或取消上限本身是安全的：等待中的运行仍留在同一队列并继续被准入。新上限不会计入在此之前已启动的运行，因此在那些运行结束之前，副本可能会短暂超出该限制。

等待中的运行在启动之前无法被取消，因为取消是一条投递给正在运行的工作流的消息。该请求会被保存下来，运行一旦被准入就会自行停止。

DBOS 通常会自动照看每一个队列。如果你的应用改为向 DBOS.listen_queues 显式传入一个列表，就必须把本运行时的队列加进去（runtime.workflow_queues）—— 它们是在注册完工作流之后、启动之前收集到的。

事件流行为

在进程内（不经服务端）使用 handler.stream_events() 时，DBOS 的流每次调用都会从头重放。也就是说你会收到该工作流历史上发出的全部事件，而不只是新事件。

工作流服务端则改用基于游标的方式 —— 它的 GET /events/{handler_id} 端点会跟踪位置，因此每个消费者只会收到一次事件。

崩溃恢复

副本重启时，DBOS 会自动检测并重新拉起属于其 executor_id 的所有未完成工作流。无需人工干预 —— 副本通过重放自己的日志从中断处继续。