Router HITL

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

def quick_analysis(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Quick analysis: Basic metrics computed")

def deep_analysis(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Deep analysis: Full statistical analysis")

def custom_analysis(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Custom analysis: User-defined parameters")

workflow = Workflow(
    name="analysis_workflow",
    db=SqliteDb(db_file="workflow.db"),
    steps=[
        Step(name="prepare", executor=prepare_data),
        Router(
            name="analysis_router",
            choices=[
                Step(name="quick", description="Fast analysis (2 min)", executor=quick_analysis),
                Step(name="deep", description="Full analysis (10 min)", executor=deep_analysis),
                Step(name="custom", description="Custom parameters", executor=custom_analysis),
            ],
            human_review=HumanReview(
                requires_user_input=True,
                user_input_message="Select analysis type:",
            ),
            allow_multiple_selections=False,
        ),
        Step(name="report", executor=generate_report),
    ],
)

run_output = workflow.run("Analyze Q4 data")

if run_output.is_paused:
    for req in run_output.steps_requiring_route:
        print(f"Router: {req.step_name}")
        print(f"Message: {req.user_input_message}")
        print(f"Options: {req.available_choices}")
        
        choice = input("Select: ")
        req.select(choice)
    
    run_output = workflow.continue_run(run_output)

print(run_output.content)

Router(
    name="processing_pipeline",
    choices=[
        Step(name="clean", description="Clean data", executor=clean_data),
        Step(name="validate", description="Validate data", executor=validate_data),
        Step(name="enrich", description="Enrich data", executor=enrich_data),
        Step(name="transform", description="Transform data", executor=transform_data),
    ],
    human_review=HumanReview(
        requires_user_input=True,
        user_input_message="Select processing steps:",
    ),
    allow_multiple_selections=True,
)

for req in run_output.steps_requiring_route:
    print(f"Available: {req.available_choices}")
    
    # User selects: "clean, validate, transform"
    selections = input("Select (comma-separated): ").split(",")
    selections = [s.strip() for s in selections]
    
    req.select_multiple(selections)

def route_by_priority(step_input: StepInput) -> str:
    content = step_input.previous_step_content or ""
    if "urgent" in content.lower():
        return "urgent_handler"
    elif "billing" in content.lower():
        return "billing_handler"
    return "general_handler"

Router(
    name="request_router",
    choices=[
        Step(name="urgent_handler", executor=handle_urgent),
        Step(name="billing_handler", executor=handle_billing),
        Step(name="general_handler", executor=handle_general),
    ],
    selector=route_by_priority,
    human_review=HumanReview(
        requires_confirmation=True,
        confirmation_message="Proceed with the selected route?",
    ),
)

for req in run_output.steps_requiring_confirmation:
    print(f"Router: {req.step_name}")
    print(f"Message: {req.confirmation_message}")
    
    if input("Confirm? (y/n): ").lower() == "y":
        req.confirm()
    else:
        req.reject()

from agno.workflow import Workflow, OnReject
from agno.workflow.router import Router
from agno.workflow.step import Step
from agno.workflow.types import HumanReview, StepInput, StepOutput
from agno.db.sqlite import SqliteDb

workflow = Workflow(
    name="router_review_workflow",
    db=SqliteDb(db_file="workflow.db"),
    steps=[
        Router(
            name="analysis_router",
            selector=lambda si: [Step(name="quick", executor=quick_analysis)],
            choices=[
                Step(name="quick", description="Fast analysis (2 min)", executor=quick_analysis),
                Step(name="deep", description="Thorough analysis (10 min)", executor=deep_analysis),
                Step(name="custom", description="Custom analysis", executor=custom_analysis),
            ],
            human_review=HumanReview(
                requires_output_review=True,
                output_review_message="Review the analysis. Approve, or pick a different type?",
                on_reject=OnReject.retry,
                max_retries=5,
            ),
        ),
        Step(name="report", executor=generate_report),
    ],
)

run_output = workflow.run("Analyze Q4 sales data")

while run_output.is_paused:
    for req in run_output.steps_requiring_output_review:
        print(req.step_output.content)
        print(f"Available routes: {req.available_choices}")

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

    # After rejection, pick a different route
    for req in run_output.steps_requiring_route:
        print(f"Pick a different route: {req.available_choices}")
        selection = input("Your choice: ").strip()
        req.select(selection)

    run_output = workflow.continue_run(run_output)

from agno.workflow import OnTimeout

Router(
    name="timed_router",
    choices=[...],
    selector=select_route,
    human_review=HumanReview(
        requires_confirmation=True,
        confirmation_message="Approve this route?",
        timeout=60,
        on_timeout=OnTimeout.approve,
    ),
)

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 router: {event.step_name}")

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

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