run-llama/workflows-py
1
0
Fork 0
mirror of https://github.com/run-llama/workflows-py.git synced 2026-07-19 08:46:21 -04:00
Code Issues 18 Packages Projects Releases 10 Wiki Activity

[PR #130] [CLOSED] Support durable execution with DBOS #147

New Issue
Closed
opened 2026-02-16 02:16:44 -05:00 by yindo · 0 comments
yindo commented 2026-02-16 02:16:44 -05:00
Owner
Copy Link
Copy Source

📋 Pull Request Information

Original PR: https://github.com/run-llama/workflows-py/pull/130
Author: @qianl15
Created: 10/8/2025
Status: ❌ Closed

Base: main ← Head: qian/dbos-workflow

📝 Commits (10+)

📊 Changes

7 files changed (+1111 additions, -2 deletions)

View changed files

📝 pyproject.toml (+1 -0)
➕ src/workflows/dbos/__init__.py (+7 -0)
➕ src/workflows/dbos/dbos_context.py (+422 -0)
➕ src/workflows/dbos/dbos_workflow.py (+235 -0)
➕ tests/dbos/__init__.py (+0 -0)
➕ tests/dbos/test_dbos.py (+111 -0)
📝 uv.lock (+335 -2)

📄 Description

Summary

This PR integrates DBOS with workflow and context to provide out-of-the-box durable execution and checkpointing.

Changes

  • DBOSWorkflow:

    • It subclasses Workflow to start a DBOS durable workflow for each step worker, and use durable notification for sending the StartEvent.
    • Once the _done step finishes, it broadcasts a message to all running DBOS workflows to signal the end. This cleanly terminates running tasks.

  • DBOSContext:

    • It subclasses Context to provide durable execution, making each step worker's main loop a DBOS workflow (including the cancel worker). It also uses DBOS.recv to durably receive incoming events for each step.
    • Because DBOS requires each workflow to be uniquely and statically defined, each DBOSContext needs to have a unique name and be created in a static function. This is important for DBOS to correctly find the definition of workflows for failure recovery.
    • For each user defined LlamaIndex step, it's automatically within the DBOS workflow. Therefore, if the step needs to call any external API or non-deterministic functions, users should wrap them in DBOS.step.
    • ctx.send_event and ctx.send_event_async use DBOS durable send instead of in-memory queues for communicating events between steps.
    • Use DBOS.sleep and _durable_time to make sure determinism within a step workflow.
@DBOS.step()
async def _durable_time() -> float:
    return time.time()

Example

Here is a simple example to use DBOS.

import asyncio
from pydantic import BaseModel, Field
from workflows import Context, Workflow, step
from workflows.events import Event, StartEvent, StopEvent
from dbos import DBOS, DBOSConfig

class MyEvent(Event):
    msg: list[str]

class RunState(BaseModel):
    num_runs: int = Field(default=0)

class MyWorkflow(DBOSWorkflow):
    @step
    async def start(self, ctx: Context[RunState], ev: StartEvent) -> MyEvent:
        async with ctx.store.edit_state() as state:
            state.num_runs += 1

            return MyEvent(msg=[ev.input_msg] * state.num_runs)

    @step
    async def process(self, ctx: Context[RunState], ev: MyEvent) -> StopEvent:
        data_length = len("".join(ev.msg))
        new_msg = f"Processed {len(ev.msg)} times, data length: {data_length}"
        return StopEvent(result=new_msg)

async def main():
    config: DBOSConfig = {
        "name": "dbos-llamaindex",
        "system_database_url": "sqlite:///dbos_llamaindex.sqlite",
    }
    DBOS(config=config)
    DBOS.launch()
    workflow = MyWorkflow()

    # For failure recovery, DBOS context should be defined statically (not in a dynamic function) with a unique name so that the recovery process can find it.
    ctx = DBOSContext(workflow, "run_my_workflow")
    result = await workflow.run(input_msg="Hello, world!", ctx=ctx)
    print("Workflow result:", result)


if __name__ == "__main__":
    asyncio.run(main())

Discussion

  • Potentially change the way of the _done step behavior -- it currently throws a WorkflowDone exception. Though it works, its workflow generates error stack trace that can be annoying. An alternative way is to use durable events to signal the end of the workflow.
  • How does it interact with LlamaIndex workflow server? Maybe the top level Workflow.run should also become a DBOS workflow, and each step is a sub-workflow. However, currently, _run_workflow uses asyncio.wait which is not deterministic (a fundamental requirement for durable workflows). This may require changes to how it waits for the workflow to finish.
  • How to integrate with StateStore. Currently it's in-memory and users need to access it from DBOS.step functions.

🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.

## 📋 Pull Request Information **Original PR:** https://github.com/run-llama/workflows-py/pull/130 **Author:** [@qianl15](https://github.com/qianl15) **Created:** 10/8/2025 **Status:** ❌ Closed **Base:** `main` ← **Head:** `qian/dbos-workflow` --- ### 📝 Commits (10+) - [`c451999`](https://github.com/run-llama/workflows-py/commit/c4519990aac6c2466f7a0d58da02acce1333e0d2) init DBOS - [`52da5f5`](https://github.com/run-llama/workflows-py/commit/52da5f5599d87593aa4070117ea5daf98e8acea5) Merge branch 'run-llama:main' into qian/dbos-workflow - [`55a5cc4`](https://github.com/run-llama/workflows-py/commit/55a5cc4c9f25141ca4ef58f8ce67b749d988e407) Start DBOSWorkflow - [`77eeb15`](https://github.com/run-llama/workflows-py/commit/77eeb1522ef2c5c5226ae3607ea9fa7dedbad7bf) DBOS workflow initial pass done - [`5052a75`](https://github.com/run-llama/workflows-py/commit/5052a75c1d8cb9166f2c1360fc03e9e78ef64111) nit - [`f1087a3`](https://github.com/run-llama/workflows-py/commit/f1087a311ce0f645c6928a392bbfb6c9e88e02ed) Merge branch 'run-llama:main' into qian/dbos-workflow - [`b778f0c`](https://github.com/run-llama/workflows-py/commit/b778f0cd058839c2f3430b9e5f5391e7b8cbef20) durable send event - [`81ebff5`](https://github.com/run-llama/workflows-py/commit/81ebff58499c9cbd370c4486e0167ee4d25a4a0b) clean up test - [`3200617`](https://github.com/run-llama/workflows-py/commit/32006179541722c33cc632fa74a6f58bd212800d) Simplify - [`3e8c5f1`](https://github.com/run-llama/workflows-py/commit/3e8c5f1896f43bff91ed55be7a9d484fdfc9b1a5) Wrap time.time() in a step ### 📊 Changes **7 files changed** (+1111 additions, -2 deletions) <details> <summary>View changed files</summary> 📝 `pyproject.toml` (+1 -0) ➕ `src/workflows/dbos/__init__.py` (+7 -0) ➕ `src/workflows/dbos/dbos_context.py` (+422 -0) ➕ `src/workflows/dbos/dbos_workflow.py` (+235 -0) ➕ `tests/dbos/__init__.py` (+0 -0) ➕ `tests/dbos/test_dbos.py` (+111 -0) 📝 `uv.lock` (+335 -2) </details> ### 📄 Description ### Summary This PR integrates [DBOS](https://github.com/dbos-inc/dbos-transact-py) with workflow and context to provide out-of-the-box durable execution and checkpointing. ### Changes * **DBOSWorkflow:** * It subclasses `Workflow` to start a DBOS durable workflow for each step worker, and use durable notification for sending the `StartEvent`. * Once the `_done` step finishes, it broadcasts a message to all running DBOS workflows to signal the end. This cleanly terminates running tasks. * **DBOSContext:** * It subclasses `Context` to provide durable execution, making each step worker's main loop a DBOS workflow (including the cancel worker). It also uses `DBOS.recv` to durably receive incoming events for each step. * Because DBOS requires each workflow to be uniquely and statically defined, each `DBOSContext` needs to have a unique name and be created in a static function. This is important for DBOS to correctly find the definition of workflows for failure recovery. * For each user defined LlamaIndex step, it's automatically within the DBOS workflow. Therefore, if the step needs to call any external API or non-deterministic functions, users should wrap them in `DBOS.step`. * `ctx.send_event` and `ctx.send_event_async` use DBOS durable send instead of in-memory queues for communicating events between steps. * Use `DBOS.sleep` and `_durable_time` to make sure determinism within a step workflow. ```python @DBOS.step() async def _durable_time() -> float: return time.time() ``` ### Example Here is a simple example to use DBOS. ```python import asyncio from pydantic import BaseModel, Field from workflows import Context, Workflow, step from workflows.events import Event, StartEvent, StopEvent from dbos import DBOS, DBOSConfig class MyEvent(Event): msg: list[str] class RunState(BaseModel): num_runs: int = Field(default=0) class MyWorkflow(DBOSWorkflow): @step async def start(self, ctx: Context[RunState], ev: StartEvent) -> MyEvent: async with ctx.store.edit_state() as state: state.num_runs += 1 return MyEvent(msg=[ev.input_msg] * state.num_runs) @step async def process(self, ctx: Context[RunState], ev: MyEvent) -> StopEvent: data_length = len("".join(ev.msg)) new_msg = f"Processed {len(ev.msg)} times, data length: {data_length}" return StopEvent(result=new_msg) async def main(): config: DBOSConfig = { "name": "dbos-llamaindex", "system_database_url": "sqlite:///dbos_llamaindex.sqlite", } DBOS(config=config) DBOS.launch() workflow = MyWorkflow() # For failure recovery, DBOS context should be defined statically (not in a dynamic function) with a unique name so that the recovery process can find it. ctx = DBOSContext(workflow, "run_my_workflow") result = await workflow.run(input_msg="Hello, world!", ctx=ctx) print("Workflow result:", result) if __name__ == "__main__": asyncio.run(main()) ``` ### Discussion * Potentially change the way of the `_done` step behavior -- it currently throws a `WorkflowDone` exception. Though it works, its workflow generates error stack trace that can be annoying. An alternative way is to use durable events to signal the end of the workflow. * How does it interact with LlamaIndex workflow server? Maybe the top level `Workflow.run` should also become a DBOS workflow, and each step is a sub-workflow. However, currently, `_run_workflow` uses `asyncio.wait` which is not deterministic (a fundamental requirement for durable workflows). This may require changes to how it waits for the workflow to finish. * How to integrate with StateStore. Currently it's in-memory and users need to access it from `DBOS.step` functions. --- <sub>🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.</sub>
yindo added the pull-request label 2026-02-16 02:16:44 -05:00
yindo closed this issue 2026-02-16 02:16:44 -05:00
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
bug
enhancement
enhancement
good first issue
pull-request
Mirrored from GitHub Pull Request
 question
No Label pull-request
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: run-llama/workflows-py#147
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?