Nested Workflow

Run a workflow as a step inside another workflow for complex multi-level pipelines.

Pass a Workflow as a step inside another Workflow. The inner workflow runs as a single step in the outer workflow, with output chained to the next step.

Basic Example

1from kern.agent import Agent
2from kern.models.openai import OpenAIChat
3from kern.workflow.step import Step
4from kern.workflow.types import StepInput, StepOutput
5from kern.workflow.workflow import Workflow
6
7
8def create_summary(step_input: StepInput) -> StepOutput:
9    previous_content = step_input.get_last_step_content()
10    summary = (
11        f"Summary of research:\n{previous_content[:500]}..."
12        if previous_content
13        else "No content to summarize"
14    )
15    return StepOutput(content=summary)
16
17
18# Inner workflow: research pipeline
19research_agent = Agent(
20    name="Research Agent",
21    model=OpenAIChat(id="gpt-4o-mini"),
22    instructions="You are a research assistant. Provide concise, factual information.",
23)
24
25inner_workflow = Workflow(
26    name="Research Workflow",
27    steps=[
28        Step(name="research", agent=research_agent),
29        Step(name="summary", executor=create_summary),
30    ],
31)
32
33# Outer workflow: uses inner workflow as a step
34writer_agent = Agent(
35    name="Writer Agent",
36    model=OpenAIChat(id="gpt-4o-mini"),
37    instructions="Take the research provided and write a polished article.",
38)
39
40outer_workflow = Workflow(
41    name="Research and Write Workflow",
42    steps=[
43        Step(name="research_phase", workflow=inner_workflow),
44        Step(name="writing_phase", agent=writer_agent),
45    ],
46)
47
48outer_workflow.print_response(
49    input="Tell me about the history of artificial intelligence",
50    stream=True,
51)

The outer workflow runs inner_workflow as its first step. The inner workflow's output flows into the writing_phase step.

How It Works

The outer workflow reaches a step with workflow=inner_workflow
The inner workflow's .run() executes with the prepared input (chained from the previous step)
Session state is deep-copied into the inner workflow and merged back after execution
The inner workflow's output is converted to a StepOutput with step_type=StepType.WORKFLOW
Execution continues to the next step in the outer workflow

Two Ways to Declare

Method	Syntax	When to use
Explicit `Step` wrapper	`Step(name="research", workflow=inner_workflow)`	Custom step name, clarity
Auto-wrap	`steps=[inner_workflow]`	Concise shorthand (uses the workflow's `name` as step name)

1# These are equivalent:
2steps=[Step(name="Research Workflow", workflow=inner_workflow)]
3steps=[inner_workflow]

Inner workflows and primitives

An inner workflow can use the same primitives and combinations as any top-level workflow in Kern: agents, executors, nested Steps, Condition, Loop, Router, and Parallel, mixed however your pipeline needs.

The basic example uses agent and executor steps inside the inner workflow. Deep nesting shows multiple levels with Parallel and sub-workflows.

Primitive	Role
`Condition`	Branch on a boolean evaluator
`Loop`	Repeat steps until an end condition or max iterations
`Router`	Choose a branch from a selector
`Parallel`	Run branches concurrently

Deep Nesting

Workflows can be nested multiple levels deep. Each level runs its own sub-workflows independently.

1from kern.workflow import Parallel
2
3# Level 3: Mini-workflows
4data_workflow = Workflow(
5    name="Data Collection",
6    steps=[
7        Step(name="gather", agent=data_agent),
8        Step(name="analyze", agent=analysis_agent),
9    ],
10)
11
12opinion_workflow = Workflow(
13    name="Expert Opinion",
14    steps=[Step(name="opinion", agent=opinion_agent)],
15)
16
17# Level 2: Parallel research with Level 3 workflows
18level2_workflow = Workflow(
19    name="Comprehensive Research",
20    steps=[
21        Parallel(
22            Step(name="data_branch", workflow=data_workflow),
23            Step(name="opinion_branch", workflow=opinion_workflow),
24            name="parallel_research",
25        ),
26        Step(name="merge", executor=merge_results),
27    ],
28)
29
30# Level 1: Top-level pipeline
31outer_workflow = Workflow(
32    name="Full Pipeline",
33    steps=[
34        Step(name="research", workflow=level2_workflow),
35        Step(name="write", agent=writer),
36    ],
37)

Streaming Events

When streaming, inner workflow events bubble up with a nested_depth field. Use this to distinguish inner vs. outer events.

Field	Description
`nested_depth`	`0` for outer workflow, `1` for first-level inner, `2` for deeper nesting
`workflow_id`	Unique ID of the workflow that emitted the event
`workflow_name`	Name of the workflow that emitted the event

1from kern.run.workflow import (
2    StepCompletedEvent,
3    StepStartedEvent,
4    WorkflowCompletedEvent,
5    WorkflowStartedEvent,
6)
7
8for event in outer_workflow.run(input="...", stream=True, stream_events=True):
9    if isinstance(event, (WorkflowStartedEvent, StepStartedEvent)):
10        depth = event.nested_depth
11        name = event.workflow_name
12        print(f"{'  ' * depth}[depth={depth}] {type(event).__name__} from {name}")

Developer Resources

Reference

For complete API documentation, see Step Reference.