Sub-workflow
Run another Flow as one step — inputs and outputs, isolation per run, tracing, error propagation, recursion guards, Map fan-out, and nested flows in YAML.
SubWorkflow runs a Flow as a single node of another flow. The node's mapped inputs become the inner flow's input, the inner run is traced under the node's run, and the result of the inner flow's Output node is the node's output. It is how a decision — an eligibility check, a pricing chain — is built once and reused by several workflows, and how a batch runs it once per item through a Map.
A first sub-workflow
The inner flow is an ordinary flow with an Input and an Output node — the shape every workflow built in the editor has:
from dynamiq import Workflow
from dynamiq.flows import Flow
from dynamiq.nodes import InputTransformer
from dynamiq.nodes.node import NodeDependency
from dynamiq.nodes.operators import DecisionTable, SubWorkflow
from dynamiq.nodes.types import DecisionRule, NamedField, SubWorkflowField
from dynamiq.nodes.utils import Input, Output
from dynamiq.runnables import RunnableConfig
def eligibility_flow() -> Flow:
start = Input(id="start", name="start")
table = DecisionTable(
id="table",
name="eligibility",
input_columns=[NamedField(name="fico", type="int")],
output_columns=[NamedField(name="decision", type="string")],
rules=[
DecisionRule(id="r1", name="prime", when=[">= 740"], then=["approve"]),
DecisionRule(id="r2", name="rest", when=[""], then=["decline"]),
],
depends=[NodeDependency(node=start)],
input_transformer=InputTransformer(selector={"fico": "$.start.output.fico"}),
)
end = Output(
id="end",
name="end",
depends=[NodeDependency(node=table)],
input_transformer=InputTransformer(selector={"decision": "$.table.output.decision"}),
)
return Flow(id="eligibility-flow", name="Eligibility", nodes=[start, table, end])
outer_start = Input(id="outer-start", name="start")
check = SubWorkflow(
id="check",
name="eligibility_check",
flow=eligibility_flow(),
input_fields=[SubWorkflowField(name="fico", type="int", required=True)],
output_fields=[SubWorkflowField(name="decision", type="string")],
depends=[NodeDependency(node=outer_start)],
input_transformer=InputTransformer(selector={"fico": "$.outer-start.output.fico"}),
)
outer_end = Output(
id="outer-end",
name="end",
depends=[NodeDependency(node=check)],
input_transformer=InputTransformer(selector={"decision": "$.check.output.decision"}),
)
workflow = Workflow(id="outer", flow=Flow(id="outer-flow", nodes=[outer_start, check, outer_end]))
result = workflow.run(input_data={"fico": 760}, config=RunnableConfig(callbacks=[]))
print(result.output["outer-end"]["output"])
# {'decision': 'approve'}Configuration
namestrflowFlowinput_fieldslist[SubWorkflowField]output_fieldslist[SubWorkflowField]workflow_idstrworkflow_version_idstrSubWorkflowField(name, type="Any", required=False) comes from dynamiq.nodes.types.
How a run works
- The node checks that no flow with the same id is already running above it, then that every
requiredinput is present. - It copies the flow and runs the copy with the mapped inputs. A flow keeps its run state on itself, so two nodes that hold the same flow in parallel branches, a
Mapfanning the node out over a batch, or a retried run would otherwise overwrite each other's results. The copy keeps the node ids, so selectors, traces, and overrides still line up; it costs well under a millisecond for a small flow. - The copy runs on its own checkpoint settings. The parent's checkpoint configuration — a resume id above all — describes the parent's run and is not forwarded: the inner flow runs whole, and the parent records the node once it completes.
- Under a dry run, the copy does not clean up when it returns. The node keeps the copies that ran and cleans up what their writers ingested when the flow holding the node ends — a
Mapdoes the same for the clones it ran per item — so a retriever after the sub-workflow still reads the documents, the way it would with the same nodes inlined. - The node returns the inner Output node's output. A flow without exactly one Output node returns every node's output keyed by node id instead.
Every node of the inner flow, and the inner flow itself, is traced under the sub-workflow node's run, so a trace tree shows the callee nested beneath the step that called it. Add a tracing handler to the outer run and the inner nodes come with it.
Failures, timeouts, and retries
- A node failing inside the flow fails the sub-workflow with
Sub-workflow '<name>' failed: <node>: <message>, naming every failed inner node. Cancellation propagates. - The node's own error handling applies around the whole inner run: a
timeout_secondsbounds the flow, retries run the copied flow again — one execution run per attempt in the trace — andbehavior=Behavior.RETURNlets the outer flow continue past a failed callee. - A missing required input fails before anything runs:
Sub-workflow '<name>': required inputs missing: fico.
Recursion is refused
A flow that calls itself — directly, or through another workflow that calls back — would recurse into its own run state. The ids of the flows running above a node travel with the run, and a node whose flow is already active fails with flow '<id>' is already running above this node instead of recursing. The YAML loader refuses a flow that holds itself before anything runs, naming the chain.
from dynamiq.flows import Flow
from dynamiq.nodes.operators import SubWorkflow
from dynamiq.runnables import RunnableConfig
loop = SubWorkflow(id="loop", name="loop")
loop.flow = Flow(id="loop-flow", nodes=[loop])
result = loop.run(input_data={}, config=RunnableConfig(callbacks=[]))
print(result.status, result.error.message)
# RunnableStatus.FAILURE Sub-workflow 'loop' failed: loop: Sub-workflow 'loop': flow 'loop-flow' is already running above this nodeBatches: a sub-workflow inside a Map
Map clones its inner node once per item, and each clone runs its own copy of the flow, so a batch prices every application in isolation and in parallel:
from dynamiq import Workflow
from dynamiq.flows import Flow
from dynamiq.nodes import InputTransformer
from dynamiq.nodes.node import NodeDependency
from dynamiq.nodes.operators import DecisionTable, Map, SubWorkflow
from dynamiq.nodes.types import DecisionRule, NamedField, SubWorkflowField
from dynamiq.nodes.utils import Input, Output
from dynamiq.runnables import RunnableConfig
def eligibility_flow() -> Flow:
start = Input(id="start", name="start")
table = DecisionTable(
id="table",
name="eligibility",
input_columns=[NamedField(name="fico", type="int")],
output_columns=[NamedField(name="decision", type="string")],
rules=[
DecisionRule(id="r1", name="prime", when=[">= 740"], then=["approve"]),
DecisionRule(id="r2", name="rest", when=[""], then=["decline"]),
],
depends=[NodeDependency(node=start)],
input_transformer=InputTransformer(selector={"fico": "$.start.output.fico"}),
)
end = Output(
id="end",
name="end",
depends=[NodeDependency(node=table)],
input_transformer=InputTransformer(selector={"decision": "$.table.output.decision"}),
)
return Flow(id="eligibility-flow", name="Eligibility", nodes=[start, table, end])
check = SubWorkflow(
id="check",
name="eligibility_check",
flow=eligibility_flow(),
input_fields=[SubWorkflowField(name="fico", type="int", required=True)],
output_fields=[SubWorkflowField(name="decision", type="string")],
)
batch = Map(id="batch", name="batch", node=check, max_workers=4)
workflow = Workflow(flow=Flow(nodes=[batch]))
result = workflow.run(
input_data={"input": [{"fico": 760}, {"fico": 640}, {"fico": 800}]},
config=RunnableConfig(callbacks=[]),
)
print(result.output["batch"]["output"]["output"])
# [{'decision': 'approve'}, {'decision': 'decline'}, {'decision': 'approve'}]Cloning a flow re-links the dependencies and output references between the copied nodes and rewrites id-based selectors, so Choice gates and .inputs() references inside the flow keep holding in every clone — including a Map inside a Map.
YAML: nested flows
A sub-workflow references its flow by id, and that flow is declared in the flows section like any other, with its nodes in nodes. The loader builds referenced flows innermost first, so a sub-workflow can sit at any depth; the dumper writes them back the same way and refuses a duplicate flow or node id, which the id-keyed sections would otherwise merge silently. to_yaml_file writes one entry in workflows, which is what the platform runtime expects, since it loads a file without naming a workflow; a hand-written file may still define several and pick one with wf_id:
nodes:
start:
type: dynamiq.nodes.utils.Input
name: start
table:
type: dynamiq.nodes.operators.DecisionTable
name: eligibility
input_columns:
- { id: fico, name: fico, type: int }
output_columns:
- { id: decision, name: decision, type: string }
rules:
- { id: r1, name: prime, when: [">= 740"], then: [approve] }
- { id: r2, name: rest, when: [""], then: [decline] }
depends:
- node: start
input_transformer:
selector:
fico: $.start.output.fico
end:
type: dynamiq.nodes.utils.Output
name: end
depends:
- node: table
input_transformer:
selector:
decision: $.table.output.decision
batch-start:
type: dynamiq.nodes.utils.Input
name: start
batch:
type: dynamiq.nodes.operators.Map
name: batch
max_workers: 4
node:
id: check
type: dynamiq.nodes.operators.SubWorkflow
name: eligibility_check
flow: eligibility-flow
input_fields:
- { name: fico, type: int, required: true }
output_fields:
- { name: decision, type: string }
depends:
- node: batch-start
input_transformer:
selector:
input: $.batch-start.output.applications
batch-end:
type: dynamiq.nodes.utils.Output
name: end
depends:
- node: batch
input_transformer:
selector:
decisions: $.batch.output.output
flows:
eligibility-flow:
name: Eligibility
nodes: [start, table, end]
batch-flow:
name: Batch eligibility
nodes: [batch-start, batch, batch-end]
workflows:
batch-eligibility:
flow: batch-flowfrom dynamiq import Workflow
from dynamiq.runnables import RunnableConfig
workflow = Workflow.from_yaml_file(file_path="batch_eligibility.yaml", init_components=True)
result = workflow.run(
input_data={"applications": [{"fico": 760}, {"fico": 640}]},
config=RunnableConfig(callbacks=[]),
)
print(result.output["batch-end"]["output"])
# {'decisions': [{'decision': 'approve'}, {'decision': 'decline'}]}Node ids must be unique across every flow in the file, since the nodes section is keyed by id — the two Input nodes above are start and batch-start, even though both are named start. Dumping a workflow that holds sub-workflows with to_yaml_file produces this layout, including a sub-workflow held as an agent tool.
On the platform
In the workflow editor the node picks a released workflow of the same project and a version (or the latest release), and the platform inlines that version's nodes as the flow when the workflow is saved, tested, or deployed, filling workflow_id and workflow_version_id. A deployment keeps the version it was built with. See the Sub-workflow node guide.