Parallelism

Workflow execution is readiness-based: a node runs as soon as its inputs are complete, so independent branches run concurrently. You get parallelism two ways — the one-shot Workflow.parallel sugar, or graph-level 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

Four 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 the sequential model, so router merge points behave as before. 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.

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

Purely sequential graphs (chains, routers) execute exactly as before — same events, same order.

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 stay first-class under the parallel scheduler — feedback / refinement loops work exactly as they did sequentially. 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 + merge sugar

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,
    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 three agents, 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 the exception propagates.

Which one when