Dynamic Selector

1from typing import List, Union
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
10researcher = Agent(
11    name="researcher",
12    model=OpenAIChat(id="gpt-4o-mini"),
13    instructions="You are a researcher.",
14)
15
16writer = Agent(
17    name="writer",
18    model=OpenAIChat(id="gpt-4o-mini"),
19    instructions="You are a writer.",
20)
21
22reviewer = Agent(
23    name="reviewer",
24    model=OpenAIChat(id="gpt-4o-mini"),
25    instructions="You are a reviewer.",
26)
27
28
29def dynamic_selector(step_input: StepInput, step_choices: list) -> Union[str, Step, List[Step]]:
30    """
31    Selector receives step_choices - can select by name or return Step directly.
32    step_choices contains the prepared Step objects from Router.choices.
33    """
34    user_input = step_input.input.lower()
35
36    # Build name map from step_choices
37    step_map = {s.name: s for s in step_choices if hasattr(s, "name") and s.name}
38
39    print(f"Available steps: {list(step_map.keys())}")
40
41    # Can return step name as string
42    if "research" in user_input:
43        return "researcher"
44
45    # Can return Step object directly
46    if "write" in user_input:
47        return step_map.get("writer", step_choices[0])
48
49    # Can return list of Steps for chaining
50    if "full" in user_input:
51        return [step_map["researcher"], step_map["writer"], step_map["reviewer"]]
52
53    # Default
54    return step_choices[0]
55
56
57workflow = Workflow(
58    name="Dynamic Routing (step_choices)",
59    steps=[
60        Router(
61            name="Dynamic Router",
62            selector=dynamic_selector,
63            choices=[researcher, writer, reviewer],
64        ),
65    ],
66)
67
68if __name__ == "__main__":
69    workflow.print_response("I need to research something", stream=True)