Router HITL

Select routes, confirm routing decisions, or review router output before continuing.

Routers support three HITL modes: user selection (user chooses routes), confirmation (user approves automated routing), and output review (review router output after execution).

All HITL settings are configured via HumanReview.

User Selection

Let users choose which route(s) to execute. The router pauses and presents available choices.

1from kern.workflow import Workflow
2from kern.workflow.router import Router
3from kern.workflow.step import Step
4from kern.workflow.types import HumanReview, StepInput, StepOutput
5from kern.db.sqlite import SqliteDb
6
7def quick_analysis(step_input: StepInput) -> StepOutput:
8    return StepOutput(content="Quick analysis: Basic metrics computed")
9
10def deep_analysis(step_input: StepInput) -> StepOutput:
11    return StepOutput(content="Deep analysis: Full statistical analysis")
12
13def custom_analysis(step_input: StepInput) -> StepOutput:
14    return StepOutput(content="Custom analysis: User-defined parameters")
15
16workflow = Workflow(
17    name="analysis_workflow",
18    db=SqliteDb(db_file="workflow.db"),
19    steps=[
20        Step(name="prepare", executor=prepare_data),
21        Router(
22            name="analysis_router",
23            choices=[
24                Step(name="quick", description="Fast analysis (2 min)", executor=quick_analysis),
25                Step(name="deep", description="Full analysis (10 min)", executor=deep_analysis),
26                Step(name="custom", description="Custom parameters", executor=custom_analysis),
27            ],
28            human_review=HumanReview(
29                requires_user_input=True,
30                user_input_message="Select analysis type:",
31            ),
32            allow_multiple_selections=False,
33        ),
34        Step(name="report", executor=generate_report),
35    ],
36)
37
38run_output = workflow.run("Analyze Q4 data")
39
40if run_output.is_paused:
41    for req in run_output.steps_requiring_route:
42        print(f"Router: {req.step_name}")
43        print(f"Message: {req.user_input_message}")
44        print(f"Options: {req.available_choices}")
45        
46        choice = input("Select: ")
47        req.select(choice)
48    
49    run_output = workflow.continue_run(run_output)
50
51print(run_output.content)

Parameters

Field	Type	Description
`requires_user_input`	`bool`	Pause for user to select route(s)
`user_input_message`	`str`	Message shown to the user
`allow_multiple_selections`	`bool`	Allow selecting multiple routes (default: `False`)

Selection Methods

Method	Description
`req.select("route_name")`	Select a single route
`req.select_single("route_name")`	Select exactly one route
`req.select_multiple(["a", "b"])`	Select multiple routes (requires `allow_multiple_selections=True`)

Multiple Selection

Allow users to select multiple routes. Selected routes execute in sequence.

1Router(
2    name="processing_pipeline",
3    choices=[
4        Step(name="clean", description="Clean data", executor=clean_data),
5        Step(name="validate", description="Validate data", executor=validate_data),
6        Step(name="enrich", description="Enrich data", executor=enrich_data),
7        Step(name="transform", description="Transform data", executor=transform_data),
8    ],
9    human_review=HumanReview(
10        requires_user_input=True,
11        user_input_message="Select processing steps:",
12    ),
13    allow_multiple_selections=True,
14)

Handle multiple selections:

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

The selected steps execute in the order they appear in choices, not the selection order.

Confirmation Mode

Confirm automated routing decisions. A selector function determines the route, but the user must approve before execution.

1def route_by_priority(step_input: StepInput) -> str:
2    content = step_input.previous_step_content or ""
3    if "urgent" in content.lower():
4        return "urgent_handler"
5    elif "billing" in content.lower():
6        return "billing_handler"
7    return "general_handler"
8
9Router(
10    name="request_router",
11    choices=[
12        Step(name="urgent_handler", executor=handle_urgent),
13        Step(name="billing_handler", executor=handle_billing),
14        Step(name="general_handler", executor=handle_general),
15    ],
16    selector=route_by_priority,
17    human_review=HumanReview(
18        requires_confirmation=True,
19        confirmation_message="Proceed with the selected route?",
20    ),
21)

Handle confirmation:

1for req in run_output.steps_requiring_confirmation:
2    print(f"Router: {req.step_name}")
3    print(f"Message: {req.confirmation_message}")
4    
5    if input("Confirm? (y/n): ").lower() == "y":
6        req.confirm()
7    else:
8        req.reject()

Confirmation Parameters

Field	Type	Description
`requires_confirmation`	`bool`	Pause for user to confirm routing decision
`confirmation_message`	`str`	Message shown to the user
`on_reject`	`OnReject`	Action when rejected: `skip` (default), `cancel`

Output Review

Review a router's output after execution. If rejected, the reviewer can pick a different route. Set on_reject=OnReject.retry to re-route on rejection. See the dedicated Output Review page for full details.

1from kern.workflow import Workflow, OnReject
2from kern.workflow.router import Router
3from kern.workflow.step import Step
4from kern.workflow.types import HumanReview, StepInput, StepOutput
5from kern.db.sqlite import SqliteDb
6
7workflow = Workflow(
8    name="router_review_workflow",
9    db=SqliteDb(db_file="workflow.db"),
10    steps=[
11        Router(
12            name="analysis_router",
13            selector=lambda si: [Step(name="quick", executor=quick_analysis)],
14            choices=[
15                Step(name="quick", description="Fast analysis (2 min)", executor=quick_analysis),
16                Step(name="deep", description="Thorough analysis (10 min)", executor=deep_analysis),
17                Step(name="custom", description="Custom analysis", executor=custom_analysis),
18            ],
19            human_review=HumanReview(
20                requires_output_review=True,
21                output_review_message="Review the analysis. Approve, or pick a different type?",
22                on_reject=OnReject.retry,
23                max_retries=5,
24            ),
25        ),
26        Step(name="report", executor=generate_report),
27    ],
28)
29
30run_output = workflow.run("Analyze Q4 sales data")
31
32while run_output.is_paused:
33    for req in run_output.steps_requiring_output_review:
34        print(req.step_output.content)
35        print(f"Available routes: {req.available_choices}")
36
37        choice = input("Approve? (yes/no): ").strip().lower()
38        if choice in ("yes", "y"):
39            req.confirm()
40        else:
41            req.reject()
42
43    # After rejection, pick a different route
44    for req in run_output.steps_requiring_route:
45        print(f"Pick a different route: {req.available_choices}")
46        selection = input("Your choice: ").strip()
47        req.select(selection)
48
49    run_output = workflow.continue_run(run_output)

The router executes the selected branch, then pauses for review. On rejection with OnReject.retry, the workflow re-pauses for route selection so the reviewer can pick a different branch.

Output Review Parameters

Parameter	Type	Default	Description
`requires_output_review`	`bool`	`False`	Pause after execution for review
`output_review_message`	`str`	`None`	Message shown to the reviewer
`on_reject`	`OnReject`	`OnReject.skip`	Action on rejection: `skip`, `cancel`, `retry` (re-route)
`max_retries`	`int`	`3`	Maximum re-route attempts

Comparing HITL Modes

Mode	When It Pauses	User Action	Use Case
User Selection	Before execution	Chooses route(s)	Interactive wizards, user-driven workflows
Confirmation	Before execution	Approves/rejects	Oversight of automated decisions
Output Review	After execution	Approves/rejects/re-routes	Review results before continuing

Use user selection when the user should decide the path. Use confirmation when the system decides but needs human approval. Use output review when the result matters more than the route chosen.

Timeout

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

1from kern.workflow import OnTimeout
2
3Router(
4    name="timed_router",
5    choices=[...],
6    selector=select_route,
7    human_review=HumanReview(
8        requires_confirmation=True,
9        confirmation_message="Approve this route?",
10        timeout=60,
11        on_timeout=OnTimeout.approve,
12    ),
13)

Streaming

Handle router HITL in streaming workflows:

1from kern.run.workflow import StepPausedEvent
2
3for event in workflow.run("input", stream=True, stream_events=True):
4    if isinstance(event, StepPausedEvent):
5        print(f"Paused at router: {event.step_name}")
6
7session = workflow.get_session()
8run_output = session.runs[-1]
9
10while run_output.is_paused:
11    for req in run_output.steps_requiring_route:
12        req.select(req.available_choices[0])
13    
14    for event in workflow.continue_run(run_output, stream=True, stream_events=True):
15        pass
16    
17    session = workflow.get_session()
18    run_output = session.runs[-1]