Parallel block inside a Loop to run several research steps at once on each
iteration. The end condition checks the combined output and decides whether the loop
needs another pass.
When to use: A loop whose iteration contains independent tasks that can run
concurrently before the end condition is evaluated.
Install dependencies:
uv pip install agno openai ddgs
loop_with_parallel_steps_stream.py
from typing import List
from agno.agent import Agent
from agno.tools.hackernews import HackerNewsTools
from agno.tools.websearch import WebSearchTools
from agno.workflow import Loop, Parallel, Step, Workflow
from agno.workflow.types import StepOutput
# Create agents for research
news_research_agent = Agent(
name="HackerNews Research Agent",
role="Research specialist",
tools=[HackerNewsTools()],
instructions="Research the given topic on Hacker News.",
markdown=True,
)
web_research_agent = Agent(
name="Web Research Agent",
role="Research specialist",
tools=[WebSearchTools()],
instructions="Research the given topic on the web.",
markdown=True,
)
analysis_agent = Agent(
name="Analysis Agent",
role="Data analyst",
instructions="You are a data analyst. Analyze and summarize research findings.",
markdown=True,
)
content_agent = Agent(
name="Content Agent",
role="Content creator",
instructions="You are a content creator. Create engaging content based on research.",
markdown=True,
)
# Create research steps
research_hackernews_step = Step(
name="Research HackerNews",
agent=news_research_agent,
description="Research trending topics on HackerNews",
)
research_web_step = Step(
name="Research Web",
agent=web_research_agent,
description="Research additional information from web sources",
)
# Create analysis steps
trend_analysis_step = Step(
name="Trend Analysis",
agent=analysis_agent,
description="Analyze trending patterns in the research",
)
sentiment_analysis_step = Step(
name="Sentiment Analysis",
agent=analysis_agent,
description="Analyze sentiment and opinions from the research",
)
content_step = Step(
name="Create Content",
agent=content_agent,
description="Create content based on research findings",
)
# End condition function
def research_evaluator(outputs: List[StepOutput]) -> bool:
"""
Evaluate if research results are sufficient
Returns True to break the loop, False to continue
"""
# Check if we have good research results
if not outputs:
return False
# Calculate total content length from all outputs
total_content_length = sum(len(output.content or "") for output in outputs)
# Check if we have substantial content (more than 500 chars total)
if total_content_length > 500:
print(
f"Research evaluation passed - found substantial content ({total_content_length} chars total)"
)
return True
print(
f"Research evaluation failed - need more substantial research (current: {total_content_length} chars)"
)
return False
# Create workflow with loop containing parallel steps
workflow = Workflow(
name="Advanced Research and Content Workflow",
description="Research topics with parallel execution in a loop until conditions are met, then create content",
steps=[
Loop(
name="Research Loop with Parallel Execution",
steps=[
Parallel(
research_hackernews_step,
research_web_step,
name="Parallel Research",
description="Run Hacker News and web research concurrently",
),
trend_analysis_step,
sentiment_analysis_step,
],
end_condition=research_evaluator,
max_iterations=3, # Maximum 3 iterations
),
content_step,
],
)
if __name__ == "__main__":
workflow.print_response(
input="Research the latest trends in AI and machine learning, then create a summary",
stream=True,
)