Workflow Examples

from agno.agent import Agent
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow
from agno.models.openai import OpenAIChat
from agno.tools.websearch import WebSearchTools
from agno.team import Team

# Step 1: Research agent
research_agent = Agent(
    name="Researcher",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[WebSearchTools()],
    role="Search the web for the latest news and trends",
)

# Step 2: Content planner
content_planner = Agent(
    name="Content Planner",
    model=OpenAIChat(id="gpt-4o"),
    instructions=[
        "Plan a content schedule over 4 weeks for the provided topic",
        "Ensure 3 posts per week",
    ],
)

# Step 3: Writer
writer = Agent(
    name="Writer",
    model=OpenAIChat(id="gpt-4o-mini"),
    instructions="Write a blog post on the topic",
)

# Create workflow
content_workflow = Workflow(
    name="Content Creation Workflow",
    description="Research → Plan → Write",
    steps=[
        Step(name="Research", agent=research_agent),
        Step(name="Planning", agent=content_planner),
        Step(name="Writing", agent=writer),
    ],
)

if __name__ == "__main__":
    # Sync
    content_workflow.print_response(
        input="AI trends in 2024",
        markdown=True,
    )
    
    # Async
    import asyncio
    asyncio.run(
        content_workflow.aprint_response(
            input="AI agent frameworks 2025",
            markdown=True,
            stream=True,
        )
    )

from agno.workflow.step import Step
from agno.workflow.types import StepInput, StepOutput
from agno.workflow.workflow import Workflow
from typing import AsyncIterator
from textwrap import dedent

# Function step: Prepare input
async def prepare_research_input(
    step_input: StepInput,
) -> AsyncIterator[StepOutput]:
    topic = step_input.input
    content = dedent(
        f"""
        Research the following topic:
        <topic>
        {topic}
        </topic>
        
        Find at least 10 credible sources.
        """
    )
    yield StepOutput(content=content)

# Function step: Format output
async def format_output(
    step_input: StepInput,
) -> AsyncIterator[StepOutput]:
    research = step_input.previous_step_content
    formatted = dedent(
        f"""
        # Research Summary
        
        {research}
        
        ---
        Generated by Research Workflow
        """
    )
    yield StepOutput(content=formatted)

# Workflow mixing agents and functions
workflow = Workflow(
    name="Research Workflow",
    steps=[
        prepare_research_input,  # Function step
        research_agent,          # Agent step
        format_output,           # Function step
    ],
)

from agno.agent import Agent
from agno.tools.websearch import WebSearchTools
from agno.workflow.condition import Condition
from agno.workflow.step import Step
from agno.workflow.types import StepInput
from agno.workflow.workflow import Workflow

# Create agents
researcher = Agent(
    name="Researcher",
    instructions="Research the given topic and provide detailed findings.",
    tools=[WebSearchTools()],
)

summarizer = Agent(
    name="Summarizer",
    instructions="Create a clear summary of the research findings.",
)

fact_checker = Agent(
    name="Fact Checker",
    instructions="Verify facts and check for accuracy in the research.",
    tools=[WebSearchTools()],
)

writer = Agent(
    name="Writer",
    instructions="Write a comprehensive article based on research and verification.",
)

# Define condition evaluator
def needs_fact_checking(step_input: StepInput) -> bool:
    """Check if content needs fact-checking."""
    summary = step_input.previous_step_content or ""
    fact_indicators = [
        "study shows",
        "research indicates",
        "statistics",
        "data shows",
        "percent",
        "million",
        "billion",
    ]
    return any(indicator in summary.lower() for indicator in fact_indicators)

# Create workflow with condition
workflow = Workflow(
    name="Research Workflow",
    description="Research → Summarize → Condition(Fact Check) → Write",
    steps=[
        Step(name="research", agent=researcher),
        Step(name="summarize", agent=summarizer),
        Condition(
            name="fact_check_gate",
            description="Check if fact-checking is needed",
            evaluator=needs_fact_checking,
            steps=[Step(name="fact_check", agent=fact_checker)],
        ),
        Step(name="write_article", agent=writer),
    ],
)

if __name__ == "__main__":
    workflow.print_response(
        input="Recent breakthroughs in quantum computing",
        stream=True,
    )

from agno.workflow.condition import Condition

def is_technical_topic(step_input: StepInput) -> bool:
    """Determine if topic is technical."""
    topic = step_input.input.lower()
    technical_keywords = [
        "api", "code", "algorithm", "database",
        "programming", "software", "technical"
    ]
    return any(keyword in topic for keyword in technical_keywords)

# Technical writer for code topics
technical_writer = Agent(
    name="Technical Writer",
    instructions="Write technical documentation with code examples.",
)

# General writer for other topics
general_writer = Agent(
    name="General Writer",
    instructions="Write engaging content for general audiences.",
)

workflow = Workflow(
    name="Smart Writing Workflow",
    steps=[
        research_step,
        Condition(
            name="topic_type_router",
            evaluator=is_technical_topic,
            steps=[Step(name="technical_write", agent=technical_writer)],  # If true
            else_steps=[Step(name="general_write", agent=general_writer)],  # If false
        ),
    ],
)

from agno.workflow.step import Step
from agno.workflow.workflow import Workflow
from agno.workflow.parallel import Parallel

# Create specialists
market_analyst = Agent(
    name="Market Analyst",
    tools=[YFinanceTools(all=True)],
)

news_analyst = Agent(
    name="News Analyst",
    tools=[WebSearchTools()],
)

technical_analyst = Agent(
    name="Technical Analyst",
    tools=[YFinanceTools(all=True)],
)

# Synthesizer combines all analysis
synthesizer = Agent(
    name="Synthesizer",
    instructions="Combine market, news, and technical analysis into a report.",
)

workflow = Workflow(
    name="Parallel Analysis Workflow",
    steps=[
        Parallel(
            name="gather_analysis",
            description="Run all analyses in parallel",
            steps=[
                Step(name="market", agent=market_analyst),
                Step(name="news", agent=news_analyst),
                Step(name="technical", agent=technical_analyst),
            ],
        ),
        Step(name="synthesize", agent=synthesizer),
    ],
)

if __name__ == "__main__":
    workflow.print_response(
        input="Analyze Tesla (TSLA) stock",
        stream=True,
    )

from agno.workflow.loop import Loop
from agno.workflow.step import Step
from agno.workflow.types import StepInput

def quality_check(step_input: StepInput) -> bool:
    """Check if content meets quality standards."""
    content = step_input.previous_step_content or ""
    
    # Simple quality checks
    min_length = len(content) > 500
    has_structure = "##" in content  # Has headings
    has_conclusion = "conclusion" in content.lower()
    
    return min_length and has_structure and has_conclusion

writer = Agent(
    name="Writer",
    instructions="Write a comprehensive article. Include structure and conclusion.",
)

reviewer = Agent(
    name="Reviewer",
    instructions="Review the article and suggest improvements.",
)

workflow = Workflow(
    name="Iterative Writing Workflow",
    steps=[
        Loop(
            name="write_review_loop",
            description="Write and review until quality check passes",
            steps=[
                Step(name="write", agent=writer),
                Step(name="review", agent=reviewer),
            ],
            max_iterations=3,
            exit_condition=quality_check,
        ),
    ],
)

import asyncio
from typing import AsyncIterator
from agno.run.workflow import WorkflowRunEvent, WorkflowRunOutputEvent

async def stream_workflow_events():
    events: AsyncIterator[WorkflowRunOutputEvent] = workflow.arun(
        input="AI trends in 2024",
        markdown=True,
        stream=True,
        stream_events=True,
    )
    
    async for event in events:
        if event.event == WorkflowRunEvent.workflow_started.value:
            print(f"Workflow started: {event.workflow_name}")
        
        elif event.event == WorkflowRunEvent.step_started.value:
            print(f"Step started: {event.step_name}")
        
        elif event.event == WorkflowRunEvent.step_completed.value:
            print(f"Step completed: {event.step_name}")
        
        elif event.event == WorkflowRunEvent.condition_execution_started.value:
            print(f"Evaluating condition: {event.condition_name}")
        
        elif event.event == WorkflowRunEvent.workflow_completed.value:
            print(f"Workflow completed: {event.content}")

if __name__ == "__main__":
    asyncio.run(stream_workflow_events())

from agno.workflow.condition import Condition

# CEL expression for condition
workflow = Workflow(
    name="CEL Workflow",
    steps=[
        research_step,
        Condition(
            name="length_check",
            # CEL expression: check content length
            evaluator="len(step_input.previous_step_content) > 1000",
            steps=[Step(name="summarize", agent=summarizer)],
        ),
    ],
)