Human-in-the-loop (HITL) review for agents¶

HITL Review adds an interactive feedback loop to agentic operators. After the LLM Agent produces an initial output, a human reviewer can approve, reject, or request changes through a chat UI. The operator blocks until a terminal action, or until a timeout is reached, max_iterations is reached, or polling the review XCom keeps failing.

This document describes the architecture, workflow, API, XCom schema, and usage.

Overview¶

Workflow¶

The operator and the plugin never talk to each other directly — every arrow below crosses the XCom store. The operator holds its worker slot for the whole loop; it polls instead of deferring because the agent’s message history and tool state live in-process.

Once the operator reads a human action, it resolves to one of five outcomes:

Using HITL review with `AgentOperator`¶

Enable the review loop with enable_hitl_review=True:

from airflow.providers.common.ai.operators.agent import AgentOperator
from airflow.providers.common.ai.toolsets.sql import SQLToolset
from datetime import timedelta

AgentOperator(
    task_id="summarize",
    prompt="Summarize the sales data",
    llm_conn_id="openai",
    toolsets=[SQLToolset(db_conn_id="postgres")],
    enable_hitl_review=True,
    hitl_timeout=timedelta(minutes=30),
    hitl_poll_interval=10.0,
)

Parameters

enable_hitl_review: When True, the operator enters the review loop after the first generation. Default False.
max_hitl_iterations: Maximum outputs the reviewer can see (1 = initial output plus subsequent regenerations). When the reviewer requests changes at iteration >= max_hitl_iterations, the task fails with HITLMaxIterationsError without running the LLM. For example, 5 allows changes at iterations 1 to 4; the fifth output must be either approved or rejected. Default 5.
hitl_timeout: Maximum wall-clock time to wait for all review rounds. None = no wall-clock timeout. The task still fails, re-raising the XCom error, if polling the human action XCom fails 10 times in a row.
hitl_poll_interval: Seconds between XCom polls while waiting for a human response. Default 10.

Accessing the chat UI: The chat loads as a React plugin on the task instance page. Use the HITL Review extra link on the task instance, or navigate to /dags/{dag_id}/runs/{run_id}/tasks/{task_id}/plugin/hitl-review.

Example Dag

airflow/providers/common/ai/example_dags/example_agent.py[source]

@dag(tags=["example"])
def example_agent_operator_hitl_review():
    """AgentOperator with HITL review — a human approves output via hitl-review plugin UI."""
    AgentOperator(
        task_id="summarize_with_review",
        prompt="Summarize the Q4 sales report in 3 bullet points.",
        llm_conn_id="pydantic_ai_default",
        system_prompt="You are a concise business analyst.",
        enable_hitl_review=True,
        max_hitl_iterations=5,
        hitl_timeout=timedelta(minutes=30),
        hitl_poll_interval=10.0,
    )

REST API¶

The plugin exposes a FastAPI app at /hitl-review. Base URL:

{AIRFLOW_BASE_URL}/hitl-review

Common query parameters (where applicable):

dag_id: Dag ID.
run_id: Dag run ID.
task_id: Task ID.
map_index: Map index for mapped tasks. Use -1 for non-mapped tasks or index for dynamic mapping.
region_id and region_index: Select one loop pass or one mapped slot. Use the region_id and region_index of the task instance as returned by the public task instance endpoints. Send them together; region_index without region_id is rejected. Only supported on Airflow 3.4 or later; earlier hosts return 400 when either is sent. A task inside a loop addressed without them returns 400, and a selection that matches more than one task instance returns 409.

Endpoints¶

Method	Path	Description
GET	`/health`	Liveness check. Returns `{"status": "ok"}`.
GET	`/sessions/find`	Find the feedback session for a task instance. Returns `HITLReviewResponse` or 404 if no session.
POST	`/sessions/feedback`	Request changes. Body: `{"feedback": "..."}`. Session must be `pending_review`. Returns updated session.
POST	`/sessions/approve`	Approve the current output. Session must be `pending_review`.
POST	`/sessions/reject`	Reject the output. Session must be `pending_review`.

Response model: HITLReviewResponse¶

{
    "dag_id": str,
    "run_id": str,
    "task_id": str,
    "status": "pending_review"
    | "changes_requested"
    | "approved"
    | "rejected"
    | "max_iterations_exceeded"
    | "timeout_exceeded",
    "iteration": int,
    "max_iterations": int,
    "prompt": str,
    "current_output": str,
    "conversation": [{"role": "assistant" | "human", "content": str, "iteration": int}],
    "task_completed": bool,
}

XCom keys and storage¶

All keys use the prefix airflow_hitl_review_.

Key	Writer	Value
`airflow_hitl_review_agent_session`	Worker	`AgentSessionData`: status, iteration, prompt, current_output
`airflow_hitl_review_human_action`	Plugin	`HumanActionData`: action (approve\|reject\|changes_requested), feedback, iteration
`airflow_hitl_review_agent_output_1`, `_2`, …	Worker	Per-iteration AI output (string or JSON)
`airflow_hitl_review_human_feedback_1`, `_2`, …	Plugin	Per-iteration human feedback text

Session lifecycle¶

pending_review: Awaiting human action. Plugin accepts approve, reject, or feedback.
changes_requested: Feedback submitted; worker is regenerating (or polling for the next action). Plugin does not accept new actions until the worker pushes a new output and status returns to pending_review.
approved / rejected: Terminal. Worker has exited the loop.

Chat UI¶

The plugin provides an interactive chat UI that loads in the task instance page. The UI:

Fetches session and conversation from the REST API
Displays the current output and feedback history
Submits approve, reject, or feedback via POST endpoints