Explicit graph builder

The builder lets you declare a workflow node by node and edge by edge, instead of using the chain / route / parallel shortcuts. Use it when the shortcuts don’t fit: loops, several stages of branching, or flows such as multi-step approval chains where the shape of the graph is itself something people need to review.

Each node receives the previous node’s output, and a router’s function sees that same output to pick the next node:

from loomflow import Workflow, END
 
 
def classify(ticket: str) -> dict:
    team = "billing" if "invoice" in ticket.lower() else "tech"
    return {"team": team, "ticket": ticket}
 
 
async def billing(item: dict) -> str:
    return f"billing handles: {item['ticket']}"
 
 
async def tech(item: dict) -> str:
    return f"tech handles: {item['ticket']}"
 
 
wf = Workflow("triage")
wf.add_node("classify", classify)
wf.add_node("billing",  billing)
wf.add_node("tech",     tech)
wf.set_start("classify")
wf.add_router(
    "classify",
    lambda item: item["team"],
    {"billing": "billing", "tech": "tech"},
)
wf.add_edge("billing", END)
wf.add_edge("tech", END)
 
result = await wf.run("My invoice is wrong")
print(result.output)   # billing handles: My invoice is wrong
print(result.visited)  # ['classify', 'billing']

Constructor

class Workflow:
    def __init__(
        self,
        name: str = "workflow",
        *,
        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,
        max_concurrency: int = 8,
        workspace: Any | str | Mapping[str, Any] | None = None,
    ) -> None: ...

Both safety caps raise RuntimeError when exceeded, naming the offending node. Hitting one usually means your routing logic never picks the branch that ends the loop.

Builder methods

add_node(name, fn)

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

Register a node. fn can be an async def, sync function, Agent, or nested Workflow. Returns self for chaining.

wf.add_node("extract", extract_fn)
wf.add_node("score",   scoring_agent)        # Agent works
wf.add_node("subflow", another_workflow)     # nested Workflow works

add_edge(source, target)

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

Unconditional edge from source to target. Calling add_edge more than once with the same source fans out: every target receives the source’s output and the branches run concurrently (see Parallel).

wf.add_edge("extract", "score")
wf.add_edge("score",   END)

add_edge(START, "node") is an alias for set_start("node"), so graphs read symmetrically with the END sentinel:

from loomflow import Workflow, START, END
 
wf = Workflow("triage")
wf.add_node("classify", classify)
wf.add_node("billing",  billing)
wf.add_node("tech",     tech)
wf.add_edge(START, "classify")    # alias for set_start("classify")
wf.add_router(
    "classify",
    lambda item: item["team"],
    {"billing": "billing", "tech": "tech"},
)
wf.add_edge("billing", END)
wf.add_edge("tech", END)

add_edge(START, END) and add_edge(END, ...) are rejected with messages pointing at the right method. set_start("node") continues to work. Pick whichever form reads better.

add_router(source, fn, routes, *, default=None)

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

Conditional edge. At run time fn(input_value) is called; the returned key is looked up in routes to pick the next node.

A source has either a router or plain edges, never both. Calling add_edge(s, ...) after add_router(s, ...) replaces the router, and vice versa. set_start("node") and add_router(START, ...) are also mutually exclusive.

add_router(START, ...): branch at the entry of the graph. Classify the workflow’s input and dispatch to one of N first nodes without a synthetic passthrough “entry” step:

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,
)
wf.add_edge("step_1", END)
wf.add_edge("step_3", END)

When the source is START, route targets are validated at build time, so a typo raises a clear error right away. For any other source, every edge, route and default is checked when the run starts: a target that names an unregistered node raises ValueError (listing each bad edge) before any node runs. The to_mermaid / to_dot renders show START -->|key| node directly (no synthetic entry node in the diagram).

wf.add_router(
    "score",
    lambda r: "high" if r["score"] >= 0.8 else "low",
    {"high": "ship", "low": "revise"},
    default="revise",
)

set_start(node)

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

Mark a registered node as the entry point. Required before run / stream. Calling them without a start raises RuntimeError.

wf.set_start("extract")

Running

run(input, *, user_id=None, session_id=None, metadata=None)

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, return a WorkflowResult.

Returns a WorkflowResult with output, visited, per_step, interrupted and interruption_reason (see below). user_id, session_id and metadata reach every Agent node and nested Workflow node.

stream(input, *, user_id=None, session_id=None, metadata=None)

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:

• WORKFLOW_STARTED: once at the start.
• WORKFLOW_STEP_STARTED / WORKFLOW_STEP_COMPLETED: once per visited node.
• WORKFLOW_STEP_FAILED: if a node raises, followed by an ERROR event (the run also raises after cleanup).
• WORKFLOW_COMPLETED: once at the end. When an Agent node came back interrupted, its WORKFLOW_STEP_COMPLETED and the WORKFLOW_COMPLETED payloads carry interrupted: True and the interruption_reason.

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.

WorkflowResult

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

Cycles

Loomflow’s workflow supports cycles. Feedback loops (A → B → classify → (C \| D \| END) → B and similar refinement patterns) are built-in. The two safety caps (max_steps / max_visits_per_node) keep buggy routers from looping forever.

from loomflow import Workflow, END
 
wf = Workflow("refine_until_good", max_visits_per_node=5)
wf.add_node("draft",  draft_fn)
wf.add_node("review", review_fn)
wf.add_node("judge",  judge_fn)
wf.add_node("revise", revise_fn)
 
wf.set_start("draft")
wf.add_edge("draft", "review")
wf.add_edge("review", "judge")
wf.add_router(
    "judge",
    lambda r: "accept" if r["score"] >= 0.85 else "revise",
    {"accept": END, "revise": "revise"},
)
wf.add_edge("revise", "review")     # ← back-edge: closes the cycle

If the judge keeps picking “revise”, the workflow stops at max_visits_per_node=5 re-entries of review and raises:

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

For cycle-handling examples see examples/08_workflow_loop.py (in the framework repo).

Composition with Agent

Pass an Agent to add_node; the framework calls .run(input).output and threads the live RunContext through. Pass another Workflow to nest. See Composition.

When to reach for the explicit builder

• Compliance flows: multi-stage approval chains and sign-off gates, where the graph itself is what gets reviewed.
• Cycles: refinement loops, retry-with-feedback, escalation chains.
• Branching beyond “one classifier, then dispatch”: multi-stage routing, conditional skips, parallel-then-merge inside a larger flow.

For linear pipelines use Workflow.chain. For one-shot classify-and-dispatch use Workflow.route. The explicit builder is worth its ceremony when the graph shape is the documentation.