**Apache Airflow version**:
2.0 / master

**Environment**:

breeze

**What happened**:

Using this DAG:

```py
from airflow.operators.bash import BashOperator
from airflow.operators.python import task
from airflow.models import DAG
from airflow.utils.task_group import TaskGroup

@task
def show():
    print("Cats are awesome!")

with DAG(
    "using_task_group",
    default_args={'owner': 'airflow'},
    start_date=days_ago(2),
    schedule_interval=None,
) as dag3:
    start_task = BashOperator(
        task_id="start_task",
        bash_command="echo start",
    )

    end_task = BashOperator(
        task_id="end_task",
        bash_command="echo end",
    )

    with TaskGroup(group_id="show_tasks") as tg1:
        previous_show = show()
        for _ in range(100):
            next_show = show()
            previous_show >> next_show
            previous_show = next_show
```
I get:
```
Broken DAG: [/files/dags/test.py] Traceback (most recent call last):
  File "/opt/airflow/airflow/models/baseoperator.py", line 410, in __init__
    task_group.add(self)
  File "/opt/airflow/airflow/utils/task_group.py", line 140, in add
    raise DuplicateTaskIdFound(f"Task id '{key}' has already been added to the DAG")
airflow.exceptions.DuplicateTaskIdFound: Task id 'show_tasks.show' has already been added to the DAG
```

If I remove the task group the task are generated as expected.

**What you expected to happen**:

I expect to be able to generate tasks dynamically using TaskGroup and task decoratos.

**How to reproduce it**:

Use the DAG from above.

**Anything else we need to know**:

N/A

airflow/operators/python.py
if self.op_args or self.op_kwargs:
            with open(filename, 'wb') as file:
                self.pickling_library.dump({'args': self.op_args, 'kwargs': self.op_kwargs}, file)
    def _get_serializable_context_keys(self):
        def _is_airflow_env():
            return self.system_site_packages or 'apache-airflow' in self.requirements
        def _is_pendulum_env():
            return 'pendulum' in self.requirements and 'lazy_object_proxy' in self.requirements
        serializable_context_keys = self.BASE_SERIALIZABLE_CONTEXT_KEYS.copy()
        if _is_airflow_env():
            serializable_context_keys.update(self.AIRFLOW_SERIALIZABLE_CONTEXT_KEYS)
        if _is_pendulum_env() or _is_airflow_env():
            serializable_context_keys.update(self.PENDULUM_SERIALIZABLE_CONTEXT_KEYS)
        return serializable_context_keys
    def _write_string_args(self, filename):
        with open(filename, 'w') as file:
            file.write('\n'.join(map(str, self.string_args)))
    def _read_result(self, filename):
        if os.stat(filename).st_size == 0:
            return None
        with open(filename, 'rb') as file:
            try:
                return self.pickling_library.load(file)
            except ValueError:
                self.log.error(
                    "Error deserializing result. Note that result deserialization "
                    "is not supported across major Python versions."
                )
                raise
def get_current_context() -> Dict[str, Any]: