Condition and Parallel Steps Workflow

This example demonstrates **Workflows 2.0** advanced pattern combining conditional execution with parallel processing.

This example shows how to create sophisticated workflows where multiple conditions evaluate simultaneously, each potentially triggering different research strategies based on comprehensive content analysis.

When to use: When you need comprehensive, multi-dimensional content analysis where different aspects of the input may trigger different specialized research pipelines simultaneously. Ideal for adaptive research workflows that can leverage multiple sources based on various content characteristics.

1from kern.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# === RESEARCH STEPS ===
29research_hackernews_step = Step(
30    name="ResearchHackerNews",
31    description="Research tech news from Hacker News",
32    agent=hackernews_agent,
33)
34
35research_finance_step = Step(
36    name="ResearchFinance",
37    description="Research financial data and trends",
38    agent=finance_agent,
39)
40
41prepare_input_for_write_step = Step(
42    name="PrepareInput",
43    description="Prepare and organize research data for writing",
44    agent=content_agent,
45)
46
47write_step = Step(
48    name="WriteContent",
49    description="Write the final content based on research",
50    agent=content_agent,
51)
52
53
54# === CONDITION EVALUATORS ===
55def check_if_we_should_search_hn(step_input: StepInput) -> bool:
56    """Check if we should search Hacker News"""
57    topic = step_input.input or step_input.previous_step_content or ""
58    tech_keywords = [
59        "ai",
60        "machine learning",
61        "programming",
62        "software",
63        "tech",
64        "startup",
65        "coding",
66    ]
67    return any(keyword in topic.lower() for keyword in tech_keywords)
68
69
70def check_if_we_should_search_finance(step_input: StepInput) -> bool:
71    """Check if we should search for financial data"""
72    topic = step_input.input or step_input.previous_step_content or ""
73    finance_keywords = ["stock", "market", "finance", "investment", "trading", "price"]
74    return any(keyword in topic.lower() for keyword in finance_keywords)
75
76
77if __name__ == "__main__":
78    workflow = Workflow(
79        name="Conditional Workflow",
80        steps=[
81            Parallel(
82                Condition(
83                    name="HackerNewsCondition",
84                    description="Check if we should search Hacker News for tech topics",
85                    evaluator=check_if_we_should_search_hn,
86                    steps=[research_hackernews_step],
87                ),
88                Condition(
89                    name="FinanceCondition",
90                    description="Check if we should search for financial data",
91                    evaluator=check_if_we_should_search_finance,
92                    steps=[research_finance_step],
93                ),
94                name="ConditionalResearch",
95                description="Run conditional research steps in parallel",
96            ),
97            prepare_input_for_write_step,
98            write_step,
99        ],
100    )
101
102    try:
103        workflow.print_response(
104            input="Latest AI developments in machine learning",
105            stream=True,
106        )
107    except Exception as e:
108        print(f"Error: {e}")
109    print()