动态任务生成函数中的xcom_pull不起作用

可以使用jinja2模板语言来动态生成任务，并使用task_instance.xcom_pull方法获取上一个任务的输出。下面是一个示例代码：

from airflow.models import DAG, TaskInstance
from datetime import datetime
from airflow.operators.python_operator import PythonOperator
from jinja2 import Template

def create_task(task_id, dag):
    t = PythonOperator(
        task_id=task_id,
        python_callable=print_xcom_value,
        provide_context=True,
        dag=dag
    )

    return t

def print_xcom_value(**kwargs):
    ti = kwargs['ti']
    val = ti.xcom_pull(task_ids='previous_task')
    print(val)

dag = DAG(
    dag_id='dynamic_dag',
    default_args={
        'owner': 'airflow',
        'start_date': datetime(2022, 1, 1)
    },
    schedule_interval=None
)

for i in range(5):
    task_id = f'task_{i}'
    t = create_task(task_id, dag)

    if i == 0:
        # 第一个任务只是打印一个字符串，不需要xcom_push
        t.set_upstream(None)
    else:
        # 后面的任务可以使用jinja2模板语言动态生成任务，并使用task_instance.xcom_pull方法获取上一个任务的输出
        prev_task_id = f'task_{i-1}'
        task_template = Template("print_xcom_value_{{task_id}} = create_task('task_{{task_id}}', dag)\n"
                                "{{task_template}}\n"
                                "print_xcom_value_{{task_id}}.set_upstream(print_xcom_value_{{prev_task_id}})")
        task_template = task_template.render(task_id=i, prev_task_id=i-1, task_template='ti.xcom_push(key="value", value=101)\n')

        exec(task_template)