phanikumv commented on code in PR #74685: URL: https://github.com/apache/airflow/pull/74685#discussion_r4268029657
########## providers/common/ai/docs/why_common_ai.rst: ########## @@ -0,0 +1,201 @@ + .. 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/why-common-ai: + +Why Common AI +============= + +You can call a model from Airflow without this provider: put an agent framework or a +vendor SDK in a ``@task`` and it runs. The trouble starts when the agent does real work. +On retry, every model and tool call runs again and is billed again. A spend limit resets +with each attempt. Pausing for a person means wiring up a separate approval task, and +giving the agent tools means writing them yourself. This provider handles those cases +for agents, and its retry policies bring a model's judgment to any task that fails. + +For a single model call with no tools, where a retry costs little, a plain ``@task`` is +enough; see `When a plain task is enough`_. + +The same job, two ways +---------------------- + +An agent reads a reports bucket, queries the orders table and says whether the numbers +agree. Without this provider, a Pydantic AI agent in a ``@task`` looks like this, with +the model key in the worker's environment and the tools written by hand: + +.. code-block:: python + + from pydantic_ai import Agent + + from airflow.sdk import BaseHook, ObjectStoragePath, dag, task + + + @dag + def check_report_plain(): + @task(retries=2) + def check_report() -> str: + # The key comes from OPENAI_API_KEY in the worker's environment. + agent = Agent( + "openai:gpt-5", + instructions=( + "You check reports against the warehouse. Read the report files, query " + "the orders table, and say whether the numbers agree." + ), + ) + + @agent.tool_plain + def query(sql: str) -> list: + # Runs whatever SQL the model writes, on any table the connection can read. + return BaseHook.get_connection("warehouse").get_hook().get_records(sql) + + @agent.tool_plain + def read_report(name: str) -> str: + reports = ObjectStoragePath("s3://acme-reports/finance/", conn_id="aws_default") + return (reports / name).read_text() + + return agent.run_sync("Does the September report's revenue match the orders table?").output + + check_report() + +This is the same job with ``AgentOperator``. Its ``system_prompt`` is passed to the agent +as Pydantic AI ``instructions``: + +.. exampleinclude:: /../../ai/src/airflow/providers/common/ai/example_dags/example_why_common_ai.py + :language: python + :start-after: [START example_why_common_ai] + :end-before: [END example_why_common_ai] + +Both need an LLM connection or key, a ``warehouse`` connection with an ``orders`` table, +and a bucket of report files. What differs: + +- **The tools.** The hand-written ``query`` runs any SQL the model writes, including + ``DELETE``, on any table the connection can reach, and returns every row it gets. + ``SQLToolset`` allows only the ``orders`` table, rejects statements that write, and caps + the rows and bytes it returns. ``ObjectStorageToolset`` (experimental) keeps reads under + the prefix you gave it. Both run Airflow's secrets masker over results, which hides + values Airflow knows are secret, such as connection passwords, not personal data in your + tables. +- **The model.** The plain task names a vendor in code and reads its key from the + environment. ``AgentOperator`` takes both from a connection, so moving to another vendor + or adding a fallback is a connection change. See :doc:`model_providers`. +- **A retry.** The plain task starts again from the beginning and pays for every call + again. ``durable=True`` replays the calls that finished and continues from there. It is + experimental, and before Airflow 3.3 it needs ``[common.ai] durable_cache_path`` set; see + :doc:`durable_execution`. +- **Spend.** In the plain task, Pydantic AI's default limit of 50 model requests applies + to each attempt on its own, so three attempts can make up to 150. With ``usage_limits`` + on Airflow 3.3 and later, one total covers all three attempts, and the task fails once + it passes 30 requests or 200,000 tokens. Calls replayed from the durable cache do not + count. +- **The task log.** ``AgentOperator`` logs each tool call it makes. + +:doc:`frameworks/pydantic_ai` lists what ``AgentOperator`` adds on top of an agent you +run yourself, including when you keep your own agent and use only this provider's +toolsets. + +What a plain task leaves to you +------------------------------- + +.. list-table:: + :header-rows: 1 + :widths: 28 40 32 + + * - When this happens + - What this provider does + - Where it is documented + * - The task retries after the agent has already made model and tool calls. + - ``durable=True`` on ``AgentOperator`` replays the calls that finished and runs + the rest live. If the prompt, model or tools changed since the failed attempt, + the affected calls run again. Experimental. + - :doc:`durable_execution` + * - A run must not spend past a limit, retries included. + - ``usage_limits`` counts every attempt of the task instance together on + Airflow 3.3 and later, and fails the task once the total passes a limit. + Pydantic AI's own limits apply to one run at a time. + - :ref:`Usage budget across attempts <agent-usage-budget>` + * - A person has to approve, edit or reject a model's output before the Dag goes + on. + - ``require_approval=True`` on ``LLMOperator`` and the operators built on it + pauses the task after it generates output, without holding a worker slot. On + Airflow 3.1 and 3.2 the wait runs on the triggerer. Airflow 3.1 and later. A + plain task can get the same step from the standard provider's + ``HITLOperator`` placed after it. + - :doc:`approval_gates` + * - A person has to approve a tool call before it runs, such as a refund or a write + to a production table. + - A tool marked with Pydantic AI's approval API pauses ``AgentOperator`` before the + call, without holding a worker slot. Airflow 3.3 and later. Experimental. Review Comment: The example right above sets durable=True, but with durable set, `_supports_tool_approval()` returns False. If someone starts from this example and marks a tool requires_approval, the task fails on the first call, after it has already spent tokens. `tool_approval.rst` covers this, but this table is where people will pick features to combine. Worth calling it out here? -- 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]
