Parallel With Condition

parallel_with_condition.py

"""
Parallel With Condition
=======================

Demonstrates combining conditional branches with parallel execution for adaptive research pipelines.
"""

import asyncio

from agno.agent import Agent
from agno.tools.exa import ExaTools
from agno.tools.hackernews import HackerNewsTools
from agno.tools.websearch import WebSearchTools
from agno.workflow.condition import Condition
from agno.workflow.parallel import Parallel
from agno.workflow.step import Step
from agno.workflow.types import StepInput
from agno.workflow.workflow import Workflow

# ---------------------------------------------------------------------------
# Create Agents
# ---------------------------------------------------------------------------
hackernews_agent = Agent(
    name="HackerNews Researcher",
    instructions="Research tech news and trends from Hacker News",
    tools=[HackerNewsTools()],
)

web_agent = Agent(
    name="Web Researcher",
    instructions="Research general information from the web",
    tools=[WebSearchTools()],
)

exa_agent = Agent(
    name="Exa Search Researcher",
    instructions="Research using Exa advanced search capabilities",
    tools=[ExaTools()],
)

content_agent = Agent(
    name="Content Creator",
    instructions="Create well-structured content from research data",
)

# ---------------------------------------------------------------------------
# Define Steps
# ---------------------------------------------------------------------------
research_hackernews_step = Step(
    name="ResearchHackerNews",
    description="Research tech news from Hacker News",
    agent=hackernews_agent,
)

research_web_step = Step(
    name="ResearchWeb",
    description="Research general information from web",
    agent=web_agent,
)

research_exa_step = Step(
    name="ResearchExa",
    description="Research using Exa search",
    agent=exa_agent,
)

prepare_input_for_write_step = Step(
    name="PrepareInput",
    description="Prepare and organize research data for writing",
    agent=content_agent,
)

write_step = Step(
    name="WriteContent",
    description="Write the final content based on research",
    agent=content_agent,
)

tech_analysis_step = Step(
    name="TechAnalysis",
    description="Deep dive tech analysis and trend identification",
    agent=content_agent,
)


# ---------------------------------------------------------------------------
# Define Condition Evaluators
# ---------------------------------------------------------------------------
def should_conduct_research(step_input: StepInput) -> bool:
    topic = step_input.input or step_input.previous_step_content or ""
    research_keywords = [
        "ai",
        "machine learning",
        "programming",
        "software",
        "tech",
        "startup",
        "coding",
        "news",
        "information",
        "research",
        "facts",
        "data",
        "analysis",
        "comprehensive",
        "trending",
        "viral",
        "social",
        "discussion",
        "opinion",
        "developments",
    ]
    return any(keyword in topic.lower() for keyword in research_keywords)


def is_tech_related(step_input: StepInput) -> bool:
    topic = step_input.input or step_input.previous_step_content or ""
    tech_keywords = [
        "ai",
        "machine learning",
        "programming",
        "software",
        "tech",
        "startup",
        "coding",
    ]
    return any(keyword in topic.lower() for keyword in tech_keywords)


# ---------------------------------------------------------------------------
# Create Workflow
# ---------------------------------------------------------------------------
workflow = Workflow(
    name="Conditional Research Workflow",
    description="Conditionally execute parallel research based on topic relevance",
    steps=[
        Condition(
            name="ResearchCondition",
            description="Check if comprehensive research is needed for this topic",
            evaluator=should_conduct_research,
            steps=[
                Parallel(
                    research_hackernews_step,
                    research_web_step,
                    name="ComprehensiveResearch",
                    description="Run multiple research sources in parallel",
                ),
                research_exa_step,
            ],
        ),
        Condition(
            name="TechResearchCondition",
            description="Additional tech-focused research if topic is tech-related",
            evaluator=is_tech_related,
            steps=[tech_analysis_step],
        ),
        prepare_input_for_write_step,
        write_step,
    ],
)

# ---------------------------------------------------------------------------
# Run Workflow
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    try:
        # Sync Streaming
        workflow.print_response(
            input="Latest AI developments in machine learning",
            stream=True,
        )

        # Async Streaming
        asyncio.run(
            workflow.aprint_response(
                input="Latest AI developments in machine learning",
                stream=True,
            )
        )
    except Exception as e:
        print(f"[ERROR] Error: {e}")
    print()

Run the Example

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

uv venv --python 3.12
.venv\Scripts\activate

Install dependencies

uv pip install -U agno cel-python ddgs exa-py fastapi openai

Export your API keys

export EXA_API_KEY="your_exa_api_key_here"
export OPENAI_API_KEY="your_openai_api_key_here"

$Env:EXA_API_KEY="your_exa_api_key_here"
$Env:OPENAI_API_KEY="your_openai_api_key_here"

Run the example

Save the code above as parallel_with_condition.py, then run:

python parallel_with_condition.py

Full source: cookbook/04_workflows/04_parallel_execution/parallel_with_condition.py

​Run the Example

Run the Example