Workflow

from loomflow import Workflow, WorkflowResult, step, START, END

A Workflow runs a graph of steps that you define in code: plain functions, Agents or other workflows, joined by edges and routers. Use it when you know the order of the steps and want it fixed and testable; use an Agent when the model should decide what to do next. The two compose: an Agent can be a step, and a Workflow can be a tool.

Build a graph explicitly with add_node / add_edge / add_router, or use one of the shortcut constructors (chain, route, parallel) for the common shapes. For the conceptual page see Workflow.

Class signature

class Workflow:
    name: str
 
    def __init__(
        self,
        name: str = "workflow",
        *,
        telemetry: Telemetry | None = None,
        audit_log: AuditLog | str | Path | dict[str, Any] | None = None,
        memory: Memory | None = None,
        response_tone: str | None = None,
        max_steps: int = 100,
        max_visits_per_node: int = 25,
        max_concurrency: int = 8,
        workspace: Workspace | str | Mapping[str, Any] | None = None,
    ) -> None: ...

Constructor parameters

name

Used in telemetry spans and audit entries. Pick something descriptive, since it shows up in traces and dashboards.

memory

Memory shared by every nested Agent step that didn’t bring its own. Episodes and facts written by one agent can be recalled by the next without per-agent wiring.

from loomflow import Workflow, Agent, InMemoryMemory
 
mem = InMemoryMemory()
agent_a = Agent("...", model="gpt-4.1-mini")
agent_b = Agent("...", model="gpt-4.1-mini")
 
wf = Workflow.chain([agent_a, agent_b], memory=mem)
await wf.run("hi", user_id="alice", session_id="conv-1")
# Both agents wrote to / read from `mem`.

Resolution order (explicit always wins):

1. Agent(memory=my_mem) keeps using my_mem even inside a workflow that has memory= set.
2. Otherwise Workflow(memory=mem) is used as the fallback.
3. Otherwise the agent’s per-instance default (in-memory).

The workflow sets this for the duration of each run and resets it afterwards, so memory does not leak across workflow runs. Reuse the same Memory instance across wf.run() calls to keep memory between conversations.

telemetry

Wire any Telemetry sink to emit a loom.workflow.step span per node, tagged with step / user_id / session_id / pattern="workflow". See Telemetry.

audit_log

Wire FileAuditLog(...) to record step_started / step_completed / step_failed entries with user_id attribution. It is the same audit log the rest of the framework writes to, so agent and workflow entries interleave under the same session_id.

Shortcut: pass a str or Path and the framework wraps it as a FileAuditLog, so you don’t need to import the backend just to write to a file. The dict form from Agent(audit_log=...) works here too.

wf = Workflow.chain([...], audit_log="run.log")
# equivalent to: audit_log=FileAuditLog("run.log")

Anything else (e.g. a bare list) is rejected at construction time with a TypeError listing the valid forms.

max_steps

Hard cap on total step executions per run / stream call. Linear chains visit each step once; cyclic flows spend this budget on every iteration. Must be >= 1. Hitting the cap raises RuntimeError naming the offending node.

max_visits_per_node

Per-node visit cap. Tighter than max_steps because most runaways are one node looping on itself. Must be >= 1.

RuntimeError: workflow 'refine_until_good' re-entered 'review'
more than max_visits_per_node=25 times; the router controlling the
loop probably never picks the termination branch

max_concurrency

How many nodes may run at the same time when the graph fans out (see Fan-out and joins). Sequential graphs never have more than one node ready, so it has no effect on them. Must be >= 1.

workspace

A shared notebook for every nested Agent step that didn’t set its own workspace=. Those agents get the five notebook tools and the notebook section of the system prompt automatically, so the steps of one workflow can leave notes for each other. Accepts the same shapes as Agent(workspace=...).

response_tone

Tone applied to every nested Agent step that didn’t set its own response_tone. Preset name ("casual" / "professional" / "technical" / "legal" / "finance" / "executive" / "academic") or any free-form string.

from loomflow import Workflow
 
# Every agent step inherits "executive" unless it set its own tone.
wf = Workflow.chain([researcher, writer, reviewer], response_tone="executive")

Like memory, the tone is set at the start of run / stream and reset at the end, so it never leaks across workflow runs.

Resolution order for any nested Agent.run:

1. agent.run(..., response_tone=...) per-call.
2. Agent(tuning=Tuning(response_tone=...)) agent default.
3. Workflow(response_tone=...) (this kwarg).
4. None: no tone directive.

See response_tone on the Agent page for the full preset table and how tone differs from instructions and persona.

Builder methods

add_node

def add_node(self, name: str, fn: StepLike) -> Workflow: ...

Register a node. Returns self for chaining. Re-registering the same name raises ValueError.

How each kind of step is called:

The metadata passed on is the run’s metadata without keys that start with _ (the framework keeps its own entries there). An Agent or nested Workflow node whose result comes back interrupted stops the run instead of passing its partial output on; see WorkflowResult. On a durable runtime each Agent node runs as a child run of the workflow run, with its own journal namespace; see Composition: on a durable runtime.

add_edge

def add_edge(
    self,
    source: str | _Sentinel,
    target: str | _Sentinel,
) -> Workflow: ...

Unconditional edge. The target can be a registered node name or END; the source can be a registered node or the START sentinel.

add_edge(START, "node") is equivalent to set_start("node"). It lets graphs read symmetrically with the END sentinel and matches the LangGraph mental model:

wf.add_edge(START, "first")     # alias for set_start("first")
wf.add_edge("first", "second")
wf.add_edge("second", END)

add_edge(START, END) and add_edge(END, ...) are rejected with messages that point at the right method.

Fan-out and joins

Calling add_edge more than once from the same source fans out: every target receives the source’s output and the branches run concurrently (up to max_concurrency at a time). Adding the same edge twice is a no-op.

A node with several incoming edges is a join. It waits until no running or upstream node can still deliver to it, then runs once. With one delivered input it receives the bare value; with two or more it receives a list of values in the order the edges were declared (not the order they finished). If any branch raises, the other running branches are cancelled and the run fails with the original exception (not an ExceptionGroup). If an Agent node comes back interrupted, the run stops there; see WorkflowResult.

Every edge target, route and router default must name a registered node. The check runs when run() / stream() starts, before any node runs, and raises ValueError listing each bad edge ("classify -> bilin"), so a typo can’t end the run silently after its source node.

wf = Workflow("research")
wf.add_node("plan", plan)
wf.add_node("web", search_web)
wf.add_node("papers", search_papers)
wf.add_node("summarise", summarise)   # receives [web_result, papers_result]
wf.set_start("plan")
wf.add_edge("plan", "web")
wf.add_edge("plan", "papers")          # second edge from "plan": fan-out
wf.add_edge("web", "summarise")
wf.add_edge("papers", "summarise")
wf.add_edge("summarise", END)

add_router

def add_router(
    self,
    source: str | _Sentinel,
    fn: Callable[[Any], Any],
    routes: Mapping[str, str | _Sentinel],
    *,
    default: str | _Sentinel | None = None,
) -> Workflow: ...

Conditional edge. At run time fn(previous_output) is called; the returned key is converted with str() and looked up in routes. A router picks exactly one branch each time; the others don’t run.

A source has either plain edges or one router, never both. Adding a router replaces any plain edges from that source, and adding a plain edge to a source that has a router replaces the router. Likewise set_start("node") and add_router(START, ...) replace each other: whichever you call last wins.

add_router(START, ...) classifies the workflow’s input and dispatches to one of several first nodes, with no pass-through “entry” step needed. It mirrors LangGraph’s add_conditional_edges(START, ...):

from loomflow import Workflow, START, END
 
wf = Workflow()
wf.add_node("step_1", step_1)
wf.add_node("step_3", step_3)
wf.add_router(
    START,
    fn=lambda q: "step_1" if "work" in q else "step_3",
    routes={"step_1": "step_1", "step_3": "step_3"},
    default=END,                    # optional fallback
)
wf.add_edge("step_1", END)
wf.add_edge("step_3", END)

The to_mermaid / to_dot renders draw START -->|key| node directly, with no synthetic entry node in the diagram.

set_start

def set_start(self, node: str) -> Workflow: ...

Mark a registered node as the entry point. A workflow needs an entry point (set_start, add_edge(START, ...) or add_router(START, ...)) before run / stream. Raises ValueError for unregistered names.

Shortcut constructors

Workflow.chain

@classmethod
def chain(
    cls,
    steps: list[StepLike],
    *,
    name: str = "chain",
    telemetry: Telemetry | None = None,
    audit_log: AuditLog | str | Path | dict[str, Any] | None = None,
    memory: Memory | None = None,
    response_tone: str | None = None,
    max_steps: int = 100,
    max_visits_per_node: int = 25,
) -> Workflow: ...

Linear sequence: each step’s output is the next step’s input. See Workflow.chain.

Workflow.route

@classmethod
def route(
    cls,
    classifier: StepLike,
    routes: Mapping[str, StepLike],
    *,
    default: StepLike | None = None,
    name: str = "route",
    telemetry: Telemetry | None = None,
    audit_log: AuditLog | str | Path | dict[str, Any] | None = None,
    memory: Memory | None = None,
    response_tone: str | None = None,
    max_steps: int = 100,
    max_visits_per_node: int = 25,
) -> Workflow: ...

Classify, then dispatch to one handler. See Workflow.route. Handlers receive the original input, not the classifier’s output.

Workflow.parallel

@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[str, Any] | None = None,
    memory: Memory | None = None,
    response_tone: str | None = None,
    max_steps: int = 100,
    max_visits_per_node: int = 25,
    workspace: Workspace | str | Mapping[str, Any] | None = None,
) -> Workflow: ...

Run every step on the same input at once, then combine the results. See Workflow.parallel.

• merge(results) produces the final output; by default the list of results is returned unchanged.
• With return_exceptions=False (the default), the first failing step cancels the others and its exception propagates as itself (not wrapped in an ExceptionGroup). With True, a failing step’s exception object is placed at its index in the results list and the workflow completes with the partial results.
• An Agent or Workflow step that comes back interrupted stops the run without calling merge (interrupted=True, interruption_reason "fan_out:<branch>:<reason>"), cancelling the others unless return_exceptions=True.
• max_concurrency caps how many steps run at once (default: all). Must be >= 1.
• Each branch gets its own loom.workflow.step telemetry span, so per-branch latency shows up in traces.

Running

run

async def run(
    self,
    input: Any = None,
    *,
    user_id: str | None = None,
    session_id: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> WorkflowResult: ...

Execute the graph. Returns a WorkflowResult.

stream

async def stream(
    self,
    input: Any = None,
    *,
    user_id: str | None = None,
    session_id: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> AsyncIterator[Event]: ...

Same execution as run, exposed as an async generator of Events. Yields:

Consumers can break out of the iterator early to cancel the run. The in-flight nodes are cancelled, and the caller’s next await runs normally. On trio, close the stream with contextlib.aclosing(wf.stream(...)) before abandoning it.

as_tool

def as_tool(
    self,
    *,
    name: str | None = None,
    description: str | None = None,
    input_arg: str = "input",
) -> Tool: ...

Expose this workflow as a Tool an Agent can invoke.

The workflow runs with the calling agent run’s user_id, session_id and metadata. See Composition (direction 2: workflow as a tool).

Visualization

to_mermaid

def to_mermaid(self) -> str: ...

Returns a Mermaid flowchart TD rendering of the graph. Paste it into GitHub Markdown (renders inline) or mermaid.live  for PNG / SVG export.

• Solid arrows are unconditional edges.
• Labelled solid arrows are router branches.
• Dotted arrows are router defaults.
• START and END render as stadium-shaped nodes.

print(wf.to_mermaid())

```mermaid
flowchart TD
    n_classify["classify"]
    n_route_billing["route_billing"]
    n_route_tech["route_tech"]
    n_route_default["route_default"]
    START([START]) --> n_classify
    n_classify -->|billing| n_route_billing
    n_classify -->|tech| n_route_tech
    n_classify -.->|default| n_route_default
    n_route_billing --> END([END])
    n_route_tech --> END([END])
    n_route_default --> END([END])
```

to_dot

def to_dot(self) -> str: ...

Same picture as to_mermaid, emitted in Graphviz DOT for users who prefer the Graphviz toolchain:

with open("graph.dot", "w") as f:
    f.write(wf.to_dot())
# dot -Tpng graph.dot -o graph.png

Mermaid is usually the easier choice, since it renders inline on GitHub, GitLab and Notion without installing anything.

_repr_markdown_

Jupyter, JupyterLab and VS Code notebooks render the diagram inline when a workflow is the last expression in a cell. No imports or extra calls are needed. The markdown representation is the Mermaid flowchart wrapped in a ```mermaid fence.

wf  # as the last line of a notebook cell: renders the diagram

Related types

WorkflowResult

@dataclass
class WorkflowResult:
    output: Any
    visited: list[str] = field(default_factory=list)
    per_step: dict[str, Any] = field(default_factory=dict)
    interrupted: bool = False
    interruption_reason: str | None = None

START / END sentinels

from loomflow import START, END

Distinct sentinel tokens compared by identity. START marks the entry of the graph and can be the source of add_edge or add_router; END is an edge or route target that ends the run:

wf.add_edge(START, "classify")             # entry point
wf.add_edge("ship", END)                   # terminate after 'ship'
wf.add_router("classify", fn, {"a": "node_a", "b": END})  # 'b' terminates

step decorator

def step(
    fn: Callable[..., Awaitable[Any]] | None = None,
    *,
    name: str | None = None,
) -> Any: ...

Adds a telemetry span (when the workflow has telemetry=) and step_started / step_completed / step_failed audit entries (when it has audit_log=) to plain Python control flow. It doesn’t journal the step. Outside a workflow run it does nothing extra. See The @step decorator.

@step
async def classify(text: str) -> str: ...
 
@step(name="ml-classifier-v2")
async def classify_v2(text: str) -> str: ...

Events emitted

Every workflow event payload carries workflow (the workflow name). Step events add node. WORKFLOW_STEP_COMPLETED looks like this:

{
    "workflow": "support",      # workflow name
    "node": "classify",         # node name
    "output": "billing",        # the node's return value
}

WORKFLOW_STEP_FAILED carries error (str(exc)) instead of output. WORKFLOW_STARTED carries the input and WORKFLOW_COMPLETED the final output. For an interrupted Agent node, its WORKFLOW_STEP_COMPLETED and the WORKFLOW_COMPLETED payloads also carry interrupted: True and interruption_reason.

Concurrency model

Workflow instances are safe to share across concurrent calls. Each run() / stream() builds its own per-run state and gets its own RunContext, so two wf.run() calls in flight on the same workflow don’t interfere.

The graph (nodes / edges / routers) is read-only at run time. Calling add_node / add_edge / add_router during a run is undefined and not protected by a lock, so build the graph before the first run.

Source

loomflow/workflow/__init__.py