def __init__(self) -> None:
    ........
    engine_args = EngineArgs(
        # 模型定义名称
        model=model,
        # HF Transformers分词器的名称
        tokenizer=tokenizer,
        # 分词器模式。"auto"将使用快速分词器, "slow"将总是使用慢速分词器
        tokenizer_mode=tokenizer_mode,
        skip_tokenizer_init=skip_tokenizer_init,
        # 当下载模型和分词器时，是否信任HF远程代码
        trust_remote_code=trust_remote_code,
        # 用于分布式执行的GPU数量，使用张量并行性。默认为1
        tensor_parallel_size=tensor_parallel_size,
        # 模型权重和激活的数据类型。目前，我们支持float32、float16和bfloat16。
        # 如果是auto，我们使用在模型配置文件中指定的torch_dtype属性。
        # 但是，如果配置中的torch_dtype是float32，我们将使用float16。默认为"auto"。
        dtype=dtype,
        # 用于量化模型权重的方法。目前，我们支持“awq”、“gptq”和“fp8”（实验性）。
        # 如果为 None，我们首先检查模型配置文件中的 `quantization_config` 属性。
        # 如果也为 None，我们假设模型权重未量化，并使用 `dtype` 来确定权重的数据类型
        quantization=quantization,
        # 默认0，表示不随机
        seed=seed,
        # GPU显存占用水位
        gpu_memory_utilization=gpu_memory_utilization,
        # cpu区域kvcache交换空间，默认4GB
        swap_space=swap_space,
        # 在cpu去开辟模型权重的存储空间，变相增加GPU显存
        cpu_offload_gb=cpu_offload_gb,
        # 默认打开eager(Extrapolation Algorithm for Greater Language-model Efficiency)算法加速，代替CUDA加速
        enforce_eager=enforce_eager,
        # CUDA Graph 相关参数，控制Graph优化间隔
        max_context_len_to_capture=max_context_len_to_capture,
        max_seq_len_to_capture=max_seq_len_to_capture,
        # 使用NCLL代替all reduce kernal
        disable_custom_all_reduce=disable_custom_all_reduce,
        disable_async_output_proc=disable_async_output_proc,
        **kwargs,
    )
    self.llm_engine = LLMEngine.from_engine_args(
        engine_args, usage_context=UsageContext.LLM_CLASS)
    self.request_counter = Counter()

def __init__(self, ...) -> None:
    # 保持各个核心子模块的配置文件
    self.model_config = model_config
    self.cache_config = cache_config
    self.lora_config = lora_config
    self.parallel_config = parallel_config
    self.scheduler_config = scheduler_config
    self.device_config = device_config
    self.speculative_config = speculative_config
    self.load_config = load_config
    self.decoding_config = decoding_config or DecodingConfig()
    self.prompt_adapter_config = prompt_adapter_config
    self.observability_config = observability_config or ObservabilityConfig()
    self.log_stats = log_stats

    # 初始化token编码器tokenizer，解码器detokenizer
    if not self.model_config.skip_tokenizer_init:
        self.tokenizer = self._init_tokenizer()
        self.detokenizer = Detokenizer(self.tokenizer)
        tokenizer_group = self.get_tokenizer_group()
    else:
        self.tokenizer = None
        self.detokenizer = None
        tokenizer_group = None

    # 通过模型配置中的start_token_id和是否为encode_decode_model等信息，进行不同的预处理
    self.input_preprocessor = InputPreprocessor(model_config, self.tokenizer)
    self.input_registry = input_registry
    # 会绑定模型配置参数到input_processor方法中，由于模型加载的过程中已经做了注册。
    # 因此input_processor就自然关联上了模型文件中的预处理方法。
    self.input_processor = input_registry.create_input_processor(model_config)

    # 初始化模型执行器，主要是对不同场景下的worker进行封装
    self.model_executor = executor_class(
        model_config=model_config,
        cache_config=cache_config,
        parallel_config=parallel_config,
        scheduler_config=scheduler_config,
        device_config=device_config,
        lora_config=lora_config,
        speculative_config=speculative_config,
        load_config=load_config,
        prompt_adapter_config=prompt_adapter_config,
        observability_config=self.observability_config,
    )

    # 如果是嵌入式模型，需要先初始话kv_cache, 嵌入模型代表一些图片风格生成类的模型
    if not self.model_config.embedding_mode:
        # 根据GPU显存大小，确定初始化block size个数。
        self._initialize_kv_caches()

    # 初始化scheduler相关模块，这里会根据pipeline_parallel_size定义的并行数量，创建多个scheduler
    self.cached_scheduler_outputs = [
        SchedulerOutputState()
        for _ in range(self.parallel_config.pipeline_parallel_size)
    ]

    self.scheduler_contexts = [
        SchedulerContext()
        for _ in range(self.parallel_config.pipeline_parallel_size)
    ]

    # for AsyncEngine
    if model_config.use_async_output_proc:
        process_model_outputs = weak_bind(self._process_model_outputs)

        self.async_callbacks = [
            partial(process_model_outputs,
                    ctx=self.scheduler_contexts[v_id])
            for v_id in range(self.parallel_config.pipeline_parallel_size)
        ]
    else:
        self.async_callbacks = []

    # Token block的核心模块
    self.scheduler = [
        Scheduler(
            scheduler_config, cache_config, lora_config,
            parallel_config.pipeline_parallel_size,
            self.async_callbacks[v_id]
            if model_config.use_async_output_proc else None)
        for v_id in range(parallel_config.pipeline_parallel_size)
    ]

    # 创建output后处理，判断是否这个sequence是finish状态
    # 为 bean search 和 speculative decoding，在seq_group中生成更多的seq
    self.output_processor = (
        SequenceGroupOutputProcessor.create_output_processor(
            self.scheduler_config,
            self.detokenizer,
            self.scheduler,
            self.seq_counter,
            get_tokenizer_for_seq,
            stop_checker=StopChecker(
                self.scheduler_config.max_model_len,
                get_tokenizer_for_seq,
            ),
        ))

 def add_request(
        self,
        request_id: str,
        prompt: PromptType,
        params: Union[SamplingParams, PoolingParams],
        arrival_time: Optional[float] = None,
        lora_request: Optional[LoRARequest] = None,
        trace_headers: Optional[Mapping[str, str]] = None,
        prompt_adapter_request: Optional[PromptAdapterRequest] = None,
    ) -> None:
        if arrival_time is None:
            arrival_time = time.time()

        # promt -> tokenlize
        preprocessed_inputs = self.input_preprocessor.preprocess(
            prompt,
            request_id=request_id,
            lora_request=lora_request,
            prompt_adapter_request=prompt_adapter_request,
        )
        processed_inputs = self.input_processor(preprocessed_inputs)

        # tokenlize -> sequence_group
        # 将sequence_group传入scheuler的add_seq_group方法进行KVCache的调度计算
        self._add_processed_request(
            request_id=request_id,
            processed_inputs=processed_inputs,
            params=params,
            arrival_time=arrival_time,
            lora_request=lora_request,
            prompt_adapter_request=prompt_adapter_request,
            trace_headers=trace_headers,
        )

    def step(self) -> List[Union[RequestOutput, EmbeddingRequestOutput]]:
        virtual_engine = 0

        # step函数是同步调用，只有一个scheduler，我们只使用给一个即可
        cached_outputs = self.cached_scheduler_outputs[virtual_engine]
        seq_group_metadata_list = cached_outputs.seq_group_metadata_list
        scheduler_outputs = cached_outputs.scheduler_outputs
        allow_async_output_proc = cached_outputs.allow_async_output_proc
        ctx = self.scheduler_contexts[virtual_engine]
        ctx.request_outputs.clear()

        # 这次batch size，已经如果没有未完成的工作了
        if not self._has_remaining_steps(seq_group_metadata_list):
            # 开始新一次的调度迭代
            (seq_group_metadata_list, scheduler_outputs,
             allow_async_output_proc
             ) = self.scheduler[virtual_engine].schedule()

            ctx.seq_group_metadata_list = seq_group_metadata_list
            ctx.scheduler_outputs = scheduler_outputs

            # 同步调用，使用同步方法做模型的后处理
            # 根据request_id，将生成的token，展开到sequence group中
            if not allow_async_output_proc and len(ctx.output_queue) > 0:
                self._process_model_outputs(ctx=ctx)

            if (self.scheduler_config.is_multi_step
                    and scheduler_outputs.num_lookahead_slots > 0):
                # 缓存上次的输出，用于下一次的输入
                self._cache_scheduler_outputs_for_multi_step(
                    virtual_engine, seq_group_metadata_list, scheduler_outputs,
                    allow_async_output_proc)

        assert seq_group_metadata_list is not None
        assert scheduler_outputs is not None

        if not scheduler_outputs.is_empty():
            finished_requests_ids = self.scheduler[
                virtual_engine].get_and_reset_finished_requests_ids()

            # 在每个引擎内部获取sample_token_id缓存，为了并发场景下，可以向worker传递此token
            last_sampled_token_ids = \
                self._get_last_sampled_token_ids(virtual_engine)

            execute_model_req = ExecuteModelRequest(
                seq_group_metadata_list=seq_group_metadata_list,
                blocks_to_swap_in=scheduler_outputs.blocks_to_swap_in,
                blocks_to_swap_out=scheduler_outputs.blocks_to_swap_out,
                blocks_to_copy=scheduler_outputs.blocks_to_copy,
                num_lookahead_slots=scheduler_outputs.num_lookahead_slots,
                running_queue_size=scheduler_outputs.running_queue_size,
                finished_requests_ids=finished_requests_ids,
                last_sampled_token_ids=last_sampled_token_ids)

            if allow_async_output_proc:
                execute_model_req.async_callback = self.async_callbacks[
                    virtual_engine]

            # 同步发送给执行器
            outputs = self.model_executor.execute_model(
                execute_model_req=execute_model_req)

            # 向缓存中存储last_sampled_token_ids，用于下次迭代.
            if self.scheduler_config.is_multi_step:
                self._update_cached_scheduler_output(virtual_engine, outputs)
        else:
            # 都完成了，那么无需模型执行器参与，直接就处理输出结果就好了。
            if len(ctx.output_queue) > 0:
                self._process_model_outputs(ctx=ctx)
            # No outputs in this case
            outputs = []

        # 将当前这次step数+1.
        if self.scheduler_config.is_multi_step:
            for seq_group in seq_group_metadata_list:
                seq_group.finish_step()

        # 没有剩余的step，我们讲缓存存入第一个缓存中(只针对step同步调用)
        if not self._has_remaining_steps(seq_group_metadata_list):
            if self.scheduler_config.is_multi_step:
                self.cached_scheduler_outputs[0] = SchedulerOutputState()

            # Add results to the output_queue
            ctx.append_output(outputs=outputs,
                              seq_group_metadata_list=seq_group_metadata_list,
                              scheduler_outputs=scheduler_outputs,
                              is_async=allow_async_output_proc,
                              is_last_step=True)

            if outputs and allow_async_output_proc:
                assert len(outputs) == 1, (
                    "Async postprocessor expects only a single output set")

                self._advance_to_next_step(
                    outputs[0], seq_group_metadata_list,
                    scheduler_outputs.scheduled_seq_groups)
        else:
            return ctx.request_outputs
        
        # 当前sequence group中没有新的request，我们将最后的output处理完毕后，停止模型的运行
        if not self.has_unfinished_requests():
            # Drain async postprocessor (if exists)
            if len(ctx.output_queue) > 0:
                self._process_model_outputs(ctx=ctx)
            assert len(ctx.output_queue) == 0
            logger.debug("Stopping remote worker execution loop.")
            self.model_executor.stop_remote_worker_execution_loop()

        return ctx.request_outputs