Grouped Steps Workflow

Organize multiple steps into reusable, logical sequences for complex workflows with clean separation of concerns

Key Benefits: Reusable sequences, cleaner branching logic, modular workflow design

Grouped steps enable modular workflow architecture with reusable components and clear logical boundaries.

Basic Example

1from kern.workflow import Steps, Step, Workflow
2
3# Create a reusable content creation sequence
4article_creation_sequence = Steps(
5    name="ArticleCreation",
6    description="Complete article creation workflow from research to final edit",
7    steps=[
8        Step(name="research", agent=researcher),
9        Step(name="writing", agent=writer),
10        Step(name="editing", agent=editor),
11    ],
12)
13
14# Use the sequence in a workflow
15workflow = Workflow(
16    name="Article Creation Workflow",
17    steps=[article_creation_sequence]  # Single sequence
18)
19
20workflow.print_response("Write an article about renewable energy", markdown=True)

Steps with Router

This is where Steps really shines - creating distinct sequences for different content types or workflows:

1from kern.workflow import Steps, Router, Step, Workflow
2
3# Define two completely different workflows as Steps
4image_sequence = Steps(
5    name="image_generation",
6    description="Complete image generation and analysis workflow",
7    steps=[
8        Step(name="generate_image", agent=image_generator),
9        Step(name="describe_image", agent=image_describer),
10    ],
11)
12
13video_sequence = Steps(
14    name="video_generation",
15    description="Complete video production and analysis workflow",
16    steps=[
17        Step(name="generate_video", agent=video_generator),
18        Step(name="describe_video", agent=video_describer),
19    ],
20)
21
22def media_sequence_selector(step_input) -> List[Step]:
23    """Route to appropriate media generation pipeline"""
24    if not step_input.input:
25        return [image_sequence]
26
27    message_lower = step_input.input.lower()
28
29    if "video" in message_lower:
30        return [video_sequence]
31    elif "image" in message_lower:
32        return [image_sequence]
33    else:
34        return [image_sequence]  # Default
35
36# Clean workflow with clear branching
37media_workflow = Workflow(
38    name="AI Media Generation Workflow",
39    description="Generate and analyze images or videos using AI agents",
40    steps=[
41        Router(
42            name="Media Type Router",
43            description="Routes to appropriate media generation pipeline",
44            selector=media_sequence_selector,
45            choices=[image_sequence, video_sequence],  # Clear choices
46        )
47    ],
48)
49
50# Usage examples
51media_workflow.print_response("Create an image of a magical forest", markdown=True)
52media_workflow.print_response("Create a cinematic video of city timelapse", markdown=True)

Developer Resources

Reference

For complete API documentation, see Steps Reference.