esflow

esflow

esflow is a lightweight Python DAG workflow framework. Declare nodes using directory conventions,@flow + edge() organize DAGs,Runner.run() output event streams, and support parallel replicas, pause for collaboration, resumption at a specific point, fan-out, and fallback chains.

For full usage, see [reference.md](reference.md). This file only keeps the key points that must be clear before use.

Requirements modeling

Do not write a Node directly. First compress the user's requirements into a minimal DAG design:

1. Inputs: what the user must provide, what can be given default values, what needs pass_check
2. Artifacts: what the final output files are, which paths or fields the stdout envelope must expose
3. Nodes: each node does only one thing, named with business actions, not step1 / process
4. Edges: express only real data dependencies; tasks with no dependencies let the DAG run in parallel in the same round
5. Pause points: for human confirmation use TO_HUMAN, for Agent writing artifacts use TO_AGENT
6. Reuse points: when a rerun at a specific point is needed, clarify which upstream artifacts should be --from / --resume reused
7. Fallback chains: for multiple candidate solutions use same-layer serial, do not pile up ifs in one Node

Before starting, first give:输入 -> DAG -> 产物. When this design is unclear, ask the user first, do not guess.

Core mental model

1. DAG topology execution: node run(ctx) returns an artifact, downstream uses ctx.get("upstream_id") to read it; nodes ready in the same round run in parallel
2. Pause / resume：checkpoint=TO_HUMAN waits for human confirmation,TO_AGENT waits for an external agent to write an artifact;--out + --from / --resume reuses upstream
3. Fan-out / fallback：replicas static replicas,FanOut dynamic fan-out,serial controls same-layer fallback order

Getting started

pip install esflow
esflow new my_skill
python my_skill/scripts/run.py --out ./runs/a

Minimal acceptance:runs/a/<node>/ has business artifacts,runs/a/.esflow/<node>/artifact.json has structured artifact.

Standard structure:

my_skill/
  SKILL.md
  scripts/
    flow.py      # 声明 nodes / edges / replicas / dynamic / serial
    run.py       # 入口，集成 pass_check 和 esflow_event
    nodes/       # 一个文件一个 Node，id 非空唯一

Regular development only changes three places:flow.py declare the DAG,nodes/ add a Node,run.py add business prechecks.

Entry suggestion

New skills use run_flow_script(...) to write the entry by default, automatically getting --out / --resume behavior. Only keep handwritten Runner.load(...) entry when customizing the stdout envelope, complex cleanup, or secondary TO_AGENT orchestration.

from pathlib import Path
import argparse
import asyncio

from esflow import run_flow_script


def build_parser():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input")
    parser.add_argument("--format", default="md")
    return parser


def build_node_args(args):
    if not args.resume and not args.input:
        raise SystemExit("--input required on first run")
    return {
        "parse": {"input": args.input},
        "export": {"format": args.format},
    }


if __name__ == "__main__":
    raise SystemExit(asyncio.run(run_flow_script(
        Path(__file__).parent,
        parser=build_parser(),
        node_args_builder=build_node_args,
    )))

Runner.load(node_args=...) First run writes .esflow/node_args.json。--resume By default inherits the first run's input parameters; only same-named fields passed this time override them. Do not pass the complete node_argsconstructed from parser default values in the resume branch. Business parameters that need resume should not be written as required positional parameters; instead use optional parameters and validate them only on the first run.

TO_AGENT minimal mode

from esflow import Node, Checkpoint


class AgentWrite(Node):
    id = "agent_write"
    checkpoint = Checkpoint.TO_AGENT

    def deliver(self, artifact) -> bool:
        return "summary.txt" in artifact.get("files", [])

python3 scripts/run.py --out ./runs/a   # 退出码 2,stderr 提示 node_dir
echo "done" > ./runs/a/agent_write/summary.txt
python3 scripts/run.py --resume ./runs/a

Key rules

• run(ctx) must return an artifact, usually dict
• Large files are written to self.output_dir, artifact only stores path and metadata
• accept=False is a legal skip, downstream continues;deliver=False is an artifact error, the process fails
• ctx.get("id") takes a single upstream;ctx.gather("base") Only accepts replicas with the same base
• self.depth is the topological depth, the entry node is 0; nodes at the same layer have the same depth, and static/dynamic replicas inherit the base depth
• For cross-base fallback chains use ctx.upstream_ids() or ctx.layer(self.depth), do not use gather
• replicas is used for the number of replicas known at load time;FanOut is used for the dynamic number of replicas at runtime
• serial is the scheduling order within the same layer, not the number of replicas
• --from / --from-depth must be paired with --out；--resume is only used to continue running after the TO_AGENT artifact is written, and automatically inherits the first run's node_args
• After the first run passes --out ./runs/a or business parameters,--resume ./runs/a does not need to be passed again; only pass the same-named node_args
• nodes/*.py at the top level you can import esflow and the standard library; third-party libraries, external SDKs, and heavy business dependencies go into run or the precheck function, otherwise the precheck cannot be triggered
• esflow view and esflow debug are both interactive web services, not text topology output; they are not suitable to rely on when there is no GUI or remote SSH without port forwarding

Common commands

esflow view ./my_skill/scripts
esflow run ./my_skill/scripts --out ./runs/a
esflow run ./my_skill/scripts --out ./runs/a --from translate
esflow run ./my_skill/scripts --out ./runs/a --from-depth 2
esflow run --resume ./runs/a
esflow debug ./my_skill/scripts
esflow debug ./my_skill/scripts --node worker#2
esflow debug ./my_skill/scripts --clear

When you need specific code syntax, TO_HUMAN / TO_AGENT, static parallelism, dynamic fan-out, fallback chains, prechecks, template consistency, and debugging restrictions, read [reference.md](reference.md)。