Conditional Workflow

Deterministic branching based on input analysis or business rules

Example Use-Cases: Content type routing, topic-specific processing, quality-based decisions

Conditional workflows provide predictable branching logic while maintaining deterministic execution paths.

How It Works

The Condition class evaluates a function and executes different steps based on the result:

If branch (steps): Executes when the evaluator returns True
Else branch (else_steps): Executes when the evaluator returns False (optional)

If the condition is False and no else_steps are provided, the condition is skipped and the workflow continues to the next step.

Basic Example

1from kern.workflow import Condition, Step, Workflow
2
3def is_tech_topic(step_input) -> bool:
4    topic = step_input.input.lower()
5    return any(keyword in topic for keyword in ["ai", "tech", "software"])
6
7workflow = Workflow(
8    name="Conditional Research",
9    steps=[
10        Condition(
11            name="Tech Topic Check",
12            evaluator=is_tech_topic,
13            steps=[Step(name="Tech Research", agent=tech_researcher)]
14        ),
15        Step(name="General Analysis", agent=general_analyst),
16    ]
17)
18
19workflow.print_response("Comprehensive analysis of AI and machine learning trends", markdown=True)

If/Else Branching

Use else_steps to define an alternative execution path when the condition is False:

1from kern.workflow import Condition, Step, Workflow
2
3def is_technical_issue(step_input) -> bool:
4    text = (step_input.input or "").lower()
5    tech_keywords = ["error", "bug", "crash", "not working", "api", "timeout"]
6    return any(kw in text for kw in tech_keywords)
7
8workflow = Workflow(
9    name="Customer Support Router",
10    steps=[
11        Condition(
12            name="TechnicalTriage",
13            evaluator=is_technical_issue,
14            # If branch: technical pipeline
15            steps=[
16                Step(name="Diagnose", agent=diagnostic_agent),
17                Step(name="Engineer", agent=engineering_agent),
18            ],
19            # Else branch: general support
20            else_steps=[
21                Step(name="GeneralSupport", agent=general_support_agent),
22            ],
23        ),
24        Step(name="FollowUp", agent=followup_agent),
25    ],
26)
27
28# Technical query -> executes Diagnose, Engineer, then FollowUp
29workflow.print_response("My app keeps crashing with a timeout error")
30
31# Non-technical query -> executes GeneralSupport, then FollowUp
32workflow.print_response("How do I change my shipping address?")

Developer Resources

Reference

For complete API documentation, see Condition Steps Reference.