pavoz

🇨🇳 简体中文

The in-process Python workflow engine — durable, replayable, forkable pipelines with zero runtime dependencies.

No server. No scheduler. No YAML. No vendor lock-in. Define a DAG with @dag.stage, run it in your own process, and get per-stage checkpoints, resume, replay, and time-travel forks for free.

Python 3.12+  |  MIT License  |  stdlib only  |  200+ tests  |  v0.3.0-rc1

Why another workflow engine?

You've written a pipeline like fetch → transform → review → save before. You know the pain:

• Flow control (retries, timeouts, state) tangled up with business logic — one big function that's scary to touch
• A crash at step 4 means the expensive steps 1–3 run again
• Debugging an LLM stage means re-running the whole pipeline — you can't see the prompt that stage actually got, or change it and re-run just that stage
• Regression-testing a stage change means mocking everything by hand

pavoz pulls graph execution out of your business code and gives you durable, replayable runs — while staying out of your way:

• In-process, zero runtime deps — it's a library, not a platform. No cluster to self-host, no server to start, no DB required (bring your own persistence via a 5-method StorageBackend adapter when you want resume across restarts).
• Never makes business decisions for you — model selection, audit loops, and prompt templates are your code. External calls go through a ctx.call(kind, op, params) seam you wire to your LLM/search/HTTP stack; nothing is hard-coded to any vendor.
• Every run is debuggable — each stage's exact input is reconstructible from the checkpoint. Replay one stage after editing a prompt, or fork a past run from any stage with edited input (see time-travel debugging).

30-second quickstart

# not on PyPI yet — install from git:
pip install pavoz
# once 0.3.0 is released this collapses to:  pip install pavoz

# pipeline.py
import asyncio
from pavoz import DAG, Runtime, CheckpointStore, FileStorage

dag = DAG("demo")

@dag.stage()
async def s_fetch(ctx):
    return {"items": ["a", "b", "c"]}

@dag.stage(depends_on=["s_fetch"], retries=2, timeout=60)
async def s_process(ctx):
    return {"count": len(ctx.state["items"]), "upper": [i.upper() for i in ctx.state["items"]]}

@dag.stage(depends_on=["s_process"])
async def s_save(ctx):
    print("saved:", ctx.state["count"], ctx.state["upper"])   # → saved: 3 ['A', 'B', 'C']
    return {"done": True}

async def main():
    rt = Runtime(checkpoint_store=CheckpointStore(FileStorage("./data")))
    result = await rt.run(dag, task_id="job-1")
    print(result.status)          # → done

asyncio.run(main())

That's the whole API. State flows as a plain JSON-serializable dict; ctx.state is read-only, and each stage returns its delta. That's it — no base classes, no context-injection magic, no framework telling you how to structure code.

Or run it from the CLI (same checkpoint store, so every command below composes):

pavoz run pipeline.py --task-id job-1
pavoz trace  --task-id job-1            # stage statuses + run_id
pavoz state  --task-id job-1 --key count

What you get

Time-travel debugging

This is the feature we built pavoz for: any past run is an object you can inspect, edit, and fork. Because every checkpoint stores each stage's raw delta (not just the merged end state), the exact input any stage saw can be rebuilt:

# 1. What did s_compose actually see? (full input JSON, human-editable)
pavoz export-input --task-id job-1 --stage s_compose > input.json

# 2. Tweak it, fork the run from that stage: upstream stages are reused (not re-run),
#    s_compose and its successors re-run on the edited input. Original run untouched.
pavoz fork-run pipeline.py --task-id job-1 --stage s_compose --input input.json

Or in code:

result = await rt.run(dag, task_id="job-1")                 # original run
result = await rt.fork_run(dag, task_id="job-1",            # fork: reuse s_fetch+s_process
                           from_stage="s_compose",          # rerun s_compose → on
                           overrides={"items": ["x", "y"]})

Single-stage replay for prompt-tuning iterations (no fork, no new run):

pavoz replay pipeline.py --task-id job-1 --stage s_compose     # rerun one stage on its rebuilt input

Each fork gets a new run_id and becomes the task's latest run; the original checkpoint stays intact — multiple forks of the same history coexist. If your LLM stage misbehaves, you're seconds away from "what did it see → change one sentence → re-run just that stage" instead of a 30-minute pipeline rerun.

Declarative param override

Since v0.3.0 fork-and-rerun no longer means hand-editing JSON — the fork input is a one-liner: point paths, automatic type inference, and a state diff against the original run:

# Switch the writer model of one stage and rerun it — upstream stages are reused.
pavoz fork-run pipeline.py --task-id job-1 --stage s_compose \
  --set llm.model=longcat --set config.target_chars=8000 \
  --compare-with <original_run_id>

# Preview what the override would apply (no execution):
pavoz fork-run pipeline.py --task-id job-1 --stage s_compose --set llm.model=x --dry-run

• --set KEY=VALUE — repeatable dot-path overrides (42 → int, 0.7 → float, true → bool), deep-merged: sibling keys are preserved instead of wiped

• --set-file overrides.yaml — bulk JSON/YAML (safe_load, 1 MB cap); pip install pavoz[yaml]

• --compare-with RUN_ID — leaf-level state diff against another run, printed after the fork

• --dry-run — preview the apply effect on the target run's pre-stage state (leaf diff + would_rerun), no execution

• PAVOZ_STORAGE_SPEC / PAVOZ_STORAGE_KWARGS — point the CLI at any StorageBackend (e.g. your MinIO adapter); {task_id} interpolated Library equivalent: parse_set_args / parse_set_file / merge_overrides / apply_overrides in pavoz.state. Full guide: docs/replay-params.md · runnable scripts in examples/.

When to use pavoz — and when not

Use it for sequential/stateful pipelines in a single process: LLM research & writing pipelines, agent steps with durable state, ETL with per-step checkpoints, anything where stage boundaries are known ahead of time and you want retries + resume + replay without adopting a platform.

Don't use it for:

• Distributed scheduling at scale — cron, a fleet of workers, multi-tenant queues: use Temporal / Prefect / Airflow, which own that layer. pavoz deliberately doesn't.
• Dynamic graphs — if your graph shape must change at runtime (agents that spawn agents, data-driven branching): LangGraph / Burr own that space. pavoz graphs are static; dynamic control flow lives inside a stage as ordinary Python.
• A web UI / observability platform — see Temporal, Hatchet, Windmill. pavoz gives you CLI + JSON checkpoints to build on.

How it compares

Known limitations

• Checkpoints store stage deltas, not full state (v0.8): full state is rebuilt from initial_state + stage_deltas on load. If a stage returns very large payloads (e.g. full document content) into shared state, checkpoints grow with them — keep big blobs in external storage and pass references/summaries through state. Checkpoint.state_stats() helps you see what is actually large.
• resume=True continues the latest run only: it refuses to resume a run that already finished (done guard) — rerun with resume=False (new run id), or use fork_run to branch from a historical stage with edited inputs.
• One process, one run at a time: pavoz is an in-process engine with no built-in queue, scheduler, or multi-worker coordination. Lease/heartbeat on the task row is your integration's job (see the reference integration).

Use cases

• LLM content pipelines — search → download → filter → compress → compose → audit → save, with per-stage checkpointing so a failed audit never repeats 40 minutes of upstream work. (reference integration)
• Agent steps with durable state — long-running multi-step agents where each step's output must survive process restarts.
• Any multi-stage Python pipeline that outgrows a script but doesn't need a platform.

Extensions

Extensions are plain decorators over stage functions — the framework itself needs no changes. The official pack, pavoz-extensions, ships @gate (worker → multi-lens review → score → revise loop) and @schema (cross-stage contract validation); examples/agent_loop.py walks the pattern end to end. Third parties are welcome to publish their own pavoz-* packages — core never depends on them.

Extensions pin a compatible core range (e.g. pavoz>=0.3,<0.4); during 0.x, minor releases may contain breaking changes (details).

Docs & development

• Architecture · API reference · Quickstart · Storage backends
• Contributing · Releasing · Changelog · MIT License

pavoz — graph execution, extracted. Everything else stays yours.