Deeply Nested Workflow (3 Levels)

1from kern.agent import Agent
2from kern.models.openai import OpenAIChat
3from kern.workflow import Parallel
4from kern.workflow.step import Step
5from kern.workflow.types import StepInput, StepOutput
6from kern.workflow.workflow import Workflow
7
8
9def merge_results(step_input: StepInput) -> StepOutput:
10    """Merge content from previous steps."""
11    prev = step_input.previous_step_content or ""
12    return StepOutput(content=f"Merged: {prev[:500]}")
13
14
15# Level 3: Mini-workflows for individual research tasks
16data_agent = Agent(
17    name="Data Agent",
18    model=OpenAIChat(id="gpt-4o-mini"),
19    instructions="Gather raw data and statistics on the topic. Be concise (2-3 sentences).",
20)
21
22analysis_agent = Agent(
23    name="Analysis Agent",
24    model=OpenAIChat(id="gpt-4o-mini"),
25    instructions="Analyze the data provided. Identify key trends. Be concise (2-3 sentences).",
26)
27
28data_workflow = Workflow(
29    name="Data Collection",
30    description="Collects and analyzes raw data",
31    steps=[
32        Step(name="gather", agent=data_agent),
33        Step(name="analyze", agent=analysis_agent),
34    ],
35)
36
37opinion_agent = Agent(
38    name="Opinion Agent",
39    model=OpenAIChat(id="gpt-4o-mini"),
40    instructions="Provide expert opinion and perspective on the topic. Be concise (2-3 sentences).",
41)
42
43opinion_workflow = Workflow(
44    name="Expert Opinion",
45    description="Gathers expert perspectives",
46    steps=[Step(name="opinion", agent=opinion_agent)],
47)
48
49# Level 2: Research workflow with parallel Level 3 workflows
50level2_workflow = Workflow(
51    name="Comprehensive Research",
52    description="Runs data collection and expert opinion in parallel",
53    steps=[
54        Parallel(
55            Step(name="data_branch", workflow=data_workflow),
56            Step(name="opinion_branch", workflow=opinion_workflow),
57            name="parallel_research",
58        ),
59        Step(name="merge", executor=merge_results),
60    ],
61)
62
63# Level 1: Outermost workflow
64writer = Agent(
65    name="Writer",
66    model=OpenAIChat(id="gpt-4o-mini"),
67    instructions="Write a polished short paragraph synthesizing all research provided.",
68)
69
70outer_workflow = Workflow(
71    name="Full Pipeline",
72    description="3-level nested workflow: research (parallel mini-workflows) -> write",
73    steps=[
74        Step(name="research", workflow=level2_workflow),
75        Step(name="write", agent=writer),
76    ],
77)
78
79
80if __name__ == "__main__":
81    outer_workflow.print_response(
82        input="What is the future of renewable energy?",
83        stream=True,
84    )

1git clone https://github.com/kern-ai/kern.git
2cd kern/cookbook/04_workflows/06_advanced_concepts/workflow_as_a_step
3
4pip install kern-ai openai
5
6python deeply_nested_workflow.py