From 0e029535401c69bf1d0449e4807b0ba6efc2ee75 Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Wed, 8 Jul 2026 17:55:27 +0800 Subject: [PATCH 1/8] Add an Examples entry point to the common.ai provider docs The provider had ~25 example Dags under example_dags/, but no page grouped them by scenario or surfaced the ones with no home in an operator/hook guide (10-K RAG comparisons, survey analysis, the AIP progress tracker). A newcomer landing on the docs index had no way to discover them. --- providers/common/ai/docs/examples.rst | 146 ++++++++++++++++++++++++++ providers/common/ai/docs/index.rst | 1 + 2 files changed, 147 insertions(+) create mode 100644 providers/common/ai/docs/examples.rst diff --git a/providers/common/ai/docs/examples.rst b/providers/common/ai/docs/examples.rst new file mode 100644 index 0000000000000..11ce55ae4463a --- /dev/null +++ b/providers/common/ai/docs/examples.rst @@ -0,0 +1,146 @@ + .. 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. + +.. _howto/examples: + +Examples +======== + +Every operator, decorator, and integration in this provider has a runnable Dag under +`example_dags `__. +This page groups them by scenario. Guides embed the same Dags inline where a step-by-step +walkthrough exists; the rest are linked directly to source below. + +New to the provider? Start with :doc:`operators/index` to pick an operator, then come back here +for a worked example of the pattern you need. + +Single-prompt tasks +-------------------- + +.. list-table:: + :header-rows: 1 + :widths: 30 70 + + * - Guide + - What it shows + * - :doc:`operators/llm` + - ``@task.llm`` for text, structured output, classification, and dynamic task mapping over + LLM results (``example_llm.py``, ``example_llm_classification.py``, + ``example_llm_analysis_pipeline.py``). + * - :doc:`operators/llm_branch` + - ``@task.llm_branch`` picking which downstream task runs. + * - :doc:`operators/llm_file_analysis` + - ``@task.llm_file_analysis`` reasoning over files, images, and PDFs. + * - :doc:`operators/llm_schema_compare` + - ``@task.llm_schema_compare`` comparing two schemas with an LLM. + * - :doc:`operators/llm_sql` + - ``@task.llm_sql`` generating SQL from a natural-language question. + +Agents & tools +-------------- + +.. list-table:: + :header-rows: 1 + :widths: 30 70 + + * - Guide + - What it shows + * - :doc:`operators/agent` + - ``AgentOperator`` / ``@task.agent`` multi-turn tool use, durable execution, and pydantic-ai + capabilities + (`example_agent.py `__, + `example_agent_durable.py `__, + `example_agent_capabilities.py `__). + * - :ref:`Toolsets ` + - Loading ``SKILL.md`` Agent Skills + (`example_agent_skills.py `__) + and exposing an Airflow toolset to a LangChain agent, the reverse bridge + (`example_langchain_toolset_bridge.py `__). + * - :doc:`connections/mcp` + - Connecting an agent to an MCP server through an Airflow connection. + * - :doc:`hitl_review` + - Adding a human-in-the-loop review gate to agent output. + * - `example_langchain_tool_agent.py `__ + - A LangChain ReAct agent that decides its own tool calls, composed with ``LLMOperator`` for + report formatting and AIP-90 HITL review. + +Retrieval & document processing +-------------------------------- + +.. list-table:: + :header-rows: 1 + :widths: 30 70 + + * - Guide + - What it shows + * - :doc:`operators/document_loader` + - Parsing PDF, DOCX, CSV, and JSON into ``list[dict]`` for embedding. + * - :doc:`hooks/pydantic_ai` + - Calling ``PydanticAIHook`` and a pydantic-ai ``Agent`` directly. + * - :doc:`hooks/langchain` + - ``LangChainHook`` chat-only, embedding-only, and combined patterns. + * - :doc:`hooks/llamaindex` + - ``LlamaIndexHook`` plus the embedding and retrieval operators. + +End-to-end scenarios +--------------------- + +Production-shaped Dags that combine several of the patterns above into one pipeline. These don't +have a dedicated guide yet -- read the source directly: + +.. list-table:: + :header-rows: 1 + :widths: 30 70 + + * - Example + - What it shows + * - `example_llamaindex_rag.py `__ + - Three RAG shapes with LlamaIndex: a single load-embed-retrieve-answer Dag, a production-shaped + split into scheduled indexing plus on-demand query Dags, and multi-source RAG. + * - `example_llamaindex_10k.py `__, + `example_langchain_10k.py `__ + - SEC 10-K financial analysis against live EDGAR filings: one Dag indexes filings on a + schedule, the other decomposes a comparison question at runtime and fans out retrieval with + Dynamic Task Mapping. One variant per RAG library. + * - `example_llm_survey_analysis.py `__ + - Natural-language querying of a CSV via ``LLMSQLQueryOperator`` and ``AnalyticsOperator``, + interactive and scheduled variants. + * - `example_llm_survey_agentic.py `__ + - The same survey data, analyzed by fanning a multi-dimensional research question into an + agentic multi-query synthesis pipeline. + * - `example_aip_progress_tracker.py `__ + - The same use case -- tracking Airflow Improvement Proposal progress -- solved two ways, a + deterministic pipeline and an autonomous agent, to compare the tradeoff. + +Reliability +----------- + +.. list-table:: + :header-rows: 1 + :widths: 30 70 + + * - Guide + - What it shows + * - :doc:`retry_policies` + - Classifying task failures with an LLM to decide retry, fail, or delay. Source: + `example_llm_retry_policy.py `__. + +.. toctree:: + :hidden: + :maxdepth: 1 + + end_to_end_pipelines diff --git a/providers/common/ai/docs/index.rst b/providers/common/ai/docs/index.rst index eb1f768de6050..c01522e37f7b8 100644 --- a/providers/common/ai/docs/index.rst +++ b/providers/common/ai/docs/index.rst @@ -79,6 +79,7 @@ OpenAI, Anthropic, or other pydantic-ai-supported connection: Hooks Toolsets Operators + Examples Retry Policies HITL Review Observability From 32ee39ec857c45bd4a4a46452be43f2dbfc2e031 Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Thu, 9 Jul 2026 15:45:05 +0800 Subject: [PATCH 2/8] fixup! Support delegating providers release process to non-PMC committers (#69417) --- .../common/ai/docs/end_to_end_pipelines.rst | 154 ++++++++++++++++++ providers/common/ai/docs/examples.rst | 7 +- providers/common/ai/docs/index.rst | 1 + 3 files changed, 158 insertions(+), 4 deletions(-) create mode 100644 providers/common/ai/docs/end_to_end_pipelines.rst diff --git a/providers/common/ai/docs/end_to_end_pipelines.rst b/providers/common/ai/docs/end_to_end_pipelines.rst new file mode 100644 index 0000000000000..fc02b23bcb016 --- /dev/null +++ b/providers/common/ai/docs/end_to_end_pipelines.rst @@ -0,0 +1,154 @@ + .. 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. + +End-to-End Pipelines +===================== + +The Dags in this guide combine several patterns from the operator and hook guides into one +production-shaped pipeline. Each section explains the architecture -- how the Dags are split, +why, and how the pieces are wired together -- rather than the mechanics of a single operator, +which the linked guides already cover. Read the full source for the runnable Dag. + +SEC 10-K financial analysis +---------------------------- + +`example_llamaindex_10k.py `__ and +`example_langchain_10k.py `__ +compare companies' SEC 10-K filings, fetched live from EDGAR. Each file splits the work into two +Dags: + +- An **indexing Dag** (``schedule="@weekly"``) that fetches each company's latest 10-K and + persists an index per ticker to disk. +- An **analysis Dag** (``schedule=None``, triggered manually or on demand) that decomposes a + comparison question, retrieves against the indexes, and produces a reviewed report. + +The two Dags are not linked by an Asset or XCom -- they agree only on a shared on-disk index +path (``INDEX_BASE_DIR/{ticker}``). Run the indexing Dag at least once before triggering +analysis for a given company. + +Both files build the same analysis shape; the only structural difference is that LlamaIndex has +dedicated ``LlamaIndexEmbeddingOperator``/``LlamaIndexRetrievalOperator`` classes, while +LangChain does not yet, so the LangChain file's indexing and retrieval steps are plain ``@task`` +functions calling :class:`~airflow.providers.common.ai.hooks.langchain.LangChainHook` and FAISS +directly. + +Analysis Dag flow: + +1. A human confirms or edits the comparison question and the tickers to compare + (``HITLEntryOperator``). +2. ``@task.llm`` decomposes the question into one sub-question per company -- the LLM decides how + many sub-questions are needed, not the Dag author: + + .. exampleinclude:: /../../ai/src/airflow/providers/common/ai/example_dags/example_llamaindex_10k.py + :language: python + :start-after: [START 10k_decompose] + :end-before: [END 10k_decompose] + +3. Dynamic Task Mapping fans retrieval out, one mapped task per sub-question, each querying that + company's index: + + .. exampleinclude:: /../../ai/src/airflow/providers/common/ai/example_dags/example_llamaindex_10k.py + :language: python + :start-after: [START 10k_dtm_retrieval] + :end-before: [END 10k_dtm_retrieval] + +4. The mapped results are zipped back to their sub-questions (mapped task outputs preserve input + order) and synthesized into a structured ``AnalysisReport`` by an ``LLMOperator`` bounded by + ``UsageLimits``. +5. ``ApprovalOperator`` gates the report before hand-off. + +See :doc:`operators/llamaindex_embedding`, :doc:`operators/llamaindex_retrieval`, and +:doc:`hooks/langchain` for how the individual operators/hooks behave. + +Natural-language survey analysis +---------------------------------- + +`example_llm_survey_analysis.py `__ +answers natural-language questions over a CSV with the same core three tasks -- +``LLMSQLQueryOperator`` generates SQL, ``AnalyticsOperator`` runs it, a ``@task`` extracts the +rows (see :doc:`operators/llm_sql` for how those two operators work together). The file defines +two Dags around that core, shaped for different operating modes: + +- ``example_llm_survey_interactive`` (``schedule=None``) -- a human confirms or edits the + question before SQL generation (``HITLEntryOperator``) and approves the extracted result + afterward (``ApprovalOperator``). It assumes the CSV is already in place. +- ``example_llm_survey_scheduled`` (``schedule="@monthly"``) -- downloads the CSV itself + (``HttpOperator``), validates its schema against a reference CSV + (``LLMSchemaCompareOperator``) before generating SQL, and ends by emailing or logging the + result. No human gate -- suited to recurring reporting. + +Agentic multi-dimensional survey synthesis +--------------------------------------------- + +`example_llm_survey_agentic.py `__ +answers a research question that a single SQL query cannot -- one that spans several +dimensions (executor, deployment, cloud, Airflow version) -- by fanning both SQL generation and +execution out with Dynamic Task Mapping, one mapped pair per dimension, then synthesizing: + +1. ``decompose_question`` returns one sub-question per dimension. +2. ``LLMSQLQueryOperator.partial(...).expand(prompt=sub_questions)`` generates SQL for each + dimension as an independent mapped task instance. +3. ``AnalyticsOperator.partial(...).expand(...)`` runs each query. If one dimension's query + fails, only that mapped instance retries -- the other three keep their results. +4. ``collect_results`` zips the dimension labels back onto the (order-preserved) results. +5. An ``LLMOperator`` synthesizes the four labeled result sets into one narrative, then + ``ApprovalOperator`` gates it. + +The Dag's own docstring explains the reasoning for fanning out rather than looping inside a +single LLM call: each sub-query becomes a named, logged task instance instead of a hidden tool +call, so a failure is isolated and retryable, and every intermediate result is visible in XCom +instead of being an opaque step inside an LLM reasoning loop. + +AIP progress tracking: pipeline vs. agent +-------------------------------------------- + +`example_aip_progress_tracker.py `__ +tracks the same thing -- progress on a set of Airflow Improvement Proposals, checked against +Confluence specs and GitHub activity -- with two different architectures, to compare the +tradeoffs directly: + +- ``example_aip_progress_tracker`` -- a **deterministic pipeline**. Evidence is gathered by + fixed tasks; dynamically-mapped ``LLMOperator`` calls analyze each AIP with structured output; + the per-AIP analyses are synthesized into one report; a second ``LLMOperator`` validates that + synthesis against the raw evidence and flags unsupported claims; a plain-Python task applies + only the flagged corrections mechanically (string/regex replacement, no LLM involved); a human + reviews the corrected report. +- ``example_aip_progress_tracker_skills`` -- an **autonomous agent**. A single ``AgentOperator`` + loaded with an Agent Skill (``AgentSkillsToolset``) and a handful of custom tool functions + decides its own evidence-gathering strategy and tool-call order, then the same human review + gate. + +The pipeline's three-layer defense against hallucination is the pattern worth taking away: +structured LLM output, then a second LLM call whose only job is to judge the first against the +original evidence, then a deterministic (non-LLM) step that applies exactly what was flagged: + +.. exampleinclude:: /../../ai/src/airflow/providers/common/ai/example_dags/example_aip_progress_tracker.py + :language: python + :start-after: [START aip_tracker_validation] + :end-before: [END aip_tracker_validation] + +The agent variant collapses that whole graph into one operator call, trading auditability of +each step for simplicity and letting the model decide how to use its tools: + +.. exampleinclude:: /../../ai/src/airflow/providers/common/ai/example_dags/example_aip_progress_tracker.py + :language: python + :start-after: [START aip_tracker_skills_operator] + :end-before: [END aip_tracker_skills_operator] + +Reach for the pipeline when accuracy matters and every step needs to be auditable; reach for the +agent when the problem is open-ended enough that a fixed task graph would just be re-encoding +what the model can already work out from the skill instructions. diff --git a/providers/common/ai/docs/examples.rst b/providers/common/ai/docs/examples.rst index 11ce55ae4463a..9c69ca5b87132 100644 --- a/providers/common/ai/docs/examples.rst +++ b/providers/common/ai/docs/examples.rst @@ -39,8 +39,7 @@ Single-prompt tasks - What it shows * - :doc:`operators/llm` - ``@task.llm`` for text, structured output, classification, and dynamic task mapping over - LLM results (``example_llm.py``, ``example_llm_classification.py``, - ``example_llm_analysis_pipeline.py``). + LLM results (``example_llm.py``, ``example_llm_classification.py``, ``example_llm_analysis_pipeline.py``). * - :doc:`operators/llm_branch` - ``@task.llm_branch`` picking which downstream task runs. * - :doc:`operators/llm_file_analysis` @@ -99,8 +98,8 @@ Retrieval & document processing End-to-end scenarios --------------------- -Production-shaped Dags that combine several of the patterns above into one pipeline. These don't -have a dedicated guide yet -- read the source directly: +Production-shaped Dags that combine several of the patterns above into one pipeline. See +:doc:`end_to_end_pipelines` for the architecture behind each one. .. list-table:: :header-rows: 1 diff --git a/providers/common/ai/docs/index.rst b/providers/common/ai/docs/index.rst index c01522e37f7b8..bbd12c9100ac1 100644 --- a/providers/common/ai/docs/index.rst +++ b/providers/common/ai/docs/index.rst @@ -80,6 +80,7 @@ OpenAI, Anthropic, or other pydantic-ai-supported connection: Toolsets Operators Examples + End-to-End Pipelines Retry Policies HITL Review Observability From 1214508f294ce4d442cd771c8d0e1e981b3da1c4 Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Thu, 9 Jul 2026 15:45:26 +0800 Subject: [PATCH 3/8] Add prek hook to check example Dag link --- providers/common/ai/.pre-commit-config.yaml | 10 ++ .../check_common_ai_examples_referenced.py | 125 ++++++++++++++++++ ...est_check_common_ai_examples_referenced.py | 109 +++++++++++++++ 3 files changed, 244 insertions(+) create mode 100755 scripts/ci/prek/check_common_ai_examples_referenced.py create mode 100644 scripts/tests/ci/prek/test_check_common_ai_examples_referenced.py diff --git a/providers/common/ai/.pre-commit-config.yaml b/providers/common/ai/.pre-commit-config.yaml index 9676e56363da3..a8d3797cfbcd9 100644 --- a/providers/common/ai/.pre-commit-config.yaml +++ b/providers/common/ai/.pre-commit-config.yaml @@ -56,3 +56,13 @@ repos: (?x) ^src/airflow/providers/common/ai/plugins/.*\.py$| ^tests/unit/common/ai/plugins/.*\.py$ + - id: check-common-ai-examples-referenced + name: Check common.ai example Dags are referenced in examples.rst + description: Every example_dags/*.py file must be reachable from docs/examples.rst + entry: ../../../scripts/ci/prek/check_common_ai_examples_referenced.py + language: python + pass_filenames: false + files: > + (?x) + ^src/airflow/providers/common/ai/example_dags/.*\.py$| + ^docs/.*\.rst$ diff --git a/scripts/ci/prek/check_common_ai_examples_referenced.py b/scripts/ci/prek/check_common_ai_examples_referenced.py new file mode 100755 index 0000000000000..b700a116f867e --- /dev/null +++ b/scripts/ci/prek/check_common_ai_examples_referenced.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python +# +# 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. +# /// script +# requires-python = ">=3.10" +# dependencies = [ +# "rich>=13.6.0", +# ] +# /// +"""Check that every common.ai example Dag is reachable from docs/examples.rst. + +``examples.rst`` does not always link an example file directly -- for scenarios that +already have a dedicated guide (an operator/hook page under ``docs/operators`` or +``docs/hooks``), it links the guide via ``:doc:``/``:ref:`` and the guide embeds the +example with a ``literalinclude``/``exampleinclude`` directive instead. So "referenced" +here means reachable by following ``:doc:``/``:ref:`` links out of ``examples.rst``, +not necessarily mentioned by filename on that page itself. +""" + +from __future__ import annotations + +import re +import sys +from pathlib import Path + +from common_prek_utils import AIRFLOW_PROVIDERS_ROOT_PATH, console + +COMMON_AI_ROOT_PATH = AIRFLOW_PROVIDERS_ROOT_PATH / "common" / "ai" +DOCS_PATH = COMMON_AI_ROOT_PATH / "docs" +EXAMPLES_RST_PATH = DOCS_PATH / "examples.rst" +EXAMPLE_DAGS_PATH = COMMON_AI_ROOT_PATH / "src" / "airflow" / "providers" / "common" / "ai" / "example_dags" + +EXAMPLE_FILENAME_RE = re.compile(r"\bexample_[A-Za-z0-9_]+\.py\b") +DOC_ROLE_RE = re.compile(r":doc:`(?:[^`<]+<)?([^`<>]+)>?`") +REF_ROLE_RE = re.compile(r":ref:`(?:[^`<]+<)?([^`<>]+)>?`") + + +def get_example_dag_filenames() -> set[str]: + return {path.name for path in EXAMPLE_DAGS_PATH.glob("example_*.py")} + + +def resolve_doc_target(target: str, referring_file: Path) -> Path | None: + """Resolve a ``:doc:`` role target to an ``.rst`` file, Sphinx-style.""" + if target.startswith("/"): + rst_path = DOCS_PATH / target.lstrip("/") + else: + rst_path = referring_file.parent / target + rst_path = rst_path.with_suffix(".rst") + return rst_path if rst_path.is_file() else None + + +def resolve_ref_target(label: str) -> Path | None: + """Resolve a ``:ref:`` role target by finding its ``.. _label:`` definition.""" + label_marker = f".. _{label}:" + for rst_path in DOCS_PATH.rglob("*.rst"): + if label_marker in rst_path.read_text(encoding="utf-8"): + return rst_path + return None + + +def find_referenced_example_filenames() -> set[str]: + """Walk :doc:/:ref: links out of examples.rst, collecting example filenames seen.""" + referenced: set[str] = set() + visited: set[Path] = set() + queue = [EXAMPLES_RST_PATH] + while queue: + rst_path = queue.pop() + if rst_path in visited or not rst_path.is_file(): + continue + visited.add(rst_path) + text = rst_path.read_text(encoding="utf-8") + referenced.update(EXAMPLE_FILENAME_RE.findall(text)) + for target in DOC_ROLE_RE.findall(text): + resolved = resolve_doc_target(target, rst_path) + if resolved is not None: + queue.append(resolved) + for label in REF_ROLE_RE.findall(text): + resolved = resolve_ref_target(label) + if resolved is not None: + queue.append(resolved) + return referenced + + +def main() -> int: + if not EXAMPLES_RST_PATH.is_file(): + console.print(f"[red]Cannot find {EXAMPLES_RST_PATH}.[/]") + return 1 + + all_examples = get_example_dag_filenames() + referenced_examples = find_referenced_example_filenames() + missing = sorted(all_examples - referenced_examples) + + if missing: + console.print( + f"[red]The following example Dags are not reachable from {EXAMPLES_RST_PATH}[/] " + "[red](directly, or via a :doc:/:ref: link to a guide that embeds them):[/]" + ) + for filename in missing: + console.print(f"[red] - {filename}[/]") + console.print( + "[yellow]Add each new example Dag to examples.rst, either directly or through a " + "guide page it links to.[/]" + ) + return 1 + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/tests/ci/prek/test_check_common_ai_examples_referenced.py b/scripts/tests/ci/prek/test_check_common_ai_examples_referenced.py new file mode 100644 index 0000000000000..c39b7d97a6b9e --- /dev/null +++ b/scripts/tests/ci/prek/test_check_common_ai_examples_referenced.py @@ -0,0 +1,109 @@ +# 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. +"""Unit tests for ``scripts/ci/prek/check_common_ai_examples_referenced.py``.""" + +from __future__ import annotations + +import check_common_ai_examples_referenced as checker +import pytest + + +@pytest.fixture +def docs_tree(tmp_path, monkeypatch): + """Build a fake docs/example_dags tree and point the module's path constants at it.""" + example_dags_path = tmp_path / "example_dags" + example_dags_path.mkdir() + docs_path = tmp_path / "docs" + docs_path.mkdir() + examples_rst_path = docs_path / "examples.rst" + + for name in ("example_direct.py", "example_via_doc.py", "example_via_ref.py", "example_missing.py"): + (example_dags_path / name).write_text("# placeholder\n") + + examples_rst_path.write_text( + "Examples\n" + "========\n\n" + "Source: `example_direct.py `__.\n\n" + "See :doc:`guide_a` and :ref:`Some label `.\n" + ) + (docs_path / "guide_a.rst").write_text( + "Guide A\n=======\n\n.. exampleinclude:: /example_dags/example_via_doc.py\n :language: python\n" + ) + (docs_path / "guide_b.rst").write_text( + "Guide B\n" + "=======\n\n" + ".. _some-label:\n\n" + ".. exampleinclude:: /example_dags/example_via_ref.py\n" + " :language: python\n" + ) + + monkeypatch.setattr(checker, "DOCS_PATH", docs_path) + monkeypatch.setattr(checker, "EXAMPLES_RST_PATH", examples_rst_path) + monkeypatch.setattr(checker, "EXAMPLE_DAGS_PATH", example_dags_path) + return tmp_path + + +class TestGetExampleDagFilenames: + def test_lists_example_files_only(self, docs_tree): + assert checker.get_example_dag_filenames() == { + "example_direct.py", + "example_via_doc.py", + "example_via_ref.py", + "example_missing.py", + } + + +class TestResolveDocTarget: + def test_relative_target_resolves_next_to_referring_file(self, docs_tree): + referring_file = docs_tree / "docs" / "examples.rst" + resolved = checker.resolve_doc_target("guide_a", referring_file) + assert resolved == docs_tree / "docs" / "guide_a.rst" + + def test_absolute_target_resolves_from_docs_root(self, docs_tree): + referring_file = docs_tree / "docs" / "examples.rst" + resolved = checker.resolve_doc_target("/guide_a", referring_file) + assert resolved == docs_tree / "docs" / "guide_a.rst" + + def test_missing_target_returns_none(self, docs_tree): + referring_file = docs_tree / "docs" / "examples.rst" + assert checker.resolve_doc_target("does_not_exist", referring_file) is None + + +class TestResolveRefTarget: + def test_finds_file_defining_label(self, docs_tree): + assert checker.resolve_ref_target("some-label") == docs_tree / "docs" / "guide_b.rst" + + def test_missing_label_returns_none(self, docs_tree): + assert checker.resolve_ref_target("no-such-label") is None + + +class TestFindReferencedExampleFilenames: + def test_follows_doc_and_ref_links_from_examples_rst(self, docs_tree): + assert checker.find_referenced_example_filenames() == { + "example_direct.py", + "example_via_doc.py", + "example_via_ref.py", + } + + +class TestMain: + def test_fails_when_an_example_is_unreferenced(self, docs_tree): + assert checker.main() == 1 + + def test_passes_when_every_example_is_reachable(self, docs_tree): + (docs_tree / "example_dags" / "example_missing.py").unlink() + assert checker.main() == 0 From 0720e53db2a34bdcb4da4b6908b40f205ca11b2c Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Thu, 9 Jul 2026 17:05:06 +0800 Subject: [PATCH 4/8] fixup! fixup! Support delegating providers release process to non-PMC committers (#69417) --- providers/common/ai/docs/end_to_end_pipelines.rst | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/providers/common/ai/docs/end_to_end_pipelines.rst b/providers/common/ai/docs/end_to_end_pipelines.rst index fc02b23bcb016..cbfdf1c331087 100644 --- a/providers/common/ai/docs/end_to_end_pipelines.rst +++ b/providers/common/ai/docs/end_to_end_pipelines.rst @@ -118,8 +118,7 @@ AIP progress tracking: pipeline vs. agent `example_aip_progress_tracker.py `__ tracks the same thing -- progress on a set of Airflow Improvement Proposals, checked against -Confluence specs and GitHub activity -- with two different architectures, to compare the -tradeoffs directly: +Confluence specs and GitHub activity -- with two different architectures, to compare the tradeoff directly: - ``example_aip_progress_tracker`` -- a **deterministic pipeline**. Evidence is gathered by fixed tasks; dynamically-mapped ``LLMOperator`` calls analyze each AIP with structured output; From 938e4d93dd7377e46aae4ac64bccc4b2f28e8432 Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Thu, 9 Jul 2026 17:39:37 +0800 Subject: [PATCH 5/8] fixup! fixup! Support delegating providers release process to non-PMC committers (#69417) --- .../common/ai/docs/end_to_end_pipelines.rst | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/providers/common/ai/docs/end_to_end_pipelines.rst b/providers/common/ai/docs/end_to_end_pipelines.rst index cbfdf1c331087..4ddc06a621e6a 100644 --- a/providers/common/ai/docs/end_to_end_pipelines.rst +++ b/providers/common/ai/docs/end_to_end_pipelines.rst @@ -23,6 +23,29 @@ production-shaped pipeline. Each section explains the architecture -- how the Da why, and how the pieces are wired together -- rather than the mechanics of a single operator, which the linked guides already cover. Read the full source for the runnable Dag. +LlamaIndex RAG shapes +----------------------- + +`example_llamaindex_rag.py `__ +walks the same load -> embed -> retrieve -> answer pattern through three shapes, from simplest to +production-shaped: + +- ``example_llamaindex_rag_pipeline`` (``schedule=None``) -- everything in one Dag: + ``DocumentLoaderOperator`` parses local files, ``LlamaIndexEmbeddingOperator`` chunks and + persists the index, ``LlamaIndexRetrievalOperator`` retrieves for a fixed question, and an + ``LLMOperator`` synthesizes the answer. +- ``example_llamaindex_index_pdf`` / ``example_llamaindex_query`` -- the load/embed step moved + into its own weekly Dag (``schedule="@weekly"``) so the index stays fresh as PDFs arrive, while + a second Dag (``schedule=None``, triggered with a ``question`` param) retrieves and answers on + demand against the persisted index -- the same index/query split the SEC 10-K pipelines below + use, without their multi-company retrieval fan-out. +- ``example_llamaindex_multi_source`` -- two ``DocumentLoaderOperator`` calls tag documents from + different sources via ``metadata_fields`` before merging and embedding them into one index, for + filtered retrieval downstream. + +See :doc:`operators/document_loader`, :doc:`operators/llamaindex_embedding`, and +:doc:`operators/llamaindex_retrieval` for how the individual operators work. + SEC 10-K financial analysis ---------------------------- From da0560c70a4c47530c2759267bbfc53fc27ee348 Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Thu, 9 Jul 2026 17:39:37 +0800 Subject: [PATCH 6/8] fixup! fixup! Support delegating providers release process to non-PMC committers (#69417) --- providers/common/ai/docs/examples.rst | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/providers/common/ai/docs/examples.rst b/providers/common/ai/docs/examples.rst index 9c69ca5b87132..a74550d5e60b0 100644 --- a/providers/common/ai/docs/examples.rst +++ b/providers/common/ai/docs/examples.rst @@ -39,7 +39,10 @@ Single-prompt tasks - What it shows * - :doc:`operators/llm` - ``@task.llm`` for text, structured output, classification, and dynamic task mapping over - LLM results (``example_llm.py``, ``example_llm_classification.py``, ``example_llm_analysis_pipeline.py``). + LLM results + (`example_llm.py `__, + `example_llm_classification.py `__, + `example_llm_analysis_pipeline.py `__). * - :doc:`operators/llm_branch` - ``@task.llm_branch`` picking which downstream task runs. * - :doc:`operators/llm_file_analysis` From 86cc962acabe83f2e40d9fb23a3fb545cbe0019b Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Thu, 9 Jul 2026 17:39:37 +0800 Subject: [PATCH 7/8] fixup! fixup! Support delegating providers release process to non-PMC committers (#69417) --- providers/common/ai/docs/index.rst | 1 - 1 file changed, 1 deletion(-) diff --git a/providers/common/ai/docs/index.rst b/providers/common/ai/docs/index.rst index bbd12c9100ac1..c01522e37f7b8 100644 --- a/providers/common/ai/docs/index.rst +++ b/providers/common/ai/docs/index.rst @@ -80,7 +80,6 @@ OpenAI, Anthropic, or other pydantic-ai-supported connection: Toolsets Operators Examples - End-to-End Pipelines Retry Policies HITL Review Observability From 801519a14abcff7acfb3ec64dbdc82ff8be40047 Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Thu, 9 Jul 2026 17:46:13 +0800 Subject: [PATCH 8/8] fixup! fixup! fixup! Support delegating providers release process to non-PMC committers (#69417) --- providers/common/ai/.pre-commit-config.yaml | 10 -- .../check_common_ai_examples_referenced.py | 125 ------------------ ...est_check_common_ai_examples_referenced.py | 109 --------------- 3 files changed, 244 deletions(-) delete mode 100755 scripts/ci/prek/check_common_ai_examples_referenced.py delete mode 100644 scripts/tests/ci/prek/test_check_common_ai_examples_referenced.py diff --git a/providers/common/ai/.pre-commit-config.yaml b/providers/common/ai/.pre-commit-config.yaml index a8d3797cfbcd9..9676e56363da3 100644 --- a/providers/common/ai/.pre-commit-config.yaml +++ b/providers/common/ai/.pre-commit-config.yaml @@ -56,13 +56,3 @@ repos: (?x) ^src/airflow/providers/common/ai/plugins/.*\.py$| ^tests/unit/common/ai/plugins/.*\.py$ - - id: check-common-ai-examples-referenced - name: Check common.ai example Dags are referenced in examples.rst - description: Every example_dags/*.py file must be reachable from docs/examples.rst - entry: ../../../scripts/ci/prek/check_common_ai_examples_referenced.py - language: python - pass_filenames: false - files: > - (?x) - ^src/airflow/providers/common/ai/example_dags/.*\.py$| - ^docs/.*\.rst$ diff --git a/scripts/ci/prek/check_common_ai_examples_referenced.py b/scripts/ci/prek/check_common_ai_examples_referenced.py deleted file mode 100755 index b700a116f867e..0000000000000 --- a/scripts/ci/prek/check_common_ai_examples_referenced.py +++ /dev/null @@ -1,125 +0,0 @@ -#!/usr/bin/env python -# -# 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. -# /// script -# requires-python = ">=3.10" -# dependencies = [ -# "rich>=13.6.0", -# ] -# /// -"""Check that every common.ai example Dag is reachable from docs/examples.rst. - -``examples.rst`` does not always link an example file directly -- for scenarios that -already have a dedicated guide (an operator/hook page under ``docs/operators`` or -``docs/hooks``), it links the guide via ``:doc:``/``:ref:`` and the guide embeds the -example with a ``literalinclude``/``exampleinclude`` directive instead. So "referenced" -here means reachable by following ``:doc:``/``:ref:`` links out of ``examples.rst``, -not necessarily mentioned by filename on that page itself. -""" - -from __future__ import annotations - -import re -import sys -from pathlib import Path - -from common_prek_utils import AIRFLOW_PROVIDERS_ROOT_PATH, console - -COMMON_AI_ROOT_PATH = AIRFLOW_PROVIDERS_ROOT_PATH / "common" / "ai" -DOCS_PATH = COMMON_AI_ROOT_PATH / "docs" -EXAMPLES_RST_PATH = DOCS_PATH / "examples.rst" -EXAMPLE_DAGS_PATH = COMMON_AI_ROOT_PATH / "src" / "airflow" / "providers" / "common" / "ai" / "example_dags" - -EXAMPLE_FILENAME_RE = re.compile(r"\bexample_[A-Za-z0-9_]+\.py\b") -DOC_ROLE_RE = re.compile(r":doc:`(?:[^`<]+<)?([^`<>]+)>?`") -REF_ROLE_RE = re.compile(r":ref:`(?:[^`<]+<)?([^`<>]+)>?`") - - -def get_example_dag_filenames() -> set[str]: - return {path.name for path in EXAMPLE_DAGS_PATH.glob("example_*.py")} - - -def resolve_doc_target(target: str, referring_file: Path) -> Path | None: - """Resolve a ``:doc:`` role target to an ``.rst`` file, Sphinx-style.""" - if target.startswith("/"): - rst_path = DOCS_PATH / target.lstrip("/") - else: - rst_path = referring_file.parent / target - rst_path = rst_path.with_suffix(".rst") - return rst_path if rst_path.is_file() else None - - -def resolve_ref_target(label: str) -> Path | None: - """Resolve a ``:ref:`` role target by finding its ``.. _label:`` definition.""" - label_marker = f".. _{label}:" - for rst_path in DOCS_PATH.rglob("*.rst"): - if label_marker in rst_path.read_text(encoding="utf-8"): - return rst_path - return None - - -def find_referenced_example_filenames() -> set[str]: - """Walk :doc:/:ref: links out of examples.rst, collecting example filenames seen.""" - referenced: set[str] = set() - visited: set[Path] = set() - queue = [EXAMPLES_RST_PATH] - while queue: - rst_path = queue.pop() - if rst_path in visited or not rst_path.is_file(): - continue - visited.add(rst_path) - text = rst_path.read_text(encoding="utf-8") - referenced.update(EXAMPLE_FILENAME_RE.findall(text)) - for target in DOC_ROLE_RE.findall(text): - resolved = resolve_doc_target(target, rst_path) - if resolved is not None: - queue.append(resolved) - for label in REF_ROLE_RE.findall(text): - resolved = resolve_ref_target(label) - if resolved is not None: - queue.append(resolved) - return referenced - - -def main() -> int: - if not EXAMPLES_RST_PATH.is_file(): - console.print(f"[red]Cannot find {EXAMPLES_RST_PATH}.[/]") - return 1 - - all_examples = get_example_dag_filenames() - referenced_examples = find_referenced_example_filenames() - missing = sorted(all_examples - referenced_examples) - - if missing: - console.print( - f"[red]The following example Dags are not reachable from {EXAMPLES_RST_PATH}[/] " - "[red](directly, or via a :doc:/:ref: link to a guide that embeds them):[/]" - ) - for filename in missing: - console.print(f"[red] - {filename}[/]") - console.print( - "[yellow]Add each new example Dag to examples.rst, either directly or through a " - "guide page it links to.[/]" - ) - return 1 - - return 0 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/scripts/tests/ci/prek/test_check_common_ai_examples_referenced.py b/scripts/tests/ci/prek/test_check_common_ai_examples_referenced.py deleted file mode 100644 index c39b7d97a6b9e..0000000000000 --- a/scripts/tests/ci/prek/test_check_common_ai_examples_referenced.py +++ /dev/null @@ -1,109 +0,0 @@ -# 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. -"""Unit tests for ``scripts/ci/prek/check_common_ai_examples_referenced.py``.""" - -from __future__ import annotations - -import check_common_ai_examples_referenced as checker -import pytest - - -@pytest.fixture -def docs_tree(tmp_path, monkeypatch): - """Build a fake docs/example_dags tree and point the module's path constants at it.""" - example_dags_path = tmp_path / "example_dags" - example_dags_path.mkdir() - docs_path = tmp_path / "docs" - docs_path.mkdir() - examples_rst_path = docs_path / "examples.rst" - - for name in ("example_direct.py", "example_via_doc.py", "example_via_ref.py", "example_missing.py"): - (example_dags_path / name).write_text("# placeholder\n") - - examples_rst_path.write_text( - "Examples\n" - "========\n\n" - "Source: `example_direct.py `__.\n\n" - "See :doc:`guide_a` and :ref:`Some label `.\n" - ) - (docs_path / "guide_a.rst").write_text( - "Guide A\n=======\n\n.. exampleinclude:: /example_dags/example_via_doc.py\n :language: python\n" - ) - (docs_path / "guide_b.rst").write_text( - "Guide B\n" - "=======\n\n" - ".. _some-label:\n\n" - ".. exampleinclude:: /example_dags/example_via_ref.py\n" - " :language: python\n" - ) - - monkeypatch.setattr(checker, "DOCS_PATH", docs_path) - monkeypatch.setattr(checker, "EXAMPLES_RST_PATH", examples_rst_path) - monkeypatch.setattr(checker, "EXAMPLE_DAGS_PATH", example_dags_path) - return tmp_path - - -class TestGetExampleDagFilenames: - def test_lists_example_files_only(self, docs_tree): - assert checker.get_example_dag_filenames() == { - "example_direct.py", - "example_via_doc.py", - "example_via_ref.py", - "example_missing.py", - } - - -class TestResolveDocTarget: - def test_relative_target_resolves_next_to_referring_file(self, docs_tree): - referring_file = docs_tree / "docs" / "examples.rst" - resolved = checker.resolve_doc_target("guide_a", referring_file) - assert resolved == docs_tree / "docs" / "guide_a.rst" - - def test_absolute_target_resolves_from_docs_root(self, docs_tree): - referring_file = docs_tree / "docs" / "examples.rst" - resolved = checker.resolve_doc_target("/guide_a", referring_file) - assert resolved == docs_tree / "docs" / "guide_a.rst" - - def test_missing_target_returns_none(self, docs_tree): - referring_file = docs_tree / "docs" / "examples.rst" - assert checker.resolve_doc_target("does_not_exist", referring_file) is None - - -class TestResolveRefTarget: - def test_finds_file_defining_label(self, docs_tree): - assert checker.resolve_ref_target("some-label") == docs_tree / "docs" / "guide_b.rst" - - def test_missing_label_returns_none(self, docs_tree): - assert checker.resolve_ref_target("no-such-label") is None - - -class TestFindReferencedExampleFilenames: - def test_follows_doc_and_ref_links_from_examples_rst(self, docs_tree): - assert checker.find_referenced_example_filenames() == { - "example_direct.py", - "example_via_doc.py", - "example_via_ref.py", - } - - -class TestMain: - def test_fails_when_an_example_is_unreferenced(self, docs_tree): - assert checker.main() == 1 - - def test_passes_when_every_example_is_reachable(self, docs_tree): - (docs_tree / "example_dags" / "example_missing.py").unlink() - assert checker.main() == 0