def trigger_check(context, dag_run_obj):
    workflow_name_var = context['ti'].xcom_pull(key='workflow_name')
    dag_run_obj.payload = {
        "workflow_name": workflow_name_var
    }
    if workflow_name_var == 'normal':
        dag_run_obj.trigger_dag_id = JOB_NAME
    elif workflow_name_var == 'reload':
        dag_run_obj.trigger_dag_id = JOB_NAME_RELOAD
    elif workflow_name_var == 'XXX':
        dag_run_obj.trigger_dag_id = XXXX
    return dag_run_obj
   trigger_check_task = TriggerDagRunOperator(
    task_id='trigger_check_task',
    python_callable= trigger_check,
    trigger_dag_id="",
    dag=dag
)

from datetime import datetime
from random import choice
from airflow import DAG
from airflow.decorators import branch_task, task
from airflow.models import TaskInstance
from airflow.operators.python import get_current_context
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
with DAG(
    dag_id="test_dag",
    schedule_interval=None,
    default_args={
        "start_date": datetime(2022, 1, 1),
    },
    render_template_as_native_obj=True,
    tags=["test"],
) as dag:
    dag.doc_md = __doc__
    workflow_to_task = {
        "normal": "task1",
        "reload": "task2",
        "XXX": "task3",
    }
    @task
    def get_workflow_name():
        return choice(list(workflow_to_task.keys()))
    @branch_task()
    def trigger_check():
        ti: TaskInstance = get_current_context()["ti"]
        workflow_name = ti.xcom_pull(task_ids="get_workflow_name")
        print(workflow_name)
        return workflow_to_task[workflow_name]
    task1 = TriggerDagRunOperator(
        task_id="task1",
        trigger_dag_id="normal",
    )
    task2 = TriggerDagRunOperator(
        task_id="task2",
        trigger_dag_id="reload",
    )
    task3 = TriggerDagRunOperator(
        task_id="task3",
        trigger_dag_id="XXX",
    )
    (get_workflow_name() >> trigger_check() >> [task1, task2, task3])