Branching Workflow

Complex decision trees requiring dynamic path selection based on content analysis

Example Use-Cases: Expert routing, content type detection, multi-path processing

Dynamic routing workflows provide intelligent path selection while maintaining predictable execution within each chosen branch.

Selector Flexibility

The Router selector function supports multiple return types:

String: Return step name - Router resolves it from choices
Step: Return Step object directly
List[Step]: Return list of steps for chaining

The selector can also receive step_choices as an optional second parameter for dynamic selection.

Example: String-based Selector

The simplest approach - return the step name as a string:

1from typing import Union, List
2
3from kern.agent import Agent
4from kern.models.openai import OpenAIChat
5from kern.workflow.router import Router
6from kern.workflow.step import Step
7from kern.workflow.types import StepInput
8from kern.workflow.workflow import Workflow
9
10tech_expert = Agent(
11    name="tech_expert",
12    model=OpenAIChat(id="gpt-4o-mini"),
13    instructions="You are a tech expert. Provide technical analysis.",
14)
15
16biz_expert = Agent(
17    name="biz_expert",
18    model=OpenAIChat(id="gpt-4o-mini"),
19    instructions="You are a business expert. Provide business insights.",
20)
21
22generalist = Agent(
23    name="generalist",
24    model=OpenAIChat(id="gpt-4o-mini"),
25    instructions="You are a generalist. Provide general information.",
26)
27
28tech_step = Step(name="Tech Research", agent=tech_expert)
29business_step = Step(name="Business Research", agent=biz_expert)
30general_step = Step(name="General Research", agent=generalist)
31
32
33def route_by_topic(step_input: StepInput) -> Union[str, Step, List[Step]]:
34    """Selector can return step name as string - Router resolves it."""
35    topic = step_input.input.lower()
36
37    if "tech" in topic or "ai" in topic or "software" in topic:
38        return "Tech Research"  # Return name as string
39    elif "business" in topic or "market" in topic or "finance" in topic:
40        return "Business Research"
41    else:
42        return "General Research"
43
44
45workflow = Workflow(
46    name="Expert Routing",
47    steps=[
48        Router(
49            name="Topic Router",
50            selector=route_by_topic,
51            choices=[tech_step, business_step, general_step],
52        ),
53    ],
54)
55
56workflow.print_response("Latest developments in artificial intelligence", markdown=True)

Example: Using step_choices Parameter

Access available choices dynamically for more flexible routing:

1def dynamic_selector(step_input: StepInput, step_choices: list) -> Union[str, Step, List[Step]]:
2    """
3    Selector receives step_choices - can select by name or return Step directly.
4    step_choices contains the prepared Step objects from Router.choices.
5    """
6    user_input = step_input.input.lower()
7
8    # Build name map from step_choices
9    step_map = {s.name: s for s in step_choices if hasattr(s, "name") and s.name}
10
11    # Can return step name as string
12    if "research" in user_input:
13        return "researcher"
14
15    # Can return Step object directly
16    if "write" in user_input:
17        return step_map.get("writer", step_choices[0])
18
19    # Can return list of Steps for chaining
20    if "full" in user_input:
21        return [step_map["researcher"], step_map["writer"], step_map["reviewer"]]
22
23    # Default
24    return step_choices[0]
25
26
27workflow = Workflow(
28    name="Dynamic Routing",
29    steps=[
30        Router(
31            name="Dynamic Router",
32            selector=dynamic_selector,
33            choices=[researcher, writer, reviewer],
34        ),
35    ],
36)

Example: Nested Choices

Nested lists in choices become Steps containers for sequential execution:

1def nested_selector(step_input: StepInput, step_choices: list) -> Union[str, Step, List[Step]]:
2    """
3    When choices contains nested lists like [step_a, [step_b, step_c]],
4    the nested list becomes a Steps container in step_choices.
5    """
6    user_input = step_input.input.lower()
7
8    # step_choices[0] = Step for step_a
9    # step_choices[1] = Steps container with [step_b, step_c]
10
11    if "single" in user_input:
12        return step_choices[0]  # Just step_a
13    else:
14        return step_choices[1]  # Steps container: step_b -> step_c
15
16
17workflow = Workflow(
18    name="Nested Choices Routing",
19    steps=[
20        Router(
21            name="Nested Router",
22            selector=nested_selector,
23            choices=[step_a, [step_b, step_c]],  # Nested list becomes Steps container
24        ),
25    ],
26)

Developer Resources

Reference

For complete API documentation, see Router Steps Reference.