Iterative Workflow

from agno.workflow import Loop, Step, StepInput, StepOutput, Workflow


def expand_research(step_input: StepInput) -> StepOutput:
    prior = step_input.previous_step_content or str(step_input.input)
    return StepOutput(content=f"{prior}\nAdditional verified research notes.")


def quality_check(outputs: list[StepOutput]) -> bool:
    return any(len(str(output.content or "")) > 120 for output in outputs)


def create_final_analysis(step_input: StepInput) -> StepOutput:
    research = step_input.previous_step_content or "No research returned"
    return StepOutput(content=f"Final analysis:\n{research}")

workflow = Workflow(
    name="Quality-Driven Research",
    steps=[
        Loop(
            name="Research Loop",
            steps=[Step(name="Deep Research", executor=expand_research)],
            end_condition=quality_check,
            max_iterations=3,
            forward_iteration_output=True,
        ),
        Step(name="Final Analysis", executor=create_final_analysis),
    ],
)

workflow.print_response("Research the impact of renewable energy on global markets", markdown=True)