class_docstr: Base abstract class for managing blocked KV caches. Will probably have a single
  implementation for now.
docstr: Get the existing sequence descriptor for a given uid or initialize one if
  it does not exist. NOTE: This will always return a valid sequence descriptor
  if one may be allocated and should not be used from APIs that are attempting
  to test the schedulability of a hypothetical batch.
docstr_tok: ["Get","the","existing","sequence","descriptor","for","a","given","uid","or","initialize","one","if","it","does","not","exist",".","NOTE",":","This","will","always","return","a","valid","sequence","descriptor","if","one","may","be","allocated","and","should","not","be","used","from","APIs","that","are","attempting","to","test","the","schedulability","of","a","hypothetical","batch","."]
code: def get_or_create_sequence(self, uid: int) -> DSSequenceDescriptor:
      """
      Get the existing sequence descriptor for a given uid or initialize one if
      it does not exist. NOTE: This will always return a valid sequence descriptor
      if one may be allocated and should not be used from APIs that are attempting
      to test the schedulability of a hypothetical batch.
      """
      seq = self.get_sequence(uid)
      if seq is not None:
          return seq
      else:
          return self._create_sequence(uid)
  
imports: import torch
  from typing import Any, Dict, Optional, Tuple
  from deepspeed.accelerator import get_accelerator
  from deepspeed.ops.op_builder import RaggedUtilsBuilder
  from deepspeed.utils.logging import logger
  from .blocked_allocator import BlockedAllocator
  from .kv_cache import BlockedKVCache
  from .manager_configs import DSStateManagerConfig, KVCacheConfig
  from .sequence_descriptor import DSSequenceDescriptor
label_desc: Use image node_id 5 for calling the DSStateManager obj's underlying member method code with example usage: obj.get_or_create_sequence(uid) and returns: seq, self