FlowMind

What	Declarative agent flows in YAML, with checkpointing and resume
Why	Production agents need to survive process restarts and look diff-able in code review
Status	🟢 Active — used internally on a couple of pipelines
Stack	Python 3.9+, PyYAML, Jinja2

TL;DR

pip install -r requirements.txt
pip install -e .

flowmind run examples/hello.yaml

Define a flow as data:

name: hello
entry: greet

nodes:
  - id: greet
    step: set
    args:
      greeting: "hello"
      who: "world"
    next: speak

  - id: speak
    step: print
    args:
      msg: "{{ greeting }}, {{ who }}!"

Run it:

from flowmind import load_flow, Runtime

flow = load_flow("examples/hello.yaml")
state = Runtime(flow, checkpoint_dir="state/").run()
print(state["greeting"], state["who"])

Why YAML, not Python?

Three reasons, all production-flavoured:

1. Diffs. A non-technical reviewer can read a YAML pull request and tell you that step 7 changed. They can't read a Python DSL.
2. Hot-swap. The flow file is configuration. Change it without redeploying the worker.
3. Single source of truth. The flow doubles as the spec and the documentation.

That said — every step is still Python. You're not stuck with the built-ins.

Built-in steps

Step	What it does
set	Write literal values to state
print	Print a Jinja-rendered message (debugging)
noop	Terminal placeholder
http_get	GET a URL, parse JSON, store under store_as
python	eval a tiny expression — sandboxed (__builtins__ cleared)

Register your own:

from flowmind import Step, register

@register("call_llm")
class CallLLM(Step):
    def run(self, state):
        from openai import OpenAI
        resp = OpenAI().chat.completions.create(
            model=self.kwargs["model"],
            messages=[{"role": "user", "content": state["prompt"]}],
        )
        return {"answer": resp.choices[0].message.content}

Branches

Linear flows are boring. Branches are first-class:

branches:
  - id: classify
    cases:
      - when: "score >= 80"
        next: high_path
      - when: "score >= 50"
        next: mid_path
    default: low_path

when is any Jinja expression with the full state in scope.

Checkpointing

Pass checkpoint_dir= to the Runtime. After every step it writes {flow_name}.state.json. Resume from the last completed node:

state = State.load("state/hello.state.json")
Runtime(flow, state=state, checkpoint_dir="state/").resume(
    from_node=state["_last_node"]
)

This is the killer feature when an agent's tool call costs real money and the worker died half-way through.

Error handling

A node can declare a fallback:

- id: try_fetch
  step: http_get
  args:
    url: "https://api.example.com/v1/thing"
  on_error: fallback_local
  next: parse

If the step raises, the runtime jumps to on_error instead of crashing. The error and its origin land in state.history().

What's intentionally missing

• No DAG fan-out / fan-in. Each step has one successor. If you need a DAG, you probably want a real workflow engine (Prefect, Temporal, Airflow). FlowMind is for the 90% of agent flows that are basically a state machine.
• No retry loops. Use on_error to route to a retry node yourself.
• No async. Steps are synchronous. Easier to reason about, especially with checkpoints.

License

Apache-2.0.