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 |
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
