How Three Workflow Engines Run: BPM Token Engine, Dify Graph Scheduler, Agent Loop

Traditional BPM Engine: Token Machine

BPMN 2.0 defines a fixed set of graphical elements (start/end events, tasks, gateways, sequence flows). A minimal "leave request" process in BPMN XML looks like this:

<process id="leaveRequest">
  <startEvent id="start"/>
  <sequenceFlow sourceRef="start" targetRef="checkAmount"/>
  <exclusiveGateway id="checkAmount"/>
  <sequenceFlow sourceRef="checkAmount" targetRef="managerApproval">
    <conditionExpression>${amount > 1000}</conditionExpression>
  </sequenceFlow>
  <sequenceFlow sourceRef="checkAmount" targetRef="autoApprove">
    <conditionExpression>${amount <= 1000}</conditionExpression>
  </sequenceFlow>
  <userTask id="managerApproval"/>
  <sequenceFlow sourceRef="managerApproval" targetRef="end"/>
  <serviceTask id="autoApprove"/>
  <sequenceFlow sourceRef="autoApprove" targetRef="end"/>
  <endEvent id="end"/>
</process>

The XML is a static definition; the engine parses it into an in‑memory graph and creates one or more execution objects (tokens). The engine repeatedly asks "where is the token now, and which edge should it follow?".

Task node : executes its business logic and moves the token forward.

Exclusive gateway : evaluates conditions on outgoing edges and follows the first true branch.

Parallel gateway : forks N child tokens, each traverses a branch independently; a join waits until all child tokens arrive before merging back into a single token.

The core challenge of a BPM engine is handling long‑running, persisted states, which forces the execution unit to be a serializable token rather than an in‑memory function call.

A minimal token‑machine implementation that supports sequential tasks, exclusive gateways, parallel forks/joins, and end events:

from dataclasses import dataclass, field
from enum import Enum

class NodeType(Enum):
    TASK = "task"
    USER_TASK = "user_task"
    EXCLUSIVE_GATEWAY = "exclusive_gateway"
    PARALLEL_FORK = "parallel_fork"
    PARALLEL_JOIN = "parallel_join"
    END = "end"

@dataclass
class Edge:
    target: str
    condition: str | None = None

@dataclass
class Node:
    id: str
    type: NodeType
    behavior = None  # callable for automatic tasks
    join_size: int = 0

@dataclass
class Token:
    instance_id: str
    node_id: str
    join_arrived: dict = field(default_factory=dict)

class ProcessEngine:
    def __init__(self, nodes: dict[str, Node], edges: dict[str, list[Edge]]):
        self.nodes = nodes
        self.edges = edges
        self.storage: dict[str, dict] = {}

    def start(self, instance_id: str, start_node: str, variables: dict):
        self.storage[instance_id] = {"variables": variables, "tokens": []}
        token = Token(instance_id, start_node)
        self._advance(token)

    def _advance(self, token: Token):
        node = self.nodes[token.node_id]
        vars = self.storage[token.instance_id]["variables"]
        if node.type == NodeType.TASK:
            if node.behavior:
                node.behavior(vars)
            self._leave(token)
        elif node.type == NodeType.USER_TASK:
            self.storage[token.instance_id]["tokens"].append(token)
            print(f"[Waiting for human] instance={token.instance_id} node={node.id}")
        elif node.type == NodeType.EXCLUSIVE_GATEWAY:
            for edge in self.edges[node.id]:
                if edge.condition is None or eval(edge.condition, {}, vars):
                    token.node_id = edge.target
                    self._advance(token)
                    return
        elif node.type == NodeType.PARALLEL_FORK:
            for edge in self.edges[node.id]:
                child = Token(token.instance_id, edge.target)
                self._advance(child)
        elif node.type == NodeType.PARALLEL_JOIN:
            arrived = token.join_arrived.get(node.id, 0) + 1
            token.join_arrived[node.id] = arrived
            if arrived < node.join_size:
                self.storage[token.instance_id]["tokens"].append(token)
                return
            self._leave(token)
        elif node.type == NodeType.END:
            print(f"[Process finished] instance={token.instance_id}")

    def _leave(self, token: Token):
        outs = self.edges.get(token.node_id, [])
        if outs:
            token.node_id = outs[0]["target"]
            self._advance(token)

    def complete_user_task(self, instance_id: str, node_id: str, updates: dict):
        self.storage[instance_id]["variables"].update(updates)
        tokens = self.storage[instance_id]["tokens"]
        token = next(t for t in tokens if t.node_id == node_id)
        tokens.remove(token)
        self._leave(token)

This engine can run a simple "amount check → approval or auto‑approve → parallel stamping → join → end" flow. Replacing the in‑memory storage with a real database, swapping the condition evaluator, or plugging external services yields an industrial‑grade BPM engine.

Dify‑Style Workflow: Event‑Driven DAG Scheduler

Dify’s engine centers on a GraphEngine that operates on a parsed graph object. The graph is split into node_mapping (node‑id → node object) and edge_mapping (node‑id → list of outgoing edges). Each edge may carry a condition expression for IF/ELSE branching. Nodes share a global VariablePool; after a node finishes, it writes its outputs as (node_id, var_name) pairs into the pool. Down‑stream nodes resolve placeholders like {{#node_a.var_x#}} to actual values before running.

Execution Algorithm

Pick the current node and call its run() method, which may invoke an LLM, HTTP service, or custom code, producing a NodeRunResult.

Write the result into the variable pool and emit either NodeRunSucceededEvent (with outputs) or NodeRunFailedEvent (with error info). The engine and node communicate via events, not direct calls.

Lookup edge_mapping[current_node]. If there is a single edge, follow it; if multiple, evaluate each edge’s condition against the variable pool and pick the first true one. Parallel branches are submitted as a "parallel block" to a thread pool; the engine records the block so that a downstream join waits for all branches, mirroring the BPM parallel‑gateway token counting.

If a node fails, check for a configured FAIL_BRANCH. If present, follow that edge; otherwise use DEFAULT_VALUE or abort. Optional retry counts cause the node to be re‑executed up to the limit.

Stop when reaching an end node or when no further node can be found. Execution is bounded by max_execution_steps and max_execution_time to avoid infinite loops.

A simplified implementation:

from concurrent.futures import ThreadPoolExecutor, wait
import re

class VariablePool:
    def __init__(self):
        self._store: dict[tuple[str, str], object] = {}
    def set(self, node_id: str, var_name: str, value):
        self._store[(node_id, var_name)] = value
    def resolve(self, raw_config: dict) -> dict:
        pattern = re.compile(r"\{\{#(\w+)\.(\w+)#\}\}")
        resolved = {}
        for key, val in raw_config.items():
            if isinstance(val, str) and pattern.search(val):
                nid, var = pattern.search(val).groups()
                resolved[key] = self._store.get((nid, var))
            else:
                resolved[key] = val
        return resolved

class Graph:
    def __init__(self, nodes: dict, edges: dict):
        self.node_mapping = nodes   # node_id -> {"run": callable, "config": {...}}
        self.edge_mapping = edges   # node_id -> [{"target": ..., "condition": fn|None}]

class GraphEngine:
    def __init__(self, graph: Graph, pool: VariablePool, max_steps: int = 50, max_seconds: int = 120):
        self.graph = graph
        self.pool = pool
        self.max_steps = max_steps
        self.max_seconds = max_seconds
        self.executor = ThreadPoolExecutor(max_workers=8)
    def run(self, start_node: str):
        import time
        start_time = time.time()
        steps = 0
        current_ids = [start_node]
        while current_ids:
            steps += 1
            if steps > self.max_steps or time.time() - start_time > self.max_seconds:
                raise RuntimeError("Execution step/time limit exceeded")
            futures = {self.executor.submit(self._run_node, nid): nid for nid in current_ids}
            wait(futures)
            next_ids = []
            for future, nid in futures.items():
                next_ids.extend(future.result())
            current_ids = next_ids
    def _run_node(self, node_id: str) -> list[str]:
        node = self.graph.node_mapping[node_id]
        inputs = self.pool.resolve(node["config"])
        try:
            outputs = node["run"](inputs)
            for k, v in outputs.items():
                self.pool.set(node_id, k, v)
        except Exception:
            fail_branch = node["config"].get("fail_branch")
            if fail_branch:
                return [fail_branch]
            raise
        edges = self.graph.edge_mapping.get(node_id, [])
        if not edges:
            return []
        chosen = [e["target"] for e in edges if e.get("condition") is None or e["condition"](self.pool)]
        return chosen

This version treats a layer of nodes as a "super‑step" executed concurrently via a thread pool, then collects the next‑layer node IDs. The core algorithm (graph lookup, conditional edge selection, parallel submission, step/time limits) matches Dify’s real engine.

Agent Workflow: ReAct Loop

Both BPM and Dify determine the next step by looking up a pre‑drawn graph. In an Agent workflow, the next tool call and its arguments are generated on‑the‑fly from the LLM’s current output; the graph grows dynamically during execution.

Minimal ReAct implementation:

def run_agent(goal: str, tools: dict, max_turns: int = 10, max_budget_usd: float = 0.5):
    messages = [{"role": "user", "content": goal}]
    spent = 0.0
    for _ in range(max_turns):
        response = call_llm(messages, tool_schemas=[t["schema"] for t in tools.values()])
        spent += estimate_cost(response.usage)
        if spent > max_budget_usd:
            return {"status": "error_max_budget", "messages": messages}
        if response.stop_reason != "tool_use":
            return {"status": "success", "answer": response.text}
        tool_call = response.tool_call
        result = tools[tool_call.name]["fn"](**tool_call.input)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": [{"type": "tool_result", "tool_use_id": tool_call.id, "content": result}]})
    return {"status": "error_max_turns", "messages": messages}

The parameters max_turns and max_budget_usd are the only hard limits not decided by the model; they must be enforced by the outer loop to prevent unbounded execution.

LangGraph: Runtime Graph for the Loop

LangGraph wraps the ReAct loop into a graph where edges are router functions that compute the next active nodes from a shared State. The router may itself invoke an LLM, so routing logic is computed at runtime rather than being pre‑drawn.

class StateGraph:
    def __init__(self):
        self.nodes: dict[str, callable] = {}
        self.reducers: dict[str, callable] = {}
        self.router: callable | None = None
        self.entry: str | None = None
    def add_node(self, name, fn):
        self.nodes[name] = fn
    def set_entry(self, name):
        self.entry = name
    def set_router(self, fn):
        self.router = fn
    def add_reducer(self, field, fn):
        self.reducers[field] = fn
    def invoke(self, initial_state: dict, max_super_steps: int = 25):
        state = dict(initial_state)
        active = [self.entry]
        for _ in range(max_super_steps):
            if not active:
                break
            updates = [self.nodes[name](state) for name in active]
            for update in updates:
                for field, value in update.items():
                    reducer = self.reducers.get(field, lambda old, new: new)
                    state[field] = reducer(state.get(field), value)
            active = self.router(state)
        return state

Comparing this with Dify’s GraphEngine, both share the "graph + shared state + scheduler" skeleton. The difference lies in where the edge definition lives: Dify stores static conditions in edge_mapping, while LangGraph’s router can call the model to decide the next edge.

Comparative Overview

Core data structure : BPM engine – process diagram + token + persisted instance; Dify – DAG (node_mapping/edge_mapping) + variable pool; Agent – conversation history / State + tool registry.

How the next step is decided : BPM – engine follows sequenceFlow / gateway conditions defined at design time; Dify – engine evaluates edge_mapping condition expressions defined at design time; Agent – model generates the next tool call at runtime.

Concurrency model : BPM – parallel‑gateway fork + join token counting; Dify – thread‑pool execution of parallel blocks, counting completions; Agent – dynamic multi‑agent delegation.

Handling long‑running tasks : BPM – serialize whole instance to DB, wake on external event; Dify – log execution, enforce max_execution_steps / time limits; Agent – max_turns / max_budget_usd hard limits + state checkpoint.

Error handling : BPM – compensation logic / job retry; Dify – FAIL_BRANCH / DEFAULT_VALUE per node, node‑level retry; Agent – tool result fed back; model decides retry or alternative strategy.

Evolution of Decision Authority

The three generations share a common skeleton—graph, shared state, scheduler—but differ along two axes:

When edges are fixed : design‑time (BPM, Dify) vs. runtime (Agent).

How state survives long execution : persisted to a database (BPM) vs. in‑memory step/budget limits (Dify, Agent).

This explains the transition:

BPM era : developers define every path; the system merely executes rules.

Dify era : developers still design the flow, but AI fills the capability of individual nodes.

Agent era : developers specify the goal, tools, and constraints; the AI autonomously decides the workflow to achieve the goal.

Understanding where the decision point moves helps engineers refactor systems: the real work is not the visual diagram but the component that determines the next step.