Condition HITL

Let users decide which branch to execute in conditional workflows.

Conditions support confirmation HITL, allowing users to decide which branch to execute at runtime.

All HITL settings are configured via HumanReview. Conditions only support requires_confirmation. Passing unsupported fields (like requires_output_review) raises a ValueError.

User-Controlled Branching

When requires_confirmation=True, the condition pauses for user decision:

Confirm: Execute the steps branch (if branch)
Reject: Behavior depends on on_reject setting

1from kern.workflow import Workflow, OnReject
2from kern.workflow.condition import Condition
3from kern.workflow.step import Step
4from kern.workflow.types import HumanReview, StepInput, StepOutput
5from kern.db.sqlite import SqliteDb
6
7def detailed_analysis(step_input: StepInput) -> StepOutput:
8    return StepOutput(content="Detailed analysis: Full review completed")
9
10def quick_summary(step_input: StepInput) -> StepOutput:
11    return StepOutput(content="Quick summary: Key highlights identified")
12
13workflow = Workflow(
14    name="analysis_workflow",
15    db=SqliteDb(db_file="workflow.db"),
16    steps=[
17        Step(name="analyze", executor=analyze_data),
18        Condition(
19            name="analysis_depth",
20            steps=[Step(name="detailed", executor=detailed_analysis)],
21            else_steps=[Step(name="quick", executor=quick_summary)],
22            human_review=HumanReview(
23                requires_confirmation=True,
24                confirmation_message="Perform detailed analysis?",
25                on_reject=OnReject.else_branch,
26            ),
27        ),
28        Step(name="report", executor=generate_report),
29    ],
30)
31
32run_output = workflow.run("Analyze Q4 data")
33
34if run_output.is_paused:
35    for req in run_output.steps_requiring_confirmation:
36        print(f"Decision: {req.step_name}")
37        print(f"Message: {req.confirmation_message}")
38        
39        if input("Confirm? (y/n): ").lower() == "y":
40            req.confirm()
41            print("Executing 'if' branch")
42        else:
43            req.reject()
44            print("Executing 'else' branch")
45    
46    run_output = workflow.continue_run(run_output)
47
48print(run_output.content)

Parameters

Parameter	Type	Description
`requires_confirmation`	`bool`	Pause for user decision
`confirmation_message`	`str`	Message shown to the user
`on_reject`	`OnReject`	Action when rejected

OnReject Options

Value	Behavior
`OnReject.else_branch`	Execute `else_steps` (default)
`OnReject.skip`	Skip the entire condition
`OnReject.cancel`	Cancel the workflow

Branch Execution

User Action	on_reject	Result
Confirm	any	Execute `steps`
Reject	`else_branch`	Execute `else_steps`
Reject	`skip`	Skip condition, continue workflow
Reject	`cancel`	Cancel workflow

Without else_steps

If no else_steps are defined and user rejects with on_reject=OnReject.else_branch, the condition is skipped:

1Condition(
2    name="optional_processing",
3    steps=[Step(name="process", executor=process)],
4    # No else_steps defined
5    human_review=HumanReview(
6        requires_confirmation=True,
7        confirmation_message="Run optional processing?",
8        on_reject=OnReject.else_branch,  # Will skip if rejected
9    ),
10)

Combining with Evaluator

When requires_confirmation=True, the evaluator is ignored. The user's decision takes precedence:

1Condition(
2    name="user_controlled",
3    # evaluator is ignored when requires_confirmation=True
4    steps=[Step(name="if_branch", ...)],
5    else_steps=[Step(name="else_branch", ...)],
6    human_review=HumanReview(
7        requires_confirmation=True,
8        confirmation_message="Proceed with if branch?",
9    ),
10)

Streaming

Handle condition 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: {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_confirmation:
12        req.confirm()  # or req.reject()
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]