jason810496 commented on code in PR #1717: URL: https://github.com/apache/airflow-site/pull/1717#discussion_r4455430438
########## landing-pages/site/content/en/blog/airflow-3.4.0/index.md: ########## @@ -0,0 +1,513 @@ +--- +title: "Apache Airflow 3.4.0: Task Loops, Dags in Any Language, and AI Agents" +linkTitle: "Apache Airflow 3.4.0: Task Loops, Dags in Any Language, and AI Agents" +author: "Rahul Vats" +github: "vatsrahul1001" +linkedin: "vats-rahul" +description: "Apache Airflow 3.4.0 introduces bounded Task Loops for iterative workflows, Dag authoring beyond Python (TypeScript, Go, Java, and YAML) on the new AIP-85 Dag Importer, and a maturing common.ai provider for running LLM and agent tasks with built-in human approval." +tags: [Release] +date: "2026-11-04" +images: ["/blog/airflow-3.4.0/images/3.4.0.jpg"] +--- + +We're proud to announce the release of **Apache Airflow 3.4.0**! Where 3.3 gave your *tasks* memory and task bodies in multiple languages, 3.4 is about **shape and reach**. It brings **Task Loops**, a native loop primitive for workflows that repeat until they are done; the ability to author an entire Dag in TypeScript, Go, Java, or YAML rather than only Python; and a steadily maturing `common.ai` provider that turns LLM calls and agents into ordinary Airflow tasks, complete with human approval. + +**🎯 Release Highlights** + +📦 PyPI: https://pypi.org/project/apache-airflow/3.4.0/ \ +📚 Docs: https://airflow.apache.org/docs/apache-airflow/3.4.0/ \ +🛠️ Release Notes: https://airflow.apache.org/docs/apache-airflow/3.4.0/release_notes.html \ +🐳 Docker Image: `docker pull apache/airflow:3.4.0` \ +🚏 Constraints: https://github.com/apache/airflow/tree/constraints-3.4.0 + +# 🔁 Task Loops (AIP-111): Workflows That Repeat Until They Are Done + +Some workflows do not have a fixed shape. You walk a tree of unknown depth, you refine an estimate until the error is small enough, or you run an agentic *generate → score → refine* loop until the output is good. Until now you reached for hard-coding a fixed depth or for Dags that trigger themselves. Airflow 3.4 makes **bounded, sequential iteration** a native concept. + +Call `.loop()` on a decorated task group to repeat it a number of times decided at runtime, where each iteration can build on the result of the last. It behaves like a `do … while` loop with a hard cap: + +```python +from airflow.sdk import dag, task, task_group + + +@dag(schedule=None, catchup=False, tags=["example"]) +def refine_estimate(): + @task_group + def refine(): + @task + def improve(*, loop): + # First pass starts from 1.0; later passes read the previous result. + estimate = 1.0 if loop.index == 1 else loop.previous["estimate"] + return (estimate + 2.0 / estimate) / 2.0 + + @task + def evaluate(estimate): + return {"estimate": estimate, "error": abs(estimate * estimate - 2.0)} + + evaluate(improve()) + + def accurate_enough(*, loop): + return loop.result["error"] < 0.000001 + + @task + def finished(): + print("Refinement finished.") + + refinement = refine.loop(max_iterations=10, until=accurate_enough) + refinement >> finished() + + +refine_estimate() +``` + +## Key Capabilities + +- **`max_iterations`**: a required upper bound, so a loop can never run away. +- **`until`**: an optional predicate that decides when to stop. Return `True` to stop, or omit it entirely for a loop that runs a fixed number of times. Because the body runs before the check, it always runs at least once. +- **A `loop` context object** inside the body and the predicate: `loop.index` (the 1-based current pass), `loop.previous` (the last iteration's result), and `loop.result` (for the predicate). +- **Lazy materialization**: Airflow only creates the next iteration when the loop decides to continue, so you do not see a tail of skipped iterations. +- **Composable**: loops work with `.partial()` for body arguments, `.override()` for group config, and even `.expand()` to run mapped tasks *inside* a loop. + +A loop with a fixed count (no `until`) always runs exactly `max_iterations` times and succeeds. If a loop reaches `max_iterations` while its `until` predicate is still `False`, meaning it never converged, the final iteration is marked failed, so a failure to converge surfaces as a real failure. The Airflow UI gains dedicated loop views so you can see each iteration's history at a glance. + +> A few limits to know: loops cannot be nested, a loop cannot sit inside a mapped task group, and a whole loop cannot itself be mapped. + +For detailed usage, see the [loops authoring guide](https://airflow.apache.org/docs/apache-airflow/3.4.0/authoring-and-scheduling/loops.html). + +<!-- TODO image (from Ash): Task Loops iteration view in the UI — e.g. images/loop-iteration-view.png --> + +# 🌐 Write Airflow Dags in TypeScript, Java and Go + +Airflow 3.3 let you write individual *tasks* in Java or Go while Python still owned the Dag. Airflow 3.4 takes the next step: you can define the entire Dag, including its schedule, tasks and dependencies, in TypeScript, Java or Go, with no Python file involved. All three SDKs are generally available. + +This is powered by **AIP-85, the Dag Importer**, a pluggable framework that teaches the Dag processor to read Dag sources beyond `.py`. Each format registers an importer. For a language SDK, the Dag processor runs the language's runtime to read the Dags an artifact declares, and Airflow then treats them exactly like Python Dags. The same framework also powers an opt-in YAML importer. + +## What's supported + +Each SDK works in two modes, and one artifact can serve both: a **Dag definition** declares the whole Dag in the SDK's language, which is what 3.4 adds, and a **Python Dag with stub TaskHandler** implements the `@task.stub` tasks of a Python Dag, including their TaskFlow arguments. A Dag definition supports Dag and task options under Airflow's own names, one queue for all of a Dag's tasks, TaskFlow inputs from task results and literals, order-only dependencies (`before` / `after`), nested task groups, edge labels, if / else and switch / case branching, and triggering another Dag's run. Task code can use Variables, Connections, XCom and the task state store. An artifact can hold several Dags, including Dags generated in a loop, and the Code view shows each Dag's own source file. All of this works the same in TypeScript, Java and Go, and the TypeScript SDK adds human-in-the-loop tasks (approval, `hitl`). + +## The same Dag declared in three languages + +Each example below declares the same Dag, `orders_daily`: + +- the `extract` task group counts orders for two regions, +- `summarize` adds them up, +- `has_orders` runs `load` when there are orders, and `alert` when there are none, +- `pick_report` picks a full or a brief report, +- `trigger_finance` starts a run of the `finance_close` Dag after `load`, on a labeled edge. + +### TypeScript + +A handler takes one object of named arguments, and calling a task names its inputs. The calls you write are the dependencies, and the compiler checks each one. + +<details> +<summary>Show the full TypeScript Dag</summary> + +```typescript +import { Bundle, Dag, label, triggerDagRun } from "apache-airflow-ts-sdk"; + +const dag = new Dag("orders_daily", { schedule: "@daily", tags: ["orders"], queue: "typescript" }); + +// A task group: its task ids are "extract.eu" and "extract.us". +const extract = dag.taskGroup("extract"); +const eu = extract.task("eu", async (): Promise<number> => 120)(); +const us = extract.task("us", async (): Promise<number> => 80)(); + +// TaskFlow: passing a task's result draws the dependency and delivers the value. +const summarize = dag.task("summarize", async ({ eu, us }: { eu: number; us: number }) => eu + us); +const total = summarize({ eu, us }); + +const load = dag.task("load", async ({ total, table }: { total: number; table: string }) => { + console.log(`Loading ${total} orders into ${table}`); +}); +const loaded = load({ total, table: "warehouse.orders" }); +const alerted = dag.task("alert", async () => console.log("No orders today"))(); + +// if / else: the side not taken is skipped. +const hasOrders = async ({ total }: { total: number }) => total > 0; +dag.if(hasOrders, { total }, { taskId: "has_orders" }).then(loaded).else(alerted); + +// switch / case: the decider returns the task to run. +const full = dag.task("report_full", async () => console.log("Full report"))(); +const brief = dag.task("report_brief", async () => console.log("Brief report"))(); +const pickReport = async ({ total }: { total: number }) => (total > 1000 ? full : brief); +dag.switch(pickReport, { total }, { taskId: "pick_report" }).case(full).case(brief); + +// Trigger another Dag after the load, on a labeled edge. +const triggered = dag.task("trigger_finance", triggerDagRun({ dagId: "finance_close" }))(); +loaded.before(label(triggered, "hand off")); + +await new Bundle(dag).serve(); +``` + +</details> + + + +### Java + +Annotations carry the configuration. A `@Airflow.Deps` class wires the graph through a view that the annotation processor generates, so `javac` type-checks every edge. An interface-based API (`DagDef`, `dag.task(...)`) is also available. + +<details> +<summary>Show the full Java Dag</summary> Review Comment: The results of both PR look identical actually 😆 , both works with me and they all look nice! I will refine the quick-start section of non-Python side to include the code example of all three SDKs as code tab for better readability and better visibility. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
