Human-in-the-Loop in Workflows

uv pip install -U agno sqlalchemy

uv pip install -U openai
export OPENAI_API_KEY="your_openai_api_key"

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

workflow = Workflow(
    name="data_pipeline",
    db=SqliteDb(db_file="workflow.db"),  # Required for HITL
    steps=[
        Step(name="fetch_data", agent=fetch_agent),
        Step(
            name="process_data",
            agent=process_agent,
            human_review=HumanReview(
                requires_confirmation=True,
                confirmation_message="Process sensitive data?",
                on_reject=OnReject.skip,
            ),
        ),
    ],
)

run_output = workflow.run("Process user data")

if run_output.is_paused:
    for req in run_output.steps_requiring_confirmation:
        req.confirm()  # or req.reject()
    run_output = workflow.continue_run(run_output)

from agno.db.sqlite import SqliteDb       # SQLite for development
from agno.db.postgres import PostgresDb   # PostgreSQL for production

workflow = Workflow(db=SqliteDb(db_file="workflow.db"))
workflow = Workflow(db=PostgresDb(db_url="postgresql+psycopg://user:password@localhost:5432/dbname"))

from agno.workflow import OnTimeout
from agno.workflow.types import HumanReview

Step(
    name="approve_deploy",
    agent=deploy_agent,
    human_review=HumanReview(
        requires_confirmation=True,
        confirmation_message="Approve deployment?",
        timeout=300,                       # 5 minutes
        on_timeout=OnTimeout.approve,      # approve | skip | cancel
    ),
)

from agno.run.workflow import StepPausedEvent, StepExecutorPausedEvent

for event in workflow.run("input", stream=True):
    if isinstance(event, (StepPausedEvent, StepExecutorPausedEvent)):
        pass

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

if run_output.is_paused:
    # resolve requirements...
    for event in workflow.continue_run(run_output, stream=True):
        pass

from agno.workflow.decorators import pause
from agno.workflow.types import StepInput, StepOutput, UserInputField

@pause(
    requires_user_input=True,
    user_input_message="Enter parameters:",
    user_input_schema=[
        UserInputField(name="threshold", field_type="float", required=True),
    ],
)
def process_data(step_input: StepInput) -> StepOutput:
    threshold = step_input.additional_data["user_input"]["threshold"]
    return StepOutput(content=f"Processed with threshold {threshold}")

Step(name="process", executor=process_data)