Router with Loop Steps

1from typing import List
2
3from kern.agent.agent import Agent
4from kern.tools.hackernews import HackerNewsTools
5from kern.tools.yfinance import YFinanceTools
6from kern.workflow.loop import Loop
7from kern.workflow.router import Router
8from kern.workflow.step import Step
9from kern.workflow.types import StepInput, StepOutput
10from kern.workflow.workflow import Workflow
11
12# Define the research agents
13hackernews_agent = Agent(
14    name="HackerNews Researcher",
15    instructions="You are a researcher specializing in finding the latest tech news and discussions from Hacker News. Focus on startup trends, programming topics, and tech industry insights.",
16    tools=[HackerNewsTools()],
17)
18
19finance_agent = Agent(
20    name="Finance Researcher",
21    instructions="You are a finance researcher. Search for stock data, market trends, and financial information to gather detailed insights.",
22    tools=[YFinanceTools()],
23)
24
25content_agent = Agent(
26    name="Content Publisher",
27    instructions="You are a content creator who takes research data and creates engaging, well-structured articles. Format the content with proper headings, bullet points, and clear conclusions.",
28)
29
30# Create the research steps
31research_hackernews = Step(
32    name="research_hackernews",
33    agent=hackernews_agent,
34    description="Research latest tech trends from Hacker News",
35)
36
37research_finance = Step(
38    name="research_finance",
39    agent=finance_agent,
40    description="Research financial data and market trends",
41)
42
43publish_content = Step(
44    name="publish_content",
45    agent=content_agent,
46    description="Create and format final content for publication",
47)
48
49# End condition function for the loop
50
51
52def research_quality_check(outputs: List[StepOutput]) -> bool:
53    """
54    Evaluate if research results are sufficient
55    Returns True to break the loop, False to continue
56    """
57    if not outputs:
58        return False
59
60    # Check if any output contains substantial content
61    for output in outputs:
62        if output.content and len(output.content) > 300:
63            print(
64                f"Research quality check passed - found substantial content ({len(output.content)} chars)"
65            )
66            return True
67
68    print("Research quality check failed - need more substantial research")
69    return False
70
71
72# Create a Loop step for deep tech research
73deep_tech_research_loop = Loop(
74    name="Deep Tech Research Loop",
75    steps=[research_hackernews],
76    end_condition=research_quality_check,
77    max_iterations=3,
78    description="Perform iterative deep research on tech topics",
79)
80
81# Router function that selects between simple web research or deep tech research loop
82
83
84def research_strategy_router(step_input: StepInput) -> List[Step]:
85    """
86    Decide between simple web research or deep tech research loop based on the input topic.
87    Returns either a single web research step or a tech research loop.
88    """
89    # Use the original workflow message if this is the first step
90    topic = step_input.previous_step_content or step_input.input or ""
91    topic = topic.lower()
92
93    # Check if the topic requires deep tech research
94    deep_tech_keywords = [
95        "startup trends",
96        "ai developments",
97        "machine learning research",
98        "programming languages",
99        "developer tools",
100        "silicon valley",
101        "venture capital",
102        "cryptocurrency analysis",
103        "blockchain technology",
104        "open source projects",
105        "github trends",
106        "tech industry",
107        "software engineering",
108        "writers",
109    ]
110
111    # Check if it's a complex tech topic that needs deep research
112    if any(keyword in topic for keyword in deep_tech_keywords) or (
113        "tech" in topic and len(topic.split()) > 3
114    ):
115        print(
116            f"Deep tech topic detected: Using iterative research loop for '{topic}'"
117        )
118        return [deep_tech_research_loop]
119    else:
120        print(f"Finance topic detected: Using finance research for '{topic}'")
121        return [research_finance]
122
123
124workflow = Workflow(
125    name="Adaptive Research Workflow",
126    description="Intelligently selects between simple web research or deep iterative tech research based on topic complexity",
127    steps=[
128        Router(
129            name="research_strategy_router",
130            selector=research_strategy_router,
131            choices=[research_finance, deep_tech_research_loop],
132            description="Chooses between simple web research or deep tech research loop",
133        ),
134        publish_content,
135    ],
136)
137
138if __name__ == "__main__":
139    print("=== Testing with deep tech topic ===")
140    workflow.print_response(
141        "Latest developments in artificial intelligence and machine learning and deep tech research trends"
142    )