Loop Steps Workflow

This example demonstrates **Workflows 2.0** loop execution for quality-driven iterative processes.

This example demonstrates Workflows 2.0 to repeatedly execute steps until specific conditions are met, ensuring adequate research depth before proceeding to content creation.

When to use: When you need iterative refinement, quality assurance, or when the required output quality can't be guaranteed in a single execution. Ideal for research gathering, data collection, or any process where "good enough" is determined by content analysis rather than a fixed number of iterations.

1from typing import List
2
3from kern.agent import Agent
4from kern.tools.hackernews import HackerNewsTools
5from kern.tools.yfinance import YFinanceTools
6from kern.workflow import Loop, Step, Workflow
7from kern.workflow.types import StepOutput
8
9# Create agents for research
10research_agent = Agent(
11    name="Research Agent",
12    role="Research specialist",
13    tools=[HackerNewsTools(), YFinanceTools()],
14    instructions="You are a research specialist. Research the given topic thoroughly.",
15    markdown=True,
16)
17
18content_agent = Agent(
19    name="Content Agent",
20    role="Content creator",
21    instructions="You are a content creator. Create engaging content based on research.",
22    markdown=True,
23)
24
25# Create research steps
26research_hackernews_step = Step(
27    name="Research HackerNews",
28    agent=research_agent,
29    description="Research trending topics on HackerNews",
30)
31
32research_web_step = Step(
33    name="Research Web",
34    agent=research_agent,
35    description="Research additional information from web sources",
36)
37
38content_step = Step(
39    name="Create Content",
40    agent=content_agent,
41    description="Create content based on research findings",
42)
43
44
45# End condition function
46def research_evaluator(outputs: List[StepOutput]) -> bool:
47    """
48    Evaluate if research results are sufficient
49    Returns True to break the loop, False to continue
50    """
51    # Check if any outputs are present
52    if not outputs:
53        return False
54
55    # Check if any output contains substantial content
56    for output in outputs:
57        if output.content and len(output.content) > 200:
58            print(
59                f"Research evaluation passed - found substantial content ({len(output.content)} chars)"
60            )
61            return True
62
63    print("Research evaluation failed - need more substantial research")
64    return False
65
66
67# Create workflow with loop
68workflow = Workflow(
69    name="Research and Content Workflow",
70    description="Research topics in a loop until conditions are met, then create content",
71    steps=[
72        Loop(
73            name="Research Loop",
74            steps=[research_hackernews_step, research_web_step],
75            end_condition=research_evaluator,
76            max_iterations=3,  # Maximum 3 iterations
77        ),
78        content_step,
79    ],
80)
81
82if __name__ == "__main__":
83    # Test the workflow
84    workflow.print_response(
85        input="Research the latest trends in AI and machine learning, then create a summary",
86    )