Condition with list of steps

This example demonstrates how to use conditional step to execute multiple steps in parallel.

This example demonstrates Workflows 2.0 advanced conditional execution where conditions can trigger multiple steps and run in parallel. Shows how to create sophisticated branching logic with complex multi-step sequences based on content analysis.

When to use: When different topics or content types require completely different processing pipelines. Ideal for adaptive workflows where the research methodology should change based on the subject matter or complexity requirements.

1from kern.agent.agent import Agent
2from kern.tools.hackernews import HackerNewsTools
3from kern.tools.yfinance import YFinanceTools
4from kern.workflow.condition import Condition
5from kern.workflow.parallel import Parallel
6from kern.workflow.step import Step
7from kern.workflow.types import StepInput
8from kern.workflow.workflow import Workflow
9
10# === AGENTS ===
11hackernews_agent = Agent(
12    name="HackerNews Researcher",
13    instructions="Research tech news and trends from Hacker News",
14    tools=[HackerNewsTools()],
15)
16
17finance_agent = Agent(
18    name="Finance Researcher",
19    instructions="Research financial data and market trends",
20    tools=[YFinanceTools()],
21)
22
23content_agent = Agent(
24    name="Content Creator",
25    instructions="Create well-structured content from research data",
26)
27
28# Additional agents for multi-step condition
29trend_analyzer_agent = Agent(
30    name="Trend Analyzer",
31    instructions="Analyze trends and patterns from research data",
32)
33
34fact_checker_agent = Agent(
35    name="Fact Checker",
36    instructions="Verify facts and cross-reference information",
37)
38
39# === RESEARCH STEPS ===
40research_hackernews_step = Step(
41    name="ResearchHackerNews",
42    description="Research tech news from Hacker News",
43    agent=hackernews_agent,
44)
45
46research_finance_step = Step(
47    name="ResearchFinance",
48    description="Research financial data",
49    agent=finance_agent,
50)
51
52# === MULTI-STEP CONDITION STEPS ===
53deep_finance_analysis_step = Step(
54    name="DeepFinanceAnalysis",
55    description="Conduct deep analysis of financial data",
56    agent=finance_agent,
57)
58
59trend_analysis_step = Step(
60    name="TrendAnalysis",
61    description="Analyze trends and patterns from the research data",
62    agent=trend_analyzer_agent,
63)
64
65fact_verification_step = Step(
66    name="FactVerification",
67    description="Verify facts and cross-reference information",
68    agent=fact_checker_agent,
69)
70
71# === FINAL STEPS ===
72write_step = Step(
73    name="WriteContent",
74    description="Write the final content based on research",
75    agent=content_agent,
76)
77
78
79# === CONDITION EVALUATORS ===
80def check_if_we_should_search_hn(step_input: StepInput) -> bool:
81    """Check if we should search Hacker News"""
82    topic = step_input.input or step_input.previous_step_content or ""
83    tech_keywords = [
84        "ai",
85        "machine learning",
86        "programming",
87        "software",
88        "tech",
89        "startup",
90        "coding",
91    ]
92    return any(keyword in topic.lower() for keyword in tech_keywords)
93
94
95def check_if_comprehensive_research_needed(step_input: StepInput) -> bool:
96    """Check if comprehensive multi-step research is needed"""
97    topic = step_input.input or step_input.previous_step_content or ""
98    comprehensive_keywords = [
99        "comprehensive",
100        "detailed",
101        "thorough",
102        "in-depth",
103        "complete analysis",
104        "full report",
105        "extensive research",
106    ]
107    return any(keyword in topic.lower() for keyword in comprehensive_keywords)
108
109
110if __name__ == "__main__":
111    workflow = Workflow(
112        name="Conditional Workflow with Multi-Step Condition",
113        steps=[
114            Parallel(
115                Condition(
116                    name="HackerNewsCondition",
117                    description="Check if we should search Hacker News for tech topics",
118                    evaluator=check_if_we_should_search_hn,
119                    steps=[research_hackernews_step],  # Single step
120                ),
121                Condition(
122                    name="ComprehensiveResearchCondition",
123                    description="Check if comprehensive multi-step research is needed",
124                    evaluator=check_if_comprehensive_research_needed,
125                    steps=[  # Multiple steps
126                        deep_finance_analysis_step,
127                        trend_analysis_step,
128                        fact_verification_step,
129                    ],
130                ),
131                name="ConditionalResearch",
132                description="Run conditional research steps in parallel",
133            ),
134            write_step,
135        ],
136    )
137
138    try:
139        workflow.print_response(
140            input="Comprehensive analysis of climate change research",
141            stream=True,
142        )
143    except Exception as e:
144        print(f"Error: {e}")
145    print()