This is an automated email from the ASF dual-hosted git repository. ashb pushed a commit to branch loops-docs in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 6b0ecba04a59be33fc1cd4338cfc5d998a74ebf2 Author: Ash Berlin-Taylor <[email protected]> AuthorDate: Wed Nov 4 11:45:03 2026 +0000 Add screenshots and a review loop example to the task loops docs The task loops page describes several UI views (the loop group in the graph, an iteration's decision reason, mapped tasks per iteration, the iteration switcher and Iterations tab, and clearing later iterations) that are much easier to follow with a picture, so each now has one in both themes. A review gate is where a loop is most useful, so the page also shows a loop that waits for a person in each iteration, with a common.ai example of a model drafting a release note that a reviewer approves or sends back. The example lives with common.ai's other example Dags and the core page includes it. The core docs render that module's source themselves rather than linking to the provider's docs, so each version of the docs shows the code it was built with. Sphinx's viewcode drops a [...] The refine_estimate example now gives a distinct reason when it converges, so the final iteration says why the loop stopped rather than repeating the threshold. --- .../docs/authoring-and-scheduling/loops.rst | 155 +++++++++++++++++++-- airflow-core/docs/img/ui-dark/loops_clear.png | Bin 0 -> 271661 bytes airflow-core/docs/img/ui-dark/loops_decision.png | Bin 0 -> 73779 bytes airflow-core/docs/img/ui-dark/loops_graph.png | Bin 0 -> 77163 bytes airflow-core/docs/img/ui-dark/loops_iterations.png | Bin 0 -> 191319 bytes .../docs/img/ui-dark/loops_iterations_tab.png | Bin 0 -> 156522 bytes airflow-core/docs/img/ui-dark/loops_mapped.png | Bin 0 -> 179885 bytes airflow-core/docs/img/ui-dark/loops_review.png | Bin 0 -> 244676 bytes airflow-core/docs/img/ui-light/loops_clear.png | Bin 0 -> 268407 bytes airflow-core/docs/img/ui-light/loops_decision.png | Bin 0 -> 72201 bytes airflow-core/docs/img/ui-light/loops_graph.png | Bin 0 -> 78174 bytes .../docs/img/ui-light/loops_iterations.png | Bin 0 -> 189121 bytes .../docs/img/ui-light/loops_iterations_tab.png | Bin 0 -> 155523 bytes airflow-core/docs/img/ui-light/loops_mapped.png | Bin 0 -> 177810 bytes airflow-core/docs/img/ui-light/loops_review.png | Bin 0 -> 243231 bytes .../src/airflow/example_dags/example_task_loops.py | 4 +- devel-common/src/sphinx_exts/exampleinclude.py | 45 ++++++ .../ai/example_dags/example_llm_review_loop.py | 81 +++++++++++ 18 files changed, 274 insertions(+), 11 deletions(-) diff --git a/airflow-core/docs/authoring-and-scheduling/loops.rst b/airflow-core/docs/authoring-and-scheduling/loops.rst index 42f45dc57ea..55c20c50593 100644 --- a/airflow-core/docs/authoring-and-scheduling/loops.rst +++ b/airflow-core/docs/authoring-and-scheduling/loops.rst @@ -77,6 +77,21 @@ is small enough: through ``loop.previous``. The gate appears as a task in the loop, named after the condition function: ``refine.accurate_enough`` in this example. +.. tab-set:: + :sync-group: ui-theme + + .. tab-item:: Dark + :sync: dark + + .. image:: /img/ui-dark/loops_graph.png + :alt: Graph view of refine_estimate showing the refine loop group with improve, evaluate and the accurate_enough gate, followed by finished (Dark Mode) + + .. tab-item:: Light + :sync: light + + .. image:: /img/ui-light/loops_graph.png + :alt: Graph view of refine_estimate showing the refine loop group with improve, evaluate and the accurate_enough gate, followed by finished (Light Mode) + Tasks inside a loop body receive ``loop`` as a context parameter, so a parameter with that name in a loop body must be keyword-only and cannot have a default other than ``None``. @@ -122,8 +137,23 @@ Returning a ``LoopDecision(stop=..., reason=...)`` instead of a ``bool`` lets yo detailed information about why the loop continued or finished. ``stop`` has the same meaning as the plain ``bool`` return value. The ``reason`` field is optional text, written by the Dag author, that the Airflow UI shows next to the iteration's outcome, for example ``error 0.17, needs < 1e-06``. It -is limited to 512 bytes of UTF-8; anything longer will result in the gate task failing outright. This -decision will then show up in the UI on Task Instance Details section of the gate task. +is limited to 512 bytes of UTF-8; anything longer will result in the gate task failing outright. The +reason then shows in the UI as the **Decision** in the header of the gate's task instance. + +.. tab-set:: + :sync-group: ui-theme + + .. tab-item:: Dark + :sync: dark + + .. image:: /img/ui-dark/loops_decision.png + :alt: Header of the accurate_enough gate task instance in iteration 3 of 4, showing its decision reason (Dark Mode) + + .. tab-item:: Light + :sync: light + + .. image:: /img/ui-light/loops_decision.png + :alt: Header of the accurate_enough gate task instance in iteration 3 of 4, showing its decision reason (Light Mode) The reason is also recorded when the loop fails because it reached ``max_iterations`` without stopping. @@ -134,6 +164,42 @@ The ``accurate_enough`` condition in the ``refine_estimate`` example above retur uses the same pattern for LLM calls, stopping with a reason when a judge passes the draft or when the loop has spent its dollar or token budget. +Ask a person in each iteration +------------------------------ + +A loop body can wait for a person, so someone reviews each iteration's result before the loop +decides whether to go on. This example asks a model for a release note with ``@task.llm``, has a +person approve it or request changes with +:class:`~airflow.providers.standard.operators.hitl.HITLOperator`, and repeats until they approve. +From the second iteration the prompt includes the previous draft and the reviewer's feedback, read +through ``loop.previous`` (see :ref:`loops-pass-data`): + +.. exampleinclude:: /../../providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llm_review_loop.py + :language: python + :start-after: [START howto_loop_llm_review] + :end-before: [END howto_loop_llm_review] + +Each iteration opens its own review, listed on the **Required Actions** page and in the review task's +**Required Action** tab. ``decide`` turns the reviewer's answer into the iteration's result, so +``until`` stops the loop on approval and otherwise records the feedback as the decision reason, which +the group's **Iterations** tab shows for every iteration. See :doc:`/tutorial/hitl` for the other +ways to ask for input. + +.. tab-set:: + :sync-group: ui-theme + + .. tab-item:: Dark + :sync: dark + + .. image:: /img/ui-dark/loops_review.png + :alt: The review loop at iteration 3 of up to 5 in the Graph view, next to the review form with the model's third draft and a feedback field (Dark Mode) + + .. tab-item:: Light + :sync: light + + .. image:: /img/ui-light/loops_review.png + :alt: The review loop at iteration 3 of up to 5 in the Graph view, next to the review form with the model's third draft and a feedback field (Light Mode) + Read the loop context ===================== @@ -171,6 +237,8 @@ task instance's position. Each task instance can have several tries, each with i The task instance also exposes the loop context as ``ti.loop``, and for a mapped task ``ti.expanded_ti_count`` is the number of instances in its expansion, counted within the iteration. +.. _loops-pass-data: + Pass data between iterations ============================ @@ -313,6 +381,21 @@ not count from the same number: iterations start at 1 and map indexes start at Mapped instances appear within their loop iteration, so you can inspect their states, tries, and logs separately. +.. tab-set:: + :sync-group: ui-theme + + .. tab-item:: Dark + :sync: dark + + .. image:: /img/ui-dark/loops_mapped.png + :alt: Mapped task instances of process_batch.process showing map indexes 0 and 1 in iterations 1 and 2 (Dark Mode) + + .. tab-item:: Light + :sync: light + + .. image:: /img/ui-light/loops_mapped.png + :alt: Mapped task instances of process_batch.process showing map indexes 0 and 1 in iterations 1 and 2 (Light Mode) + Mapping a whole loop, nesting loops, and placing a loop inside a mapped task group are not supported. @@ -373,28 +456,80 @@ Iterations and execution history ================================ In the Grid, select the loop's task group in a Dag run to open its Task Instances -tab. The **Iteration** filter narrows the table to one iteration, or shows -**All iterations**, and the **Iteration** column shows which iteration each task -instance belongs to. The gate's state and logs explain why the loop continued, +tab. Choose an iteration from the **Iteration** switcher in the breadcrumb, or with +the **Loop** filter, to narrow the table to that iteration; **All** shows every +iteration. The **Iteration** column shows which iteration each task instance +belongs to. The gate's state and logs explain why the loop continued, stopped, or failed. Each task's tries and logs remain accessible within its iteration. +.. tab-set:: + :sync-group: ui-theme + + .. tab-item:: Dark + :sync: dark + + .. image:: /img/ui-dark/loops_iterations.png + :alt: Task Instances of the refine loop group narrowed to iteration 2, with the Iteration breadcrumb, the Loop filter and the Iteration column (Dark Mode) + + .. tab-item:: Light + :sync: light + + .. image:: /img/ui-light/loops_iterations.png + :alt: Task Instances of the refine loop group narrowed to iteration 2, with the Iteration breadcrumb, the Loop filter and the Iteration column (Light Mode) + +The group's **Iterations** tab lists every iteration, newest first, with its state, start date, +duration, number of tasks and the gate's decision reason. Select an iteration's number to open the +Task Instances tab narrowed to that iteration. + +.. tab-set:: + :sync-group: ui-theme + + .. tab-item:: Dark + :sync: dark + + .. image:: /img/ui-dark/loops_iterations_tab.png + :alt: Iterations tab of the refine loop group listing iterations 4 to 1 with each gate's decision reason (Dark Mode) + + .. tab-item:: Light + :sync: light + + .. image:: /img/ui-light/loops_iterations_tab.png + :alt: Iterations tab of the refine loop group listing iterations 4 to 1 with each gate's decision reason (Light Mode) + Iterations cleared by a rerun are kept as history but are not shown in the UI in 3.4.0. Clear tasks inside a loop ========================== +Open a task instance inside the loop and select the clear button in its header, or press +:kbd:`Shift+C`, to open the **Clear Task Instance** dialog. Use these controls to select how far a clear extends through the loop: -* **Clear downstream tasks** includes downstream tasks. It starts selected unless you have +* **Downstream** includes downstream tasks. It starts selected unless you have changed your saved clear options. Clearing a task this way can also clear the gate in the same iteration. -* **Clear later loop iterations**, selected by default, clears later +* **Later iterations**, selected by default, clears later iterations when the selection includes a gate task, whether selected - directly or through **Clear downstream tasks**. The gate runs again and decides + directly or through **Downstream**. The gate runs again and decides whether the loop should continue from that point. +.. tab-set:: + :sync-group: ui-theme + + .. tab-item:: Dark + :sync: dark + + .. image:: /img/ui-dark/loops_clear.png + :alt: Clear Task Instance dialog opened from the refine.improve task instance page for iteration 2, with Downstream and Later iterations selected, listing the affected tasks in iterations 2 to 4 (Dark Mode) + + .. tab-item:: Light + :sync: light + + .. image:: /img/ui-light/loops_clear.png + :alt: Clear Task Instance dialog opened from the refine.improve task instance page for iteration 2, with Downstream and Later iterations selected, listing the affected tasks in iterations 2 to 4 (Light Mode) + Clearing later iterations does not require the loop to run the same number of iterations again. The gate can stop earlier or continue further, within the configured limit. @@ -409,7 +544,7 @@ Suppose each iteration contains this sequence: prepare → process → consume → gate The loop has already run iterations 1 through 4. You clear ``process`` in -iteration 2 with **Clear downstream tasks** and **Clear later loop iterations** both selected: +iteration 2 with **Downstream** and **Later iterations** both selected: * ``prepare`` in iteration 2 remains completed. * ``process``, ``consume``, and the gate in iteration 2 run again. @@ -423,7 +558,7 @@ created. Keep later iterations --------------------- -Deselect **Clear later loop iterations** to keep the later iterations. +Deselect **Later iterations** to keep the later iterations. Rerunning an earlier gate then does not change the loop progression that already followed it: a new stop result does not remove later iterations, and a new continue result does not create a duplicate next iteration. diff --git a/airflow-core/docs/img/ui-dark/loops_clear.png b/airflow-core/docs/img/ui-dark/loops_clear.png new file mode 100644 index 00000000000..44f7040ce10 Binary files /dev/null and b/airflow-core/docs/img/ui-dark/loops_clear.png differ diff --git a/airflow-core/docs/img/ui-dark/loops_decision.png b/airflow-core/docs/img/ui-dark/loops_decision.png new file mode 100644 index 00000000000..96d9d25d903 Binary files /dev/null and b/airflow-core/docs/img/ui-dark/loops_decision.png differ diff --git a/airflow-core/docs/img/ui-dark/loops_graph.png b/airflow-core/docs/img/ui-dark/loops_graph.png new file mode 100644 index 00000000000..f2937e8fbe1 Binary files /dev/null and b/airflow-core/docs/img/ui-dark/loops_graph.png differ diff --git a/airflow-core/docs/img/ui-dark/loops_iterations.png b/airflow-core/docs/img/ui-dark/loops_iterations.png new file mode 100644 index 00000000000..072b686ff8e Binary files /dev/null and b/airflow-core/docs/img/ui-dark/loops_iterations.png differ diff --git a/airflow-core/docs/img/ui-dark/loops_iterations_tab.png b/airflow-core/docs/img/ui-dark/loops_iterations_tab.png new file mode 100644 index 00000000000..d008c030277 Binary files /dev/null and b/airflow-core/docs/img/ui-dark/loops_iterations_tab.png differ diff --git a/airflow-core/docs/img/ui-dark/loops_mapped.png b/airflow-core/docs/img/ui-dark/loops_mapped.png new file mode 100644 index 00000000000..d7ed5029485 Binary files /dev/null and b/airflow-core/docs/img/ui-dark/loops_mapped.png differ diff --git a/airflow-core/docs/img/ui-dark/loops_review.png b/airflow-core/docs/img/ui-dark/loops_review.png new file mode 100644 index 00000000000..8f1c1ea5450 Binary files /dev/null and b/airflow-core/docs/img/ui-dark/loops_review.png differ diff --git a/airflow-core/docs/img/ui-light/loops_clear.png b/airflow-core/docs/img/ui-light/loops_clear.png new file mode 100644 index 00000000000..33844fa138b Binary files /dev/null and b/airflow-core/docs/img/ui-light/loops_clear.png differ diff --git a/airflow-core/docs/img/ui-light/loops_decision.png b/airflow-core/docs/img/ui-light/loops_decision.png new file mode 100644 index 00000000000..ec3c7c1fc53 Binary files /dev/null and b/airflow-core/docs/img/ui-light/loops_decision.png differ diff --git a/airflow-core/docs/img/ui-light/loops_graph.png b/airflow-core/docs/img/ui-light/loops_graph.png new file mode 100644 index 00000000000..e4698bf7f55 Binary files /dev/null and b/airflow-core/docs/img/ui-light/loops_graph.png differ diff --git a/airflow-core/docs/img/ui-light/loops_iterations.png b/airflow-core/docs/img/ui-light/loops_iterations.png new file mode 100644 index 00000000000..4b3c0dfb99f Binary files /dev/null and b/airflow-core/docs/img/ui-light/loops_iterations.png differ diff --git a/airflow-core/docs/img/ui-light/loops_iterations_tab.png b/airflow-core/docs/img/ui-light/loops_iterations_tab.png new file mode 100644 index 00000000000..e7d0f810614 Binary files /dev/null and b/airflow-core/docs/img/ui-light/loops_iterations_tab.png differ diff --git a/airflow-core/docs/img/ui-light/loops_mapped.png b/airflow-core/docs/img/ui-light/loops_mapped.png new file mode 100644 index 00000000000..a78e16405ab Binary files /dev/null and b/airflow-core/docs/img/ui-light/loops_mapped.png differ diff --git a/airflow-core/docs/img/ui-light/loops_review.png b/airflow-core/docs/img/ui-light/loops_review.png new file mode 100644 index 00000000000..73a1ae218c4 Binary files /dev/null and b/airflow-core/docs/img/ui-light/loops_review.png differ diff --git a/airflow-core/src/airflow/example_dags/example_task_loops.py b/airflow-core/src/airflow/example_dags/example_task_loops.py index 5ee8984a0a7..4a97ebbd1c6 100644 --- a/airflow-core/src/airflow/example_dags/example_task_loops.py +++ b/airflow-core/src/airflow/example_dags/example_task_loops.py @@ -40,7 +40,9 @@ def refine_estimate(): def accurate_enough(*, loop): error = loop.result["error"] threshold = 0.000001 - return LoopDecision(stop=error < threshold, reason=f"error {error:.2g}, needs < {threshold:g}") + if error < threshold: + return LoopDecision(stop=True, reason=f"converged with error {error:.2g}") + return LoopDecision(stop=False, reason=f"error {error:.2g}, needs < {threshold:g}") @task def finished(): diff --git a/devel-common/src/sphinx_exts/exampleinclude.py b/devel-common/src/sphinx_exts/exampleinclude.py index 7bde12bc1ee..a8e32b10ccf 100644 --- a/devel-common/src/sphinx_exts/exampleinclude.py +++ b/devel-common/src/sphinx_exts/exampleinclude.py @@ -253,15 +253,57 @@ def doctree_read(app, doctree): else: modname = ".".join(split_modname) modname = modname.replace(".py", "") + elif filepath.name.endswith(".py") and "src" in filepath.parts: + # A module from another distribution is rendered in these docs too, so each version links to + # the source it was built with, and is named by its import path rather than where it sits. + src_index = len(filepath.parts) - 1 - filepath.parts[::-1].index("src") + module_path = Path(*filepath.parts[src_index + 1 :]) + modname = ".".join(module_path.with_suffix("").parts) else: modname = None module_path = filepath.resolve() show_button = register_source(app, env, modname, filepath) + if show_button: + _included_modules(env).setdefault(modname, (str(filepath), set()))[1].add(env.docname) onlynode = create_node(env, module_path.as_posix(), show_button) objnode.replace_self(onlynode) +def _included_modules(env) -> dict[str, tuple[str, set[str]]]: + """Modules shown by an example include, with the file each comes from and the pages that include it.""" + if not hasattr(env, "_exampleinclude_modules"): + env._exampleinclude_modules = {} + return env._exampleinclude_modules + + +def env_purge_doc(app, env, docname): + for modname, (_filepath, docnames) in list(_included_modules(env).items()): + docnames.discard(docname) + if not docnames: + del env._exampleinclude_modules[modname] + + +def env_merge_info(app, env, docnames, other): + for modname, (filepath, other_docnames) in _included_modules(other).items(): + _included_modules(env).setdefault(modname, (filepath, set()))[1].update(other_docnames) + + +def env_updated(app, env): + """ + Register the included modules again before the source pages are written. + + Sphinx's viewcode drops a module whenever a page is read again unless documented code refers to it, + and a module that only an example include shows has no such reference. + """ + if not hasattr(env, "_viewcode_modules"): + env._viewcode_modules = {} + for modname, (filepath, _docnames) in _included_modules(env).items(): + if modname not in env._viewcode_modules: + register_source(app, env, modname, filepath) + return [] + + def setup(app): """ Sets the plugin up and returns configuration of the plugin. @@ -271,6 +313,9 @@ def setup(app): """ directives.register_directive("exampleinclude", ExampleInclude) app.connect("doctree-read", doctree_read) + app.connect("env-purge-doc", env_purge_doc) + app.connect("env-merge-info", env_merge_info) + app.connect("env-updated", env_updated) app.add_config_value("exampleinclude_sourceroot", None, "env") if not airflow_theme_is_available: # Sphinx airflow theme has its own styles. diff --git a/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llm_review_loop.py b/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llm_review_loop.py new file mode 100644 index 00000000000..d906e99c5cd --- /dev/null +++ b/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llm_review_loop.py @@ -0,0 +1,81 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Example Dag: a model drafts a release note, a person reviews it, and the loop repeats until they approve.""" + +from __future__ import annotations + +# [START howto_loop_llm_review] +from airflow.providers.standard.operators.hitl import HITLOperator +from airflow.sdk import LoopDecision, Param, dag, task, task_group + + +@dag(tags=["example"]) +def example_llm_review_loop(): + @task_group + def draft_and_review(): + @task.llm( + llm_conn_id="pydanticai_default", + system_prompt="You write release notes for the users of a developer platform. Be plain and specific.", + ) + def write(*, loop): + prompt = ( + "Write a one-sentence release note for this change: webhook deliveries that fail with a 5xx " + "response are now retried three times with exponential backoff instead of dropping the event." + ) + if loop.index > 1: + prompt += ( + f"\nYour previous draft:\n{loop.previous['draft']}\n" + f"\nThe reviewer asked for these changes:\n{loop.previous['feedback']}\n" + ) + return prompt + + draft = write() + review = HITLOperator( + task_id="review", + subject="Review the release note", + body="Please review the following, generated by the model:\n\n---\n\n{{ ti.xcom_pull(task_ids='draft_and_review.write') }}", + options=["Approve", "Request changes"], + params={"feedback": Param(None, type=["null", "string"], description="What should change?")}, + ) + draft >> review + + @task + def decide(draft, response): + return { + "draft": draft, + "approved": response["chosen_options"] == ["Approve"], + "feedback": response["params_input"].get("feedback") or "", + } + + decide(draft, review.output) + + def approved(*, loop): + if loop.result["approved"]: + return LoopDecision(stop=True, reason="approved") + return LoopDecision(stop=False, reason=f"changes requested: {loop.result['feedback']}") + + reviewed = draft_and_review.loop(max_iterations=5, until=approved) + + @task + def publish(rounds): + print(f"Publishing after {len(rounds)} round(s): {rounds[-1]['draft']}") + + publish(reviewed["decide"].output) + + +example_llm_review_loop() +# [END howto_loop_llm_review]
