Airflow2 Taskflow API移除collect_files到update_process_info自动依赖

自动生成的collect_files到update_process_info直连边，是Airflow 2 Taskflow API的默认规则导致的：只要在调用被@task装饰的任务函数时，传入了其他Taskflow任务返回的XComArg对象，Taskflow就会自动在两个任务间建立上下游依赖，不需要手动书写依赖关系。
当前代码中upi = update_process_info(cf)的写法，把collect_files()返回的XComArg对象cf作为入参传给了update_process_info，因此触发自动连线逻辑，生成了非预期的直连依赖。

核心思路是切断“将cf作为入参传给update_process_info”这个会触发自动依赖的传参动作，在保留正确依赖关系的前提下移除多余的直连边。
具体实现步骤：

• 调整update_process_info的函数定义，移除函数签名中来自collect_files的cf入参，从根源上避免触发自动连线逻辑。
• 如果update_process_info的业务逻辑确实需要用到collect_files返回的文件列表数据，不要通过函数传参的方式获取，而是在任务函数内部通过Airflow上下文手动拉取对应XCom值即可。因为collect_files本身就是所有plot_files任务的上游，当所有plot_files执行完成触发update_process_info时，collect_files的XCom值已经完整生成，手动拉取可以正常拿到结果，不会影响业务逻辑。

调整后的参考代码：

import airflow
from airflow.decorators import task

# 原有collect_files、plot_files的Taskflow定义保持不变
# 调整update_process_info的函数定义
@task
def update_process_info(**context):
    # 内部手动拉取collect_files的返回值，不会触发自动依赖连线
    ti = context['ti']
    collected_files = ti.xcom_pull(task_ids='collect_files')
    # 原有业务处理逻辑
    pass

with airflow.DAG("plot", schedule_interval=None, default_args=default_args) as dag:
    cf = collect_files()
    # 调用update_process_info时不再传入cf对象，不会生成collect_files到upi的直连依赖
    upi = update_process_info()

    for i in range(0, max_parallel_plot_tasks):
        # 将cf传给plot_files是符合预期的：collect_files本就是plot_files的上游
        current_plot_task = plot_files(cf, i, int(max_parallel_plot_tasks))
        # 显式建立plot_files到update_process_info的依赖关系
        current_plot_task >> upi

调整完成后DAG的依赖结构完全符合预期：collect_files分别连接到所有并行的plot_files任务，每个plot_files任务再连接到update_process_info，不存在collect_files直连update_process_info的多余边。

内容的提问来源于stack exchange，提问作者opfau