Workflow

A Workflow is a graph of steps that you define in code. Your code, not a model, decides which step runs next. Use one when you know the steps in advance and want every run to follow them predictably; use an Agent when the model should decide what to do. The two share the same telemetry and audit log, and you can put one inside the other.

Pick based on who should be in control, not on style. See Agent vs Workflow. Production systems usually need both, composed together.

Three ways to build one

from loomflow import Workflow, step

1. Plain Python with @step

Write the flow as ordinary async Python. Decorating a function with @step gives it its own telemetry span and audit entries when it runs inside a workflow; control flow stays as if/await/return:

from loomflow import step
 
@step
async def classify(text: str) -> str: ...
 
@step
async def respond(text: str, label: str) -> str: ...
 
async def my_workflow(text: str, user_id: str) -> str:
    label = await classify(text)
    return await respond(text, label)

2. Sugar constructors for common shapes

No graph builder needed; one call returns a fully-wired Workflow:

wf = Workflow.chain([classify, respond])
wf = Workflow.route(classifier, {"a": fn_a, "b": fn_b})
wf = Workflow.parallel([fn_a, fn_b, fn_c], merge=combine)

3. Explicit graph builder

For loops, multi-stage branching, or flows where the graph’s shape is itself what gets reviewed (multi-stage approval chains, for example):

from loomflow import Workflow, END
 
wf = Workflow("triage")
wf.add_node("classify", classify)    # returns {"team": ..., "ticket": ...}
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)

Read more

Shared memory across the graph

Pass memory= to a workflow and every nested Agent step that didn’t bring its own inherits it. Episodes / facts written by one agent become recall-able by the next without per-agent wiring:

from loomflow import Workflow, Agent, InMemoryMemory
 
mem = InMemoryMemory()
agent_a = Agent("Stage 1", model="gpt-4.1-mini")
agent_b = Agent("Stage 2", 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`.

Explicit always wins. The resolution order is Agent(memory=...) > Workflow(memory=...) > the agent’s per-instance default.

Ambient response tone

Same propagation pattern, different concern. response_tone= on a workflow flows to every nested Agent that didn’t bring its own:

wf = Workflow.chain(
    [researcher, writer, reviewer],
    response_tone="executive",
)
# All three agents inherit "executive" tone unless one set its own.

Resolution: agent.run(response_tone=) per-call > Agent(response_tone=) > Workflow(response_tone=) > none. Uses the same contextvar propagation as memory=, reset in finally so tones do not leak across runs. See Agent reference: response_tone for the preset effects and free-form passthrough.

Composition with Agent

• Agent inside a Workflow. Pass an Agent instance to add_node; the framework calls .run(input) automatically and threads the live RunContext (user_id / session_id / metadata) through to the inner agent run.
• Workflow inside an Agent. Call wf.as_tool() to get a Tool the agent can invoke. The whole workflow runs as one tool call from the agent’s perspective.

Both directions share the same telemetry spans, audit-log entries and user_id partitioning. In a trace, workflow steps are loom.workflow.step spans tagged pattern="workflow", so you can tell which decisions your code made and which the model made.