Airflow XCOM 拉入任务组

问：

我是 Airflow 的新手，在 Python 函数和任务组之间交换变量时遇到了一些问题。（我知道这不是气流的要点，但在我的情况下是必要的）。 为了更好地理解，我添加了这个代码片段，它应该传达我打算做什么：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.decorators import task_group
import logging
from pendulum import datetime

def push_function(**kwargs):
    files = ['a','b','c']
    kwargs['ti'].xcom_push(key='files', value=files)

with DAG(
    "tst",
    start_date=datetime(2023, 11, 7),
    schedule_interval="30 6 * * *",
    catchup=False,
) as dag:
    push = PythonOperator(
        task_id='push_task',
        python_callable=push_function,
        dag=dag,
    )

    @task_group(group_id="group")
    def pull_task(**kwargs):
        data= kwargs['ti'].xcom_pull(task_ids='push_task', key='files')
        logging.info(f"transfaired Variable: {folders_today}")
        for item in data:
                filepath = f"/tmp/{item}.xml"
                extract_load = SFTPOperator(
                    task_id=f"download_{item}",
                    ssh_conn_id="sftp",
                    remote_filepath=f"{item}.xml",
                    local_filepath=filepath,
                    operation="get",
                    create_intermediate_dirs=True
)
      
    push >> pull_task()

提供的代码 所以基本上我的问题是：在这种情况下，有没有办法让我的变量：“files”从函数“push_function”到函数“pull_task”？

像这样，这条线：

folders_today= kwargs['ti'].xcom_pull(task_ids='push_task', key='files')

给我错误：

KeyError: 'ti'

但我也尝试了其他方法，并且必须理解，任务组上下文显然在此函数中不可用。但我不确定，这意味着什么，以及我如何才能使用它。