Router HITL

from agno.workflow import Workflow
from agno.workflow.router import Router
from agno.workflow.step import Step
from agno.workflow.types import 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),
            ],
            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),
    ],
    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,
    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.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]