Human In The Loop Workflows

Combination of automation and human judgment. Automation handles standard tasks, while humans validate or intervene when needed.

Human In The Loop workflow

What are Human In The Loop workflows ?

A Human In The Loop (HITL) workflow is a type of workflow that combines automated and manual tasks that need human intervention. Rather than fully automating a process, a HITL workflow includes steps where humans review, validate, or make decisions that machines cannot handle on their own.

When to Use Human In The Loop workflows

Many real world processes do not require full automation. In fact, the best workflows are those that combine the decision-making of humans and the power of software to handle tedious tasks.

HITL workflows are widely used in following scenarios:

When the cost of errors is significant for example: financial, legal, safety, or reputational consequences.
When handling edge cases, that automated systems struggle to handle
When starting a new worflows, and wants to build the trust.

To illustrate the importance of HITL let's take the example of a legal document review process. The system starts by scanning documents for relevant keywords, clauses, and potential issues. A human review step in which lawyers review documents flagged by the system. Without the second review step, the system could make errors that have high risk that cannot be tolerated.

Workflow Orchestration Tools with out-of-the-box Support for HITL Workflows

If you would like to use an orchestration tool to handle your HITL workflows you have to make sure that the tool supports this kind of workflow and allows you to use manual steps inside the workflows. Some workflow orchestration tools do not have built-in support for HITL workflows and it may be challenging to add these manual steps. So be careful to choose the right tool before implementing your workflows. In this section we will list some of the well-known tools that support this type of workflow.

1. Prefect

Prefect provide two functions (pause_flow_run and suspend_flow_run) that can be used to implement HITL workflows. Here is an example using pause_flow_run fonction:

1from prefect import flow
2from prefect.flow_runs import pause_flow_run
3from prefect.logging import get_run_logger
4
5@flow
6def greet_user():
7  logger = get_run_logger()
8  user = pause_flow_run(wait_for_input=str)
9  logger.info(f"Hello, {user}!")
10

2. Argo workflows

In Argo workflows you can suspend the execution of workflows using the suspend template, below is an example of a workflow using this feature.

1
2apiVersion: argoproj.io/v1alpha1
3kind: Workflow
4metadata:
5  generateName: suspend-template-
6spec:
7  entrypoint: suspend
8  templates:
9  - name: suspend
10    steps:
11    - - name: build
12        template: hello-world
13    - - name: approve
14        template: approve
15    - - name: delay
16        template: delay
17    - - name: release
18        template: hello-world
19
20  - name: approve
21    suspend: {}
22
23  - name: delay
24    suspend:
25      duration: "20"    # Must be a string. Default unit is seconds. Could also be a Duration, e.g.: "2m", "6h"
26
27  - name: hello-world
28    container:
29      image: busybox
30      command: [echo]
31      args: ["hello world"]
32

Flyte

Flyte provide three different methods, flytekit.sleep, flytekit.wait_for_input, and flytekit.approve, for implementing HITL workflows. The example below use wait_for_input to implement an approval step before publishing an important report.

1import typing
2from flytekit import wait_for_input
3
4@union.task
5def create_report(data: typing.List[float]) -> dict:
6  return {"report_data": data"}
7
8@union.task
9def send_report(report: dict, title: str) -> dict:
10  return {"title": title, **report}
11
12@union.workflow
13def reporting_workflow(data: typing.List[float]) -> dict:
14  report = create_report(data=data)
15  title_input = wait_for_input("title", timeout=timedelta(hours=1), expected_type=str)
16  return send_report(report=report, title=title_input)
17

Airflow

Airflow do not support user validation tasks. But there is a work around to implement this type of task using the retry feature. Similar to this code:

1from datetime import timedelta
2
3from airflow import AirflowException
4
5def fail_task():
6    raise AirflowException("Please change this step to success to continue")
7
8user_validation_step = PythonOperator(
9    dag=dag,
10    task_id="user_validation_step",
11    python_callable=fail_task,
12    retries=1,
13    max_retry_delay=timedelta(days=7)
14)
15

Wrapping Up

With the rise of agent development, and due to the nature of LLMs and the hallucination problem, Human In The Loop workflows will be used more and more in the future. In this article, we have seen what HITL workflows are, when to use them, and some of the workflow orchestration tools that support this type of workflow.