Skip to main content
Flyte lets you call a workflow from inside another workflow, creating a subworkflow. The distinction from launch plans is conceptual: launch plans are “pass by pointer” (they trigger a separate execution) while subworkflows are “pass by value” (the entire graph is inlined into the parent).

​
Subworkflows vs. external workflows

SubworkflowExternal workflow (launch plan)
Execution contextShared with parentSeparate execution ID
Parallelism constraintsInherits parent constraintsIndependent constraints
Kubernetes clusterSame clusterCan be different cluster
ObservabilityInlined in parent graphSeparate entity in console

​
When to use subworkflows

Use subworkflows when you want to:
  • Reuse workflow logic across multiple parent workflows without duplicating code
  • Enforce parallelism constraints — subworkflow nodes obey the parent’s overall limits
  • Compose pipelines from smaller, independently testable workflow units
When workflow A is used as a subworkflow of workflow B, running B duplicates A’s entire graph into B at the point of invocation.

​
Basic example: regression line

The following example calculates slope, intercept, and the corresponding y-value by composing two workflows. Import and define tasks: 
import typing
from flytekit import task, workflow


@task
def slope(x: typing.List[int], y: typing.List[int]) -> float:
    sum_xy = sum(xi * yi for xi, yi in zip(x, y))
    sum_x = sum(x)
    sum_y = sum(y)
    n = len(x)
    return (n * sum_xy - sum_x * sum_y) / (n * sum(xi**2 for xi in x) - sum_x**2)


@task
def intercept(x: typing.List[int], y: typing.List[int], slope: float) -> float:
    return (sum(y) - slope * sum(x)) / len(x)


@task
def calc_y(x: int, slope: float, intercept: float) -> float:
    return slope * x + intercept
Define the inner workflow (used as a subworkflow): 
@workflow
def slope_intercept_wf(
    x: typing.List[int], y: typing.List[int]
) -> typing.Tuple[float, float]:
    m = slope(x=x, y=y)
    b = intercept(x=x, y=y, slope=m)
    return m, b
Call slope_intercept_wf from a parent workflow: 
@workflow
def regression_line_wf(
    x: typing.List[int] = [-3, 0, 3],
    y: typing.List[int] = [7, 4, -2],
    val: int = 5,
) -> float:
    m, b = slope_intercept_wf(x=x, y=y)
    return calc_y(x=val, slope=m, intercept=b)
When regression_line_wf runs, Flyte inlines slope_intercept_wf’s entire DAG into regression_line_wf at the point of invocation. Run locally: 
if __name__ == "__main__":
    print(regression_line_wf())

​
Nesting subworkflows

You can nest workflows arbitrarily. Each workflow remains independently executable as a standalone entity: 
@workflow
def nested_regression_line_wf(
    x: typing.List[int] = [-3, 0, 3],
    y: typing.List[int] = [7, 4, -2],
    val: int = 5,
) -> float:
    # regression_line_wf itself contains a subworkflow (slope_intercept_wf)
    return regression_line_wf(x=x, y=y, val=val)

if __name__ == "__main__":
    print(nested_regression_line_wf())
Workflows composed as subworkflows can still be launched independently. They are reusable building blocks, not tightly coupled to a parent.

​
External workflows with launch plans

When you use a launch plan inside a workflow, Flyte triggers a new, independent execution with its own execution ID. These are called external workflows. 
from flytekit import LaunchPlan

# Create a launch plan from the existing workflow
slope_intercept_lp = LaunchPlan.get_or_create(
    slope_intercept_wf,
    name="slope_intercept_lp",
)


@workflow
def nested_regression_line_lp(
    x: typing.List[int] = [-3, 0, 3],
    y: typing.List[int] = [7, 4, -2],
    val: int = 5,
) -> float:
    # Triggers a separate execution — a new execution ID is created
    m, b = slope_intercept_lp(x=x, y=y)
    return calc_y(x=val, slope=m, intercept=b)
In the Flyte console, the launch plan execution shows a different execution ID than the parent workflow.
If your deployment uses multiple Kubernetes clusters, external workflows via launch plans can distribute workloads across clusters — each external execution can run on a different cluster.
Run locally: 
if __name__ == "__main__":
    print(nested_regression_line_lp())

​
Run on the Flyte cluster

pyflyte run --remote \
  https://raw.githubusercontent.com/flyteorg/flytesnacks/69dbe4840031a85d79d9ded25f80397c6834752d/examples/advanced_composition/advanced_composition/subworkflow.py \
  regression_line_wf

Build docs developers (and LLMs) love

Get started for freeTalk to us