workflow_with_router.py
"""
Workflow With Router
====================
Demonstrates workflow with router.
"""
from typing import List
from agno.agent.agent import Agent
from agno.db.sqlite import SqliteDb
# ---------------------------------------------------------------------------
# Create Example
# ---------------------------------------------------------------------------
# Import the workflows
from agno.os import AgentOS
from agno.tools.hackernews import HackerNewsTools
from agno.tools.websearch import WebSearchTools
from agno.workflow.router import Router
from agno.workflow.step import Step
from agno.workflow.types import StepInput
from agno.workflow.workflow import Workflow
# Define the research agents
hackernews_agent = Agent(
name="HackerNews Researcher",
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.",
tools=[HackerNewsTools()],
)
web_agent = Agent(
name="Web Researcher",
instructions="You are a comprehensive web researcher. Search across multiple sources including news sites, blogs, and official documentation to gather detailed information.",
tools=[WebSearchTools()],
)
content_agent = Agent(
name="Content Publisher",
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.",
)
# Create the research steps
research_hackernews = Step(
name="research_hackernews",
agent=hackernews_agent,
description="Research latest tech trends from Hacker News",
)
research_web = Step(
name="research_web",
agent=web_agent,
description="Comprehensive web research on the topic",
)
publish_content = Step(
name="publish_content",
agent=content_agent,
description="Create and format final content for publication",
)
# Now returns Step(s) to execute
def research_router(step_input: StepInput) -> List[Step]:
"""
Decide which research method to use based on the input topic.
Returns a list containing the step(s) to execute.
"""
# Use the original workflow message if this is the first step
topic = step_input.previous_step_content or step_input.input or ""
topic = topic.lower()
# Check if the topic is tech/startup related - use HackerNews
tech_keywords = [
"startup",
"programming",
"ai",
"machine learning",
"software",
"developer",
"coding",
"tech",
"silicon valley",
"venture capital",
"cryptocurrency",
"blockchain",
"open source",
"github",
]
if any(keyword in topic for keyword in tech_keywords):
print(f"Tech topic detected: Using HackerNews research for '{topic}'")
return [research_hackernews]
else:
print(f"General topic detected: Using web research for '{topic}'")
return [research_web]
workflow = Workflow(
name="intelligent-research-workflow",
description="Automatically selects the best research method based on topic, then publishes content",
steps=[
Router(
name="research_strategy_router",
selector=research_router,
choices=[research_hackernews, research_web],
description="Intelligently selects research method based on topic",
),
publish_content,
],
db=SqliteDb(
session_table="workflow_session",
db_file="tmp/workflow.db",
),
)
# Initialize the AgentOS with the workflows
agent_os = AgentOS(
description="Example OS setup",
workflows=[workflow],
)
app = agent_os.get_app()
# ---------------------------------------------------------------------------
# Run Example
# ---------------------------------------------------------------------------
if __name__ == "__main__":
agent_os.serve(app="workflow_with_router:app", reload=True)
Run the Example
1
Set up your virtual environment
uv venv --python 3.12
source .venv/bin/activate
uv venv --python 3.12
.venv\Scripts\activate
2
Install dependencies
uv pip install -U "agno[os]" ddgs openai
3
Export your OpenAI API key
export OPENAI_API_KEY="your_openai_api_key_here"
$Env:OPENAI_API_KEY="your_openai_api_key_here"
4
Run the example
Save the code above as
workflow_with_router.py, then run:python workflow_with_router.py