Parallelism

Steps that don’t depend on each other can run at the same time, so a workflow that fetches a document and then summarizes it and extracts stats from it doesn’t have to wait for one before starting the other. A workflow node runs as soon as its inputs are ready, so independent branches run concurrently. You get parallelism two ways: the one-shot Workflow.parallel shortcut, or fan-out in the explicit builder by calling add_edge more than once from the same source.

from loomflow import Workflow, END
 
wf = Workflow("report", max_concurrency=8)
wf.add_node("fetch", fetch)
wf.add_node("summarize", summarize)
wf.add_node("extract_stats", extract_stats)
wf.add_node("combine", combine)
 
wf.set_start("fetch")
wf.add_edge("fetch", "summarize")       # fan-out: both targets get
wf.add_edge("fetch", "extract_stats")   # fetch's output, run concurrently
wf.add_edge("summarize", "combine")     # combine has two in-edges →
wf.add_edge("extract_stats", "combine") # AND-join
wf.add_edge("combine", END)
 
result = await wf.run("https://example.com/q3.pdf")

The data-flow contract

Five rules govern parallel execution in the graph builder:

1. Repeated add_edge from one source fans out. Every target receives the source’s output and the branches run concurrently (bounded by max_concurrency). A repeated identical edge is a no-op; adding a plain edge to a source that has a router replaces the router (a source is either a fan-out or a router, never both).
2. A router still picks exactly one branch. add_router evaluates its classifier and fires exactly one target per evaluation; untaken branches never run. Fan-out is for “all of these”, routers are for “one of these”.
3. A node with several in-edges is an AND-join. It waits until no in-flight (or upstream-pending) node can still deliver to it, then runs once. With a single delivered input it receives the bare value, exactly as in a sequential flow, so the node where two router branches meet again gets a plain value. With two or more delivered inputs it receives a list of the values ordered by edge declaration order (the same list contract as Workflow.parallel’s merge), not by completion order.
4. A failing node cancels in-flight siblings and fails the run with the original exception, never an anyio ExceptionGroup. An Agent (or nested Workflow) node that comes back interrupted stops the run the same way, without raising: see WorkflowResult.
5. The output doesn’t depend on timing. When several branches end without meeting again, result.output is the deepest one (longest path from the entry), then the one whose node was added last, never whichever happened to finish last. A branch that ends early (routes to END) doesn’t become the output of a run that went on elsewhere.

async def combine(inputs: list) -> str:
    summary, stats = inputs        # declaration order:
    ...                            # summarize first, extract_stats second

Purely sequential graphs (chains, routers) run one node at a time, with the same events in the same order as a simple step-by-step walk.

max_concurrency=

Workflow("name", max_concurrency=8)   # default 8

Caps how many nodes may execute simultaneously when the graph fans out. Sequential graphs never have more than one ready node, so the cap is invisible to them. Size it to whatever your slowest resource tolerates (model-provider rate limits, DB connections).

Cycles and caps are preserved

Cycles work the same with parallel branches: feedback and refinement loops behave as they do in a sequential graph. A fan-out inside a loop runs its branches concurrently and its join fires once per iteration: whether a node can still deliver follows the graph’s forward edges, so the loop’s edge back to its head doesn’t count. max_steps and max_visits_per_node are enforced globally by a single scheduler task that owns all counters, so the caps are race-free under concurrency. Join buffering is per wave: on a later loop iteration where only one in-edge delivers, the join receives that bare value.

Event streams from concurrent branches may interleave: a branch’s WORKFLOW_STEP_STARTED can appear before a sibling’s WORKFLOW_STEP_COMPLETED. Every event still carries the same session_id.

Workflow.parallel: fan-out and merge shortcut

When the shape is simply “run N steps on the same input, then merge”, skip the builder:

from loomflow import Workflow
 
wf = Workflow.parallel([fn_a, fn_b, fn_c], merge=combine)

@classmethod
def parallel(
    cls,
    steps: list[StepLike],
    *,
    merge: Callable[[list[Any]], Any] | None = None,
    return_exceptions: bool = False,
    max_concurrency: int | None = None,
    name: str = "parallel",
    telemetry: Telemetry | None = None,
    audit_log: AuditLog | str | Path | dict | None = None,
    memory: Memory | None = None,
    response_tone: str | None = None,
    max_steps: int = 100,
    max_visits_per_node: int = 25,
    workspace: Any | str | Mapping[str, Any] | None = None,
) -> Workflow: ...

Example: fan out to three agents, then merge

from loomflow import Agent, Workflow
 
researcher = Agent("Find sources.",         model="gpt-4.1-mini", tools=[search])
critic     = Agent("Identify weak claims.", model="gpt-4.1-mini")
budgeter   = Agent("Estimate effort.",      model="gpt-4.1-mini")
 
def combine(results: list[str]) -> dict:
    return {"research": results[0], "criticism": results[1], "budget": results[2]}
 
wf = Workflow.parallel([researcher, critic, budgeter], merge=combine)
result = await wf.run("Should we adopt agent harnesses?", user_id="alice")

The three Agents see the same prompt and run concurrently. Each branch runs under its own loom.workflow.step telemetry span (step="fan_out.<branch>"), so per-branch latency and failures show up in traces even though the event stream reports the fan-out as one node.

Concurrency semantics

• Result order matches input order, regardless of completion order.
• A slow step blocks the merge: wall-clock is max(step_durations), not the sum.
• With return_exceptions=False (default), a raising step cancels its siblings and its exception propagates unwrapped, so except ValueError: catches a branch’s ValueError.
• An Agent (or Workflow) branch that comes back interrupted (its max_turns, a budget, a guardrail) stops the run without calling merge: result.interrupted is True and result.interruption_reason names the branch, e.g. "fan_out:branch_0:max_turns_exceeded". With return_exceptions=False it also cancels the other branches. (In 0.17.0 to 0.17.2 the interrupted branch reached merge as an internal object and the run wasn’t marked interrupted.)
• max_concurrency= caps how many branches run at the same time, for example to stay under a provider’s rate limit.

Which one when