Workflow

A Workflow is a developer-controlled DAG. It’s the peer primitive to Agent. They share one observability spine but have very different control models.

Picking between them is an engineering decision, not stylistic. 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

Decorate any async def and it picks up telemetry + audit + the journal hooks automatically. 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 cases where the graph IS the artifact (compliance flows, multi-stage approval chains):

wf = Workflow("triage")
wf.add_node("classify", classify)
wf.add_node("billing",  billing_agent)
wf.add_node("tech",     tech_agent)
wf.set_start("classify")
wf.add_router(
    "classify",
    lambda r: r.lower(),
    {"billing": "billing", "tech": "tech"},
)
wf.add_edge("billing", END)
wf.add_edge("tech", END)

Read more

Shared memory across the graph (0.9.19+)

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`.

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

Ambient response tone (0.9.32+)

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. Implemented with the same contextvar pattern as memory=, reset in finally so tones do not leak across runs. See Agent reference: response_tone for the 7 preset effects + 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 reuse the same observability spine. Telemetry spans, audit-log entries, user_id partition. A trace shows exactly which decisions were workflow-deterministic and which were LLM-driven, tagged via the pattern span attribute.