[Misc] Replace Optional[X] with X | None syntax (#33332)

Signed-off-by: carlory <baofa.fan@daocloud.io>
Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

vllm/platforms/interface.py

@@ -7,7 +7,7 @@ import platform
 import random
 import sys
 from datetime import timedelta
-from typing import TYPE_CHECKING, Any, NamedTuple, Optional
+from typing import TYPE_CHECKING, Any, NamedTuple
 
 import numpy as np
 import torch
@@ -243,7 +243,7 @@ class Platform:
         cls,
         head_size: int,
         dtype: torch.dtype,
-        backend: Optional["AttentionBackendEnum"] = None,
+        backend: "AttentionBackendEnum | None" = None,
     ) -> "AttentionBackendEnum":
         """
         Get the vision attention backend class of a device.

vllm/platforms/rocm.py

@@ -3,7 +3,7 @@
 
 import os
 from functools import cache, lru_cache, wraps
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 import torch
 
@@ -356,7 +356,7 @@ class RocmPlatform(Platform):
         cls,
         head_size: int,
         dtype: torch.dtype,
-        backend: Optional["AttentionBackendEnum"] = None,
+        backend: "AttentionBackendEnum | None" = None,
     ) -> "AttentionBackendEnum":
         if backend is not None:
             assert backend in cls.get_supported_vit_attn_backends(), (

vllm/platforms/xpu.py

@@ -3,7 +3,7 @@
 
 import contextlib
 import os
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 import torch
 
@@ -88,7 +88,7 @@ class XPUPlatform(Platform):
         cls,
         head_size: int,
         dtype: torch.dtype,
-        backend: Optional["AttentionBackendEnum"] = None,
+        backend: "AttentionBackendEnum | None" = None,
     ) -> "AttentionBackendEnum":
         if backend is not None:
             assert backend in cls.get_supported_vit_attn_backends(), (

vllm/pooling_params.py

@@ -2,7 +2,7 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
 from copy import deepcopy
-from typing import Annotated, Any, Optional
+from typing import Annotated, Any
 
 import msgspec
 
@@ -80,7 +80,7 @@ class PoolingParams(
         return deepcopy(self)
 
     def verify(
-        self, task: PoolingTask, model_config: Optional["ModelConfig"] = None
+        self, task: PoolingTask, model_config: "ModelConfig | None" = None
     ) -> None:
         if self.task is None:
             self.task = task
@@ -106,7 +106,7 @@ class PoolingParams(
         self._verify_valid_parameters()
 
     def _merge_default_parameters(
-        self, model_config: Optional["ModelConfig"] = None
+        self, model_config: "ModelConfig | None" = None
     ) -> None:
         if model_config is None:
             return
@@ -160,7 +160,7 @@ class PoolingParams(
                 if getattr(self, k, None) is None:
                     setattr(self, k, getattr(pooler_config, k))
 
-    def _set_default_parameters(self, model_config: Optional["ModelConfig"]):
+    def _set_default_parameters(self, model_config: "ModelConfig | None"):
         if self.task in ["embed", "token_embed"]:
             if self.use_activation is None:
                 self.use_activation = True

vllm/profiler/layerwise_profile.py

@@ -5,7 +5,7 @@ import copy
 from collections import defaultdict
 from collections.abc import Callable
 from dataclasses import asdict, dataclass, field
-from typing import Any, Optional, TypeAlias
+from typing import Any, TypeAlias
 
 from torch._C._autograd import DeviceType, _KinetoEvent, _ProfilerResult
 from torch._C._profiler import _EventType, _ExperimentalConfig, _ProfilerEvent
@@ -31,7 +31,7 @@ except ImportError:
 @dataclass
 class _ModuleTreeNode:
     event: _ProfilerEvent
-    parent: Optional["_ModuleTreeNode"] = None
+    parent: "_ModuleTreeNode | None" = None
     children: list["_ModuleTreeNode"] = field(default_factory=list)
     trace: str = ""
 

vllm/transformers_utils/s3_utils.py

@@ -2,7 +2,7 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
 import fnmatch
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from vllm.utils.import_utils import PlaceholderModule
 
@@ -32,7 +32,7 @@ def _filter_ignore(paths: list[str], patterns: list[str]) -> list[str]:
 
 
 def glob(
-    s3: Optional["BaseClient"] = None,
+    s3: "BaseClient | None" = None,
     path: str = "",
     allow_pattern: list[str] | None = None,
 ) -> list[str]:

vllm/v1/attention/backends/mla/flashmla_sparse.py

 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, ClassVar, Optional
+from typing import TYPE_CHECKING, ClassVar
 
 import numpy as np
 import torch
@@ -707,7 +707,7 @@ class FlashMLASparseImpl(MLACommonBaseImpl[FlashMLASparseMetadata]):
         kv_sharing_target_layer_name: str | None,
         # MLA Specific Arguments
         topk_indice_buffer: torch.Tensor | None = None,
-        indexer: Optional["Indexer"] = None,
+        indexer: "Indexer | None" = None,
         **mla_args,
     ) -> None:
         super().__init__(

vllm/v1/attention/backends/mla/rocm_aiter_mla_sparse.py

@@ -2,7 +2,7 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, ClassVar, Optional
+from typing import TYPE_CHECKING, ClassVar
 
 import numpy as np
 import torch
@@ -284,7 +284,7 @@ class ROCMAiterMLASparseImpl(MLACommonBaseImpl[ROCMAiterMLASparseMetadata]):
         kv_sharing_target_layer_name: str | None,
         # MLA Specific Arguments
         topk_indice_buffer: torch.Tensor | None = None,
-        indexer: Optional["Indexer"] = None,
+        indexer: "Indexer | None" = None,
         **mla_args,
     ) -> None:
         super().__init__(

vllm/v1/attention/backends/tree_attn.py

@@ -4,7 +4,7 @@
 
 import ast
 from dataclasses import dataclass
-from typing import ClassVar, Optional
+from typing import ClassVar
 
 import torch
 
@@ -87,11 +87,11 @@ class TreeAttentionMetadata:
     tree_attn_bias: torch.Tensor | None = None
 
     # Cached Prefill/decode metadata.
-    _cached_prefill_metadata: Optional["TreeAttentionMetadata"] = None
-    _cached_decode_metadata: Optional["TreeAttentionMetadata"] = None
+    _cached_prefill_metadata: "TreeAttentionMetadata | None" = None
+    _cached_decode_metadata: "TreeAttentionMetadata | None" = None
 
     @property
-    def prefill_metadata(self) -> Optional["TreeAttentionMetadata"]:
+    def prefill_metadata(self) -> "TreeAttentionMetadata | None":
         if self.num_prefills == 0:
             return None
 
@@ -116,7 +116,7 @@ class TreeAttentionMetadata:
         return self._cached_prefill_metadata
 
     @property
-    def decode_metadata(self) -> Optional["TreeAttentionMetadata"]:
+    def decode_metadata(self) -> "TreeAttentionMetadata | None":
         if self.num_decode_tokens == 0:
             return None
 

vllm/v1/core/sched/interface.py

@@ -2,7 +2,7 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 from abc import ABC, abstractmethod
 from collections.abc import Iterable
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from vllm.multimodal import MULTIMODAL_REGISTRY, MultiModalRegistry
 
@@ -189,7 +189,7 @@ class SchedulerInterface(ABC):
         raise NotImplementedError
 
     @abstractmethod
-    def make_stats(self) -> Optional["SchedulerStats"]:
+    def make_stats(self) -> "SchedulerStats | None":
         """Make a SchedulerStats object for logging.
 
         The SchedulerStats object is created for every scheduling step.
@@ -201,5 +201,5 @@ class SchedulerInterface(ABC):
         """Shutdown the scheduler."""
         raise NotImplementedError
 
-    def get_kv_connector(self) -> Optional["KVConnectorBase_V1"]:
+    def get_kv_connector(self) -> "KVConnectorBase_V1 | None":
         return None

vllm/v1/engine/parallel_sampling.py

@@ -2,7 +2,7 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
 from copy import copy
-from typing import Optional, cast
+from typing import cast
 
 from vllm.outputs import CompletionOutput
 from vllm.sampling_params import RequestOutputKind, SamplingParams
@@ -133,7 +133,7 @@ class ParentRequest:
 
     @staticmethod
     def observe_finished_request(
-        parent_req: Optional["ParentRequest"],
+        parent_req: "ParentRequest | None",
         iteration_stats: IterationStats,
         num_generation_tokens: int,
     ):

vllm/v1/request.py

@@ -7,7 +7,7 @@ from collections import deque
 from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from functools import partial
-from typing import TYPE_CHECKING, Any, Optional
+from typing import TYPE_CHECKING, Any
 
 import torch
 
@@ -68,7 +68,7 @@ class Request:
         arrival_time: float | None = None,
         prompt_embeds: torch.Tensor | None = None,
         mm_features: list[MultiModalFeatureSpec] | None = None,
-        lora_request: Optional["LoRARequest"] = None,
+        lora_request: "LoRARequest | None" = None,
         cache_salt: str | None = None,
         priority: int = 0,
         trace_headers: Mapping[str, str] | None = None,

vllm/v1/sample/logits_processor/interface.py

@@ -4,7 +4,7 @@ from abc import ABC, abstractmethod
 from collections.abc import Sequence
 from dataclasses import dataclass
 from enum import Enum, auto
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 import torch
 
@@ -94,7 +94,7 @@ class LogitsProcessor(ABC):
     @abstractmethod
     def update_state(
         self,
-        batch_update: Optional["BatchUpdate"],
+        batch_update: "BatchUpdate | None",
     ) -> None:
         """Called when there are new output tokens, prior
         to each forward pass.

vllm/v1/utils.py

@@ -14,7 +14,6 @@ from typing import (
     TYPE_CHECKING,
     Any,
     Generic,
-    Optional,
     TypeVar,
     Union,
     overload,
@@ -229,7 +228,7 @@ def wait_for_completion_or_failure(
     api_server_manager: APIServerProcessManager,
     engine_manager: Union["CoreEngineProcManager", "CoreEngineActorManager"]
     | None = None,
-    coordinator: Optional["DPCoordinator"] = None,
+    coordinator: "DPCoordinator | None" = None,
 ) -> None:
     """Wait for all processes to complete or detect if any fail.
 

vllm/v1/worker/ubatching.py

 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 import threading
-from typing import Optional
 
 import torch
 
@@ -15,7 +14,7 @@ logger = init_logger(__name__)
 _THREAD_ID_TO_CONTEXT: dict = {}
 # Here we hardcode the number of microbatches to 2 for default.
 _NUM_UBATCHES: int = 2
-_CURRENT_CONTEXTS: list[Optional["UBatchContext"]] = []
+_CURRENT_CONTEXTS: list["UBatchContext | None"] = []
 
 
 class UBatchContext:

vllm/v1/worker/workspace.py

@@ -5,7 +5,6 @@ import inspect
 import os
 from itertools import accumulate
 from math import prod
-from typing import Optional
 
 import torch
 
@@ -26,7 +25,7 @@ _MB = 1024**2
 _GiB = 1024**3
 
 # Global workspace manager instance
-_manager: Optional["WorkspaceManager"] = None
+_manager: "WorkspaceManager | None" = None
 
 
 class WorkspaceManager: