> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Human-in-the-loop workflows with Airflow
Human-in-the-loop (HITL) workflows are processes that require human intervention, for example, to approve or reject an AI generated output, or choose a [branch](/docs/learn/airflow-branch-operator) in a Dag depending on the result of an upstream task. The [Airflow standard provider](https://airflow.apache.org/docs/apache-airflow-providers-standard/stable/index.html) contains a set of operators to create tasks that will wait for human input, either in the Airflow UI or through the [Airflow REST API](https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html). Note that you need to be on Airflow 3.1+ to use the human-in-the-loop operators.
This guide covers the available HITL operators, as well as how to interact with them in the UI and through the API.
## Assumed knowledge
To get the most out of this guide, you should have an existing knowledge of:
* Airflow basics. See [Introduction to Apache Airflow](/docs/learn/intro-to-airflow).
* Airflow operators. See [Operators 101](/docs/learn/what-is-an-operator).
* Deferrable operators. See [Deferrable operators](/docs/learn/deferrable-operators).
## When to use human-in-the-loop workflows
Human-in-the-loop workflows are useful whenever you need the input from a human (or another entity outside of Airflow) within a Dag. For example:
* Your Dag uses AI to create responses to a support ticket, and you want to ask a human to review the response and approve it or request changes.
* Your Dag generates a compliance report and you need a human to verify and acknowledge the results.
* You'd like to use AI to route product feature requests to the appropriate team and need a human to decide where to send edge cases.
* You have a Dag that requires input from a domain expert, for example to input feedback gathered in user research interviews.
Human-in-the-loop workflows are very common, especially with increased usage of AI to generate assets that are in need of human review. The Airflow HITL features allows you to build these workflows and have non-technical team members provide their input either in the Airflow UI or through an implementation built around the relevant [Airflow REST API](https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html) endpoints.
## Required Actions
Each task instance created by a human-in-the-loop operator creates a **Required Action** object. You can view a list of all required actions (pending and resolved) for your whole Airflow instance under **Browse > Required Actions** in the Airflow UI.
To respond to a required action you can either:
* Navigate to the task instance page's **Required Actions** tab and respond directly in the UI.
* Make a call to the [Airflow REST API](https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html)'s update HITL detail endpoint.
In Airflow 3.3+ you can reply to required actions directly from the **Required Actions** tab after clicking on the sidebar icon.
## Human-in-the-loop operators
There are 4 human-in-the-loop operators available in the Airflow standard provider:
* [`HITLOperator`](https://airflow.apache.org/registry/providers/standard#standard-hitl-HITLOperator): The base class for all human-in-the-loop operators.
* [`ApprovalOperator`](https://airflow.apache.org/registry/providers/standard#standard-hitl-ApprovalOperator): A specialised form of the `HITLOperator` where the two options are `Approve` and `Reject`.
* [`HITLBranchOperator`](https://airflow.apache.org/registry/providers/standard#standard-hitl-HITLBranchOperator): A specialised form of the `HITLOperator` where the user input provides a [branching](/docs/learn/airflow-branch-operator) decision, choosing the (set of) task(s) to run next.
* [`HITLEntryOperator`](https://airflow.apache.org/registry/providers/standard#standard-hitl-HITLEntryOperator): A specialised form of the `HITLOperator` where the user provides input to a form.
While waiting for human input, human-in-the-loop operators don't take up a worker slot. In Airflow 3.1 and 3.2, human-in-the-loop operators are implemented as [deferrable operators](/docs/learn/deferrable-operators), meaning that they run an asynchronous trigger process in the [Triggerer](/docs/learn/airflow-components) component while waiting for the human input. In Airflow 3.3+, human-in-the-loop operators enter the `awaiting_input` task state while waiting for a required action to receive a response. The `awaiting_input` state doesn't require a running triggerer or worker.
When choosing which human-in-the-loop operator to use, consider the following:
* If the human-in-the-loop action centers around a branching decision (choosing the next task(s) to run), use the `HITLBranchOperator`.
* If you are looking to make a binary decision (approval or rejection of information displayed at runtime), use the `ApprovalOperator`. If users need to give additional input add a form field using `params`.
* If you are mostly interested in letting users provide input to a form, use the `HITLEntryOperator`.
* For all other use cases, you can use the `HITLOperator` directly. It allows you to display custom options for your user to choose from, as well as additional input fields with `params`.
In a lot of cases you'll likely be using a combination of these operators to build your human-in-the-loop workflow, for example using the `HITLBranchOperator` upstream to choose whether to accept an AI generated ticket response or escalate the ticket to a human. Then, downstream, the human's answer can be provided in a `HITLEntryOperator`. Both the user decision (`chosen_options`) and any input to parameters (`params_input`) are available for downstream tasks to use by pulling the information from [XComs](/docs/learn/airflow-passing-data-between-tasks).
### `HITLOperator`
The `HITLOperator` is the base class for all human-in-the-loop operators. It is the most versatile operator in this operator family. With it you can display information, let the user choose one or more from a list of options and accept additional input using a form based on [Airflow params](/docs/learn/airflow-params).
Two parameters are mandatory when instantiating the `HITLOperator`:
* `subject` (required): The subject of the templated action which is displayed as the title in the **Required Actions** tab. This field is templatable, which means you can use [Jinja templates](/docs/learn/templating) to render information at runtime, including information computed by an upstream task.
* `options` (required): A list of strings that are rendered as response options at the bottom of the required action form. Note that the list can't be empty. The chosen options can be retrieved in downstream tasks by pulling the information from [XComs](/docs/learn/airflow-passing-data-between-tasks), which is stored as a list under the `chosen_options` key.
There are also several optional parameters that you can use to further configure the behavior of the `HITLOperator`:
* `body`: The main text body. This field is templatable as well and supports markdown formatting.
* `defaults`: Optionally, you can provide a list of one or more options that are selected by default if the task times out before a human responds. All default options need to be in the `options` list.
* `multiple`: If set to `True`, the user can select multiple options. Default is `False`.
* `params`: With this parameter you can create form fields for any user input to the required action using [Airflow params](/docs/learn/airflow-params). Note that not all param functionality is supported for human-in-the-loop operators. These params can be retrieved in downstream tasks by pulling the information from [XComs](/docs/learn/airflow-passing-data-between-tasks), which is stored as a nested dictionary under the `params_input` key.
* `execution_timeout`: This is a [`BaseOperator`](https://registry.astronomer.io/providers/apache-airflow/versions/latest/modules/BaseOperator) parameter that times out the task after a specified duration provided as a [`datetime.timedelta`](https://docs.python.org/3/library/datetime.html#datetime.timedelta) or [pendulum duration](https://pendulum.eustace.io/docs/#duration) object. Default is `None`. After the timeout has been reached the behavior depends on whether you provided a `defaults` list or not:
* If you provided a `defaults` list, the default(s) is/are chosen as the response and the task succeeds.
* If you didn't provide a `defaults` list, the task fails.
* `assigned_users`: A list of all users who are allowed to respond to the required action. Users are provided as `HITLUser` objects (`from airflow.sdk.execution_time.hitl import HITLUser`) with an `id` and `name` field.
* If you are running Airflow on [Astro](https://www.astronomer.io/lp/signup/), the id of each user is their Astro ID in the format `cl1a2b3cd456789ef1gh2ijkl3`. You can find each user's Astro ID under **Organization** -> **Access Management**.
* If you are using the [SimpleAuthManager](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/auth-manager.html#simpleauthmanager) the id of each user is their username.
* If you are using the [FabAuthManager](https://airflow.apache.org/docs/apache-airflow-providers-fab/stable/auth-manager/index.html) the id of each user is their email.
* `notifiers`: A list of notifiers of which to execute the `.notify()` method when the task starts running. See [Use notifiers with HITL operators](#use-notifiers-with-hitl-operators) for an example.
The following example shows a simple use of the `HITLOperator` where the output of an upstream task is fetched from [XComs](/docs/learn/airflow-passing-data-between-tasks) and the user is given 3 response options to choose from. One additional input field for the `expense_amount` is rendered in the Airflow UI using an [Airflow param](/docs/learn/airflow-params). After 5 minutes (`execution_timeout`) the task times out and chooses the default option: `ACH Transfer` as the response and `10000` as the expense amount.
The downstream `print_result` task prints out the information pushed to XComs by the `HITLOperator` task: the `chosen_options` and the `params_input` values.
```python expandable wrap theme={null}
from airflow.providers.standard.operators.hitl import HITLOperator
from airflow.sdk import dag, task, chain, Param
@dag
def HITLOperator_syntax_example():
@task
def upstream_task():
return "Review expense report and approve vendor payment method."
_upstream_task = upstream_task()
_hitl_task = HITLOperator(
task_id="hitl_task",
subject="Expense Approval Required", # templatable
body="{{ ti.xcom_pull(task_ids='upstream_task') }}", # templatable
options=["ACH Transfer", "Wire Transfer", "Corporate Check"], # cannot be empty!
defaults=["ACH Transfer"],
multiple=False, # default: False
params={
"expense_amount": Param(
10000,
type="number",
)
},
execution_timeout=timedelta(minutes=5), # default: None
)
@task
def print_result(hitl_output):
print(f"Expense amount: ${hitl_output['params_input']['expense_amount']}")
print(f"Payment method: {hitl_output['chosen_options']}")
_print_result = print_result(_hitl_task.output)
chain(_upstream_task, _hitl_task)
HITLOperator_syntax_example()
```
The open required action form is displayed on the task instance page's **Required Actions** tab.
### `ApprovalOperator`
The `ApprovalOperator` is a specialised form of the `HITLOperator` where the only two possible response options are `Approve` and `Reject`. Additionally, you can provide param form fields for any user input to the required action using [Airflow params](/docs/learn/airflow-params).
If the human chooses to `Approve` the task succeeds. If the human chooses to `Reject` the task succeeds but all downstream tasks are skipped.
```python expandable wrap theme={null}
from airflow.providers.standard.operators.hitl import ApprovalOperator
from airflow.sdk import dag, task, chain, Param
@dag
def ApprovalOperator_syntax_example():
@task
def upstream_task():
return "Pineapple on pizza?"
_upstream_task = upstream_task()
_hitl_task = ApprovalOperator(
task_id="approval_task",
subject="Your task:",
body="{{ ti.xcom_pull(task_ids='upstream_task') }}",
defaults="Approve", # other option: "Reject"
params={
"second_topping": Param(
"olives",
type="string",
)
},
)
@task
def print_result(hitl_output):
print(f"Params input: {hitl_output['params_input']}")
print(f"Chosen options: {hitl_output['chosen_options']}")
_print_result = print_result(_hitl_task.output)
chain(_upstream_task, _hitl_task)
ApprovalOperator_syntax_example()
```
The action form shows the two options `Approve` and `Reject` alongside any param form fields.
### `HITLBranchOperator`
If you want to [branch](/docs/learn/airflow-branch-operator) your Dag based on the human input, you can use the `HITLBranchOperator`. This operator allows the user to choose one or more tasks that are directly downstream of the `HITLBranchOperator` task to run next. All tasks that aren't chosen will be skipped.
You can use the `options_mapping` parameter to map the human facing options to the task IDs of the tasks that are downstream of the `HITLBranchOperator` task.
```python expandable wrap theme={null}
from airflow.providers.standard.operators.hitl import HITLBranchOperator
from airflow.sdk import dag, task, chain
_budget_categories = ["marketing", "research_development", "facilities", "training", "technology"]
@dag
def HITLBranchOperator_syntax_example():
@task
def upstream_task():
return {
"total_budget": "$4B",
}
_upstream_task = upstream_task()
_hitl_branch_task = HITLBranchOperator(
task_id="hitl_branch_task",
subject="Budget Category Approval",
body="""**Total Budget Available:** {{ ti.xcom_pull(task_ids='upstream_task')['total_budget'] }}
Select the funding proposals to approve for this quarter.""",
options=_budget_categories,
defaults=["marketing", "research_development"],
multiple=True,
)
for _category in _budget_categories:
@task(
task_id=f"{_category}", # needs to match options in HITLBranchOperator
)
def category_task():
print(f"Processing budget approval for {_category}")
_category_task = category_task()
chain(_hitl_branch_task, _category_task)
chain(_upstream_task, _hitl_branch_task)
HITLBranchOperator_syntax_example()
```
The screenshot below shows the graph view created by the code snippet above with 5 tasks downstream of the `HITLBranchOperator` task and the **Required Actions** tab showing the form input.
After approving the budget for 3 of the categories the Dag completes with 3 downstream tasks being run and 2 being skipped.
### `HITLEntryOperator`
The `HITLEntryOperator` is a specialised form of the `HITLOperator` where the user provides input to a form and then submits the input without choosing from a list of options.
```python expandable wrap theme={null}
from airflow.providers.standard.operators.hitl import HITLEntryOperator
from airflow.sdk import dag, task, chain, Param
@dag
def HITLEntryOperator_syntax_example():
@task
def upstream_task():
return "How can I auto-pause a dag if it fails?"
_upstream_task = upstream_task()
_hitl_task = HITLEntryOperator(
task_id="hitl_task",
subject="Please respond to this ticket!",
body="{{ ti.xcom_pull(task_ids='upstream_task') }}",
params={
"response": Param(
"You can use the max_consecutive_failed_dag_runs parameter! :)",
type="string",
),
"urgency": Param(
"p3",
type="string",
),
},
)
@task
def print_result(hitl_output):
print(f"Params input: {hitl_output['params_input']}")
print(f"Chosen options: {hitl_output['chosen_options']}")
_print_result = print_result(_hitl_task.output)
chain(_upstream_task, _hitl_task)
HITLEntryOperator_syntax_example()
```
## Use notifiers with HITL operators
You can use an [Airflow notifier](/docs/learn/error-notifications-in-airflow) to send information from the human-in-the-loop operator to another system, such as Slack or email. The `.notify()` method of the notifier is executed when the task starts running. A simple implementation is to use the `HITLOperator.generate_link_to_ui_from_context` method to return a link to the required action in the Airflow UI for users to click to respond.
The code snippet below shows a sample notifier `MyNotifier` that prints the required action information and the link to the required action to the Airflow logs.
```python expandable wrap theme={null}
from airflow.sdk import BaseNotifier, Context, dag, task, Param
from airflow.providers.standard.operators.hitl import HITLOperator
from datetime import timedelta
_BASE_URL = "http://localhost:28080"
class MyNotifier(BaseNotifier):
template_fields = ("message",)
def __init__(self, message: str = "") -> None:
self.message = message
def notify(self, context: Context) -> None:
task_state = context['ti'].state
if task_state == "running":
# this method generates a direct link to the UI page where the user can respond
url = HITLOperator.generate_link_to_ui_from_context(
context=context,
base_url=_BASE_URL,
)
# placeholder code, you can send the URL to any service you want
self.log.info(self.message)
self.log.info("Url to respond %s", url)
else:
self.log.info("Task state: %s", task_state)
self.log.info("No response needed!")
notifier_class = MyNotifier(
message="""
Subject: {{ task.subject }}
Body: {{ task.body }}
Options: {{ task.options }}
"""
)
@dag
def notifier_example():
HITLOperator(
task_id="hitl_task",
subject="Choose a number: ",
options=["23", "19", "42"],
notifiers=[notifier_class],
)
notifier_example()
```
Of course you can also add all regular callback functions such as `on_failure_callback`, `on_success_callback`, etc. to the human-in-the-loop operators.
## Human-in-the-loop API endpoints
If your human (or other entity) doesn't have access to the Airflow UI, you can use the [Airflow REST API](https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html) to poll for required actions and respond to them.
The relevant endpoints are:
* `GET api/v2/hitlDetails/` to get a list of required actions in an Airflow instance filtered by flags like `state`, `dag_id` and more.
* `GET api/v2/hitlDetails/{dag_id}/{dag_run_id}/{task_id}` to get the details of a specific required action.
* `PATCH api/v2/hitlDetails/{dag_id}/{dag_run_id}/{task_id}` to respond to a specific required action.
These API calls can be combined with others to create scripts like the one below that allows you to respond to all pending required actions in a specific Dag from the command line.
```python expandable wrap theme={null}
import requests
from typing import Any
_USERNAME = "admin"
_PASSWORD = "admin"
_HOST = "http://localhost:28080/" # To learn how to send API requests to Airflow running on Astro see: https://www.astronomer.io/docs/astro/airflow-api/
_DAG_ID = "HITLOperator_syntax_example"
_TASK_ID = "hitl_task"
def _pick_option(options: list[str]):
print("Available options: ", options)
chosen_option = input("Enter the option you want to select: ")
return chosen_option
def _pick_params(param: dict[str, Any]):
print("Input for param:", param)
param_input = input("Enter your value for the param: ")
return param_input
def _get_jwt_token():
token_url = f"{_HOST}/auth/token"
payload = {"username": _USERNAME, "password": _PASSWORD}
headers = {"Content-Type": "application/json"}
response = requests.post(token_url, json=payload, headers=headers)
token = response.json().get("access_token")
return token
def _get_running_dagruns_for_dag(dag_id: str):
url = f"{_HOST}/api/v2/dags/{dag_id}/dagRuns?state=running"
headers = {"Authorization": f"Bearer {_get_jwt_token()}"}
response = requests.get(url, headers=headers)
return response.json()
def _get_hitl_details(dag_id: str, dag_run_id: str, task_id: str):
url = f"{_HOST}/api/v2/hitlDetails/{dag_id}/{dag_run_id}/{task_id}"
headers = {"Authorization": f"Bearer {_get_jwt_token()}"}
response = requests.get(url, headers=headers)
if response.status_code == 200:
subject = response.json()["subject"]
body = response.json()["body"]
options = response.json()["options"]
params = response.json()["params"]
print("--------------------------------")
print("Required Action found for: ", dag_id, "DAG Run: ", dag_run_id, "Task: ", task_id)
print("Subject: ", subject)
print("Body: ", body)
print("Options: ", options)
print("Params: ", params)
print("--------------------------------")
return {
"subject": subject,
"body": body,
"options": options,
"params": params,
}
elif response.status_code == 404:
print("--------------------------------")
print("404 - No required action found for: ", dag_id, "DAG Run: ", dag_run_id, "Task: ", task_id)
print("Response: ", response.json())
print("--------------------------------")
return None
else:
print("--------------------------------")
print("Error: ", response.status_code)
print("Response: ", response.json())
print("--------------------------------")
return None
def _add_hitl_response(
dag_id: str, dag_run_id: str, task_id: str, options: list[str], params: dict[str, Any]
):
url = f"{_HOST}/api/v2/hitlDetails/{dag_id}/{dag_run_id}/{task_id}"
headers = {"Authorization": f"Bearer {_get_jwt_token()}"}
chosen_options = [_pick_option(options)]
if params:
params_input = {f"{param}": _pick_params(param) for param in params}
else:
params_input = {}
response = requests.patch(
url, headers=headers, json={"chosen_options": chosen_options, "params_input": params_input}
)
if response.status_code == 200:
print("--------------------------------")
print("Hitl response added for DAG: ", dag_id, "DAG Run: ", dag_run_id, "Task: ", task_id)
print("Chosen options: ", chosen_options)
print("Params input: ", params_input)
print("Response status code: ", response.status_code)
print("Response: ", response.json())
print("--------------------------------")
elif response.status_code == 409:
print("--------------------------------")
print("409 - Already updated action for: ", dag_id, "DAG Run: ", dag_run_id, "Task: ", task_id)
print("Response: ", response.json())
print("--------------------------------")
else:
print("--------------------------------")
print("Error: ", response.status_code)
print("Response: ", response.json())
print("--------------------------------")
def main():
dag_runs = _get_running_dagruns_for_dag(_DAG_ID)["dag_runs"]
if not dag_runs:
print("No running dag runs found for DAG: ", _DAG_ID)
else:
for dag_run in dag_runs:
hitl_details = _get_hitl_details(_DAG_ID, dag_run["dag_run_id"], _TASK_ID)
if hitl_details:
_add_hitl_response(
_DAG_ID, dag_run["dag_run_id"], _TASK_ID, hitl_details["options"], hitl_details["params"]
)
if __name__ == "__main__":
main()
```