Loop HITL

Confirm before starting loops and review output between iterations.

Loops support two HITL modes: start confirmation (pause before the first iteration) and iteration review (pause between iterations for human feedback).

All HITL settings are configured via HumanReview.

Start Confirmation

When requires_confirmation=True, the loop pauses before executing:

• Confirm: Execute the loop iterations
• Reject: Skip the entire loop

from agno.workflow import Workflow
from agno.workflow.loop import Loop
from agno.workflow.step import Step
from agno.workflow.types import HumanReview, StepInput, StepOutput
from agno.db.sqlite import SqliteDb

def refine_analysis(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Iteration complete: Quality improved")

workflow = Workflow(
    name="refinement_workflow",
    db=SqliteDb(db_file="workflow.db"),
    steps=[
        Step(name="prepare", executor=prepare_data),
        Loop(
            name="refinement_loop",
            steps=[Step(name="refine", executor=refine_analysis)],
            max_iterations=5,
            human_review=HumanReview(
                requires_confirmation=True,
                confirmation_message="Start refinement loop? (up to 5 iterations)",
            ),
        ),
        Step(name="finalize", executor=finalize_results),
    ],
)

run_output = workflow.run("Process data")

if run_output.is_paused:
    for req in run_output.steps_requiring_confirmation:
        print(f"Loop: {req.step_name}")
        print(f"Message: {req.confirmation_message}")
        
        if input("Start loop? (y/n): ").lower() == "y":
            req.confirm()
            print("Starting loop")
        else:
            req.reject()
            print("Skipping loop")
    
    run_output = workflow.continue_run(run_output)

print(run_output.content)

Parameters

Field	Type	Description
requires_confirmation	bool	Pause before first iteration
confirmation_message	str	Message shown to the user
on_reject	OnReject	Action when rejected: skip (default), cancel

Loop Behavior

The confirmation happens once before the loop starts. Individual iterations do not pause for confirmation. For per-iteration pauses, see Iteration Review.

User Action	Result
Confirm	Execute all iterations (up to max_iterations or until should_continue returns False)
Reject	Skip the loop entirely

Iteration Review

Set requires_iteration_review=True to pause between loop iterations. After each iteration completes, the workflow pauses for human review. The reviewer can accept the result (stopping the loop) or reject to run another iteration.

from agno.workflow import Workflow, OnReject
from agno.workflow.loop import Loop
from agno.workflow.step import Step
from agno.workflow.types import HumanReview
from agno.db.sqlite import SqliteDb

refine_agent = Agent(
    name="Refiner",
    model=OpenAIChat(id="gpt-4o-mini"),
    instructions=(
        "You refine and improve text. If the reviewer provides feedback, "
        "incorporate it. Return only the improved text."
    ),
)

workflow = Workflow(
    name="iterative_refinement",
    db=SqliteDb(db_file="workflow.db"),
    steps=[
        Loop(
            name="refinement_loop",
            steps=[Step(name="refine", agent=refine_agent)],
            max_iterations=5,
            forward_iteration_output=True,
            human_review=HumanReview(
                requires_iteration_review=True,
                iteration_review_message="Review this iteration.",
                on_reject=OnReject.retry,
            ),
        ),
    ],
)

run_output = workflow.run("Improve this paragraph...")

while run_output.is_paused:
    for req in run_output.steps_requiring_output_review:
        print(req.step_output.content)

        choice = input("Accept? (yes/no): ").strip().lower()
        if choice in ("yes", "y"):
            req.confirm()
        else:
            feedback = input("Feedback: ").strip()
            if feedback:
                req.reject(feedback=feedback)
            else:
                req.reject()

    run_output = workflow.continue_run(run_output)

Iteration Review Parameters

Parameter	Type	Default	Description
requires_iteration_review	bool	False	Pause after each iteration for review
iteration_review_message	str	None	Message shown to the reviewer
on_reject	OnReject	OnReject.skip	Action on rejection. Use OnReject.retry to run another iteration

Iteration Review Behavior

Reviewer Action	Result
confirm()	Stop the loop. Keep the current iteration’s output
reject()	Run another iteration (when on_reject=OnReject.retry)
reject(feedback="...")	Run another iteration with feedback passed to the agent

Iteration review requirements appear in run_output.steps_requiring_output_review, the same property used for step output review.

requires_iteration_review is only supported on Loop. Passing it to Step or Router raises a ValueError.

With should_continue

The should_continue function controls iteration. Confirmation happens before any iteration:

def check_quality(step_input: StepInput, iteration: int) -> bool:
    return iteration < 3

Loop(
    name="quality_loop",
    steps=[Step(name="improve", executor=improve_quality)],
    should_continue=check_quality,
    human_review=HumanReview(
        requires_confirmation=True,
        confirmation_message="Start quality improvement loop?",
    ),
)

Timeout

Set a deadline for user responses during loop HITL pauses. See the dedicated Timeout page for full details.

from agno.workflow import OnTimeout

Loop(
    name="review_loop",
    steps=[Step(name="generate", executor=generate_fn)],
    max_iterations=3,
    human_review=HumanReview(
        requires_iteration_review=True,
        iteration_review_message="Review this iteration.",
        on_reject=OnReject.retry,
        timeout=120,  # 2 minutes per pause
        on_timeout=OnTimeout.approve,
    ),
)

Streaming

Handle loop HITL in streaming workflows:

from agno.run.workflow import StepPausedEvent

for event in workflow.run("input", stream=True, stream_events=True):
    if isinstance(event, StepPausedEvent):
        print(f"Paused at: {event.step_name}")

session = workflow.get_session()
run_output = session.runs[-1]

while run_output.is_paused:
    for req in run_output.steps_requiring_confirmation:
        req.confirm()
    
    for event in workflow.continue_run(run_output, stream=True, stream_events=True):
        pass
    
    session = workflow.get_session()
    run_output = session.runs[-1]

Developer Resources

• HumanReview Config
• Output Review
• Workflow HITL overview
• Loop reference
• Iterative workflow pattern