vlllm官方代码更新频发,每个版本都有极大变动, 很难说哪个版本好用.
第一次阅读vllm源码是0.4.0版本,对这版圈复杂度极高的调度代码印象深刻
0.4.1对调度逻辑进行重构,完全大变样, 读代码速度快赶不上迭代的速度了。
现在已经更新到0.5.4, 经过长时间观察，发现主要的调度逻辑基本也稳定了下来, 应该可以作为一个固话的版本去阅读。

本文解读依据vllm 0.5.4版本. 没有修改任何代码,大家不必担心夹带私货！
打算以六篇文章的篇幅剖析vllm，希望能对大家有所帮助。

注解代码链接:
https://github.com/yblir/vllm-learn

参考文献：
https://zhuanlan.zhihu.com/p/691038809
https://zhuanlan.zhihu.com/p/681716326

一 大模型推理流程

在解析vllm源码前，我们先来回顾下llm推理流程。一个典型的推理过程如下:

在这里插入图片描述

① prefill：预填充阶段，把整段prompt喂给大模型做推理，获得kv-cache并保存。
②decode：大模型本质是个自回归模型，因此生成阶段，首先根据prompt中最后一个token的kv（input token 4）计算获得第一个推理结果（北），并保存对应的kv-cache(output token 1), 这个过程算一次推理；之后将 北 字作为输入（首次推理的输入是prompt，以后模型输入都是上次的生成token, 当然过程中要用到之前保存的kv-cache），做同样的推理生成 京 字，直到推理结束。

由于Decode阶段是逐一生成token，因此不能像prefill阶段那样能做大段prompt的并行计算，所以在LLM推理过程中，Decode阶段的耗时一般是更大的，单步生成token的耗时约占总推理时长的90%。

上述推理过程使用到了kv-cache技术，这里有些问题需要解决：
· 随着生成token的增多，kv-cache长度也变大，对gpu显存造成压力
· 生成的token长度无法预知，因此不能提前预知kv-cache所需的存储空间，给推理工作造成很大不确定性

vllm就是为解决上述问题而生，vllm的核心就是如何优化kv-cache，节省显存提高推理吞吐量。
调用方法也很简单，以下是qwen2 vllm推理代码：

# -*- coding: utf-8 -*-
# @Time    : 2024/8/18 20:14
# @Author  : yblir
# @File    : qwen2_vllm_inference.py
# explain  : 
# =======================================================
import os
import sys

sys.path.append('/mnt/e/PyCharm/insteresting/vllm-0.5.4/')

from vllm_module import LLM, SamplingParams
# from vllm import LLM, SamplingParams
from transformers import AutoTokenizer

os.environ['CUDA_VISIBLE_DEVICES'] = '0'
os.environ['CUDA_LAUNCH_BLOCKING'] = '1'
model_path = '/mnt/e/PyCharm/PreTrainModel/qwen2_15b_instruct'
# model_path = '/media/xk/D6B8A862B8A8433B/data/qwen2-15b-instruct'

params = {"repetition_penalty": 1.1,
          "temperature"       : 0.7,
          'n'                 : 4,
          "top_p"             : 0.8,
          "top_k"             : 20, }
sample_params = SamplingParams(**params)
llm = LLM(model=model_path,
          dtype='half'
            # dtype='float16'
          # 把模型层均分到n个gpu上, 而不是运行n个完整模型
          # tensor_parallel_size=1
          # gpu利用率最大70%
          # gpu_memory_utilization=0.7,
          )
tokenizer = AutoTokenizer.from_pretrained(model_path, )

# 构造模板
prompt = '介绍下京杭大运河'
messages = [
    {'role': 'system', 'content': '你是一个诗人'},
    {'role': 'user', 'content': prompt}
]

text = tokenizer.apply_chat_template(conversation=messages, tokenize=False, add_generation_prompt=True)

messages2 = [
    {'role': 'system', 'content': '你是一个诗人'},
    {'role': 'user', 'content': 'how far you go'}
]

text2 = tokenizer.apply_chat_template(conversation=messages2, tokenize=False, add_generation_prompt=True)

messages3 = [
    {'role': 'system', 'content': '你是一个诗人'},
    {'role': 'user', 'content': '中国首都城市什么名字'}
]

text3 = tokenizer.apply_chat_template(conversation=messages3, tokenize=False, add_generation_prompt=True)
# print(text)
outputs = llm.generate(
        # 当tokenizer.apply_chat_templat中 tokenize为 False 时激活prompts
        prompts=[text,text2,text3],

        # 当tokenizer.apply_chat_templat中 tokenize为 True 时激活prompt_token_ids,与prompts二选一
        # prompt_token_ids=[text,text2,text3],

        sampling_params=sample_params
)

for output in outputs:
    # prompt = output.prompt
    # print(prompt)
    # print(output)
    # print('------------------------------------------')
    for i,item in enumerate(range(4)):
        print(output.outputs[i].text)
        print(output.outputs[i].token_ids)
    print('------------------------------------------\n')

看起来很简单吧，似乎只要2步：只要把模型初始化，再调用generate方法就搞定了。实际上这两步的后面是耦合了调度与模型改造的复杂工程，本文将深度剖析潜藏在背后的源码。

二 vllm 原理分析

vllm管理kv-cache的技术称为PagedAttention，原理类似于虚拟内存分页管理技术。
正常推理流程中，生成的token长度无法预知，因此会最大化分配一块连续显存作为kv-cache的存储空间，可能到推理结束时这些空间大部分都用不到，而且这是为当前prompt分配的，其他prompt不能使用，造成极大浪费。
换个思路，如果把显存切分成多个连续小段是否可以呢！动态分配显存小段与生成的kv-cache 之间的存储关系，这样可以最大限度地使用显存，达到提升限度推理吞吐量的目的。
这种方法称为PagedAttention，主要有3个模块组成：logical kv blocks, block table, physical kv blocks. 原理如下图所示，下面我们来逐一解析。
在这里插入图片描述

  1. logical kv blocks：逻辑表， 不实际存储kv-cache,可以理解为C++语言中的指针，prefill和decode生成的kv-cache的"地址指针"存储在logical kv blocks， 逻辑表对"指针"的存储是连续的。不过在新版vllm中，logical kv这个东西已经删除了，当然逻辑块只是形式上消失了，实际上它依然隐藏在Sequence类的各个属性中，解释起来比较复杂，我们在以后的代码分析中再详解。
  2. physical kv blocks：可理解为实际存储token的物理显存，vllm中一个块默认为16（可以装16个token的k/v值），图中展示每个block大小为4。每个block内部是连续的，但block之间是不连续的，那么如何才能与logical保持对齐呢？这就需要block table了
  3. block table：存储logical与physical关系的映射表。如logical block0 -> physical block7. block table除了记录映射关系，还记录当前block槽位填充情况。如physical block7已经填满，因此filled==4； physical block1 槽位填充了3个，再填入一个father单词的token，filled会变为4。

多batch并行推理时，会有logical blocks映射到同一个physical blocks上，大家看图就能理解:
在这里插入图片描述

我们考虑另外一个问题，llm推理有时会有多个输出，这种情况PagedAttention该如何操作呢？
多输出有两种情况：
Parallel Sampling: 如果指定了n个输出，就把prompt复制n份，拼成一个batch喂给模型做推理。这时会产生prompt 的kv-cache重复存储，对这个重复的优化是另外问题，这里不展开了。
Beam Search：集束搜索，每个decode阶段，产生top k个token（k也被称为束宽），对应着当前时刻的top k个序列。它们的前置token也会有大量的kv-cache重复。
在这里插入图片描述

前面提到Parallel Sampling模式会把prompt复制 n份，如Figure 8所示，对应sample A1和sample A2，它们各自维护一套自己的logical blocks，由于内容完全相同，它们共享一套physical blocks，其中每个physical block对应引用计数ref count都为2. 进入推理阶段后，A1和A2各自独立做推理，如果生成了相同的token，会把新token kv-cache加入共享的physical block中。如果生成了不同的token（如图中的mothers和fathers），会触发copy-on-write机制，即在gpu上开辟一个新的block，如physical block1复制内容到block2，之后各自再装入生成的不同token, 同时，block1 计数-1，block2计数+1。这种操作符合vllm核心思想：节省KV cache显存，对于相同数据对应的KV cache，能复用则尽量复用；无法复用时，再考虑开辟新的物理空间。

vllm也有对Beam Search有优化，但这不是本文重点，暂时忽略。

目前为止，我们仅回顾与源码解析相关知识，PagedAttention还有许多东西没有讲到，说太多技术点反而会让人迷失在各种细节中，有兴趣可以自行去查资料了解。

三 vllm中一些基本概念

在解析源码前， 我们还需要清楚理解vllm一些概念的意思。

3.1 vllm 数据结构

如第一章中图片上展示的，一个prompt的典型例子如下：

<|im_start|>system
你是一个诗人<|im_end|>
<|im_start|>user
中国首都是<|im_end|>
<|im_start|>assistant

正常推理过程中，1个请求（batchsize）可能包含多个prompts，在vllm中，一个prompt才被看做一个请求；一个prompt可能输出多个outputs，这时每对prompt -> output序列称为一个seq序列，每条seq都维护着独立的status,可理解为当前时刻所处的推理状态，推理是否完结等:

  • WAITING：正在waiting队列中。waiting队列中的序列都没有做过prefill。
  • RUNNING：正在running队列中，即已经开始做推理。
  • SWAPPED：正在swapped队列中，表示此时gpu资源不足，相关的seq_group被抢占。

当然还有finished状态，主要记录因何种原因导致finished：

  • FINISHED_STOPPED：正常执行完毕，例如碰到符号，该seq的推理正常结束了
  • FINISHED_LENGTH_CAPPED：因为seq的长度达到最大长度限制，而结束推理
  • FINISHED_ABORTED：因不正常状态，而被终止的推理。例如客户端断开连接，则服务器会终止相关seq的推理
  • FINISHED_IGNORED：因prompt过长而被终止执行的推理。本质上也是受到长度限制

从上面可以看出单独管理seq有点复杂，所以我们需要统一管理一个prompt和它对应的所有outputs，称为一个seq_group。seq_group是vllm推理管理数据的基本单元。vllm 中设定一个seq_group中所有seq共享共一个prompt。这些变量的包含关系如下：
在这里插入图片描述
这张图是从某位大佬的文章中取来的，他用的版本应该是0.4.0，在0.5.4版本中，Sequence类已删除了logical_token_blocks属性和_append_tokens_to_blocks方法，对应的功能转移到其他代码中了。不过从图中，仍能清晰看到二者的从属关系。

3.2 调度原则

前面提到vllm的核心是对kv-cache的优化，而这种优化是通过调度系统Scheduler来完成的。
Scheduler维护着三个双端队列, 在3.1也有提到：
waiting，running，swapped。
每完成一次推理，都要对这三个队列进行动态调整，在下一次推理时实现最大限度提升吞吐量的目的.
这三个队列的作用如下：

  • waiting: 所有输入的模型的prompt都会被加入waiting队列中，这是输入数据的入口。这时的seq只有一条，就是prompt，连prefill都还没做，不管在外部设置了多输出还是Beam Search，此时只有这一条数据。waiting队列另一个数据来源是running队列。
  • running: 存储着上一次被送去做推理seq_groups，在下一次做推理前， 要对running队列中的seq_groups做检查，看系统是否有足够资源让它们留在队列中继续做下一次推理。如果当前系统资源不满足做一次推理，就把seq_group一条条pop()出来，转移到waiting或swapped队列中，直到满足下一次推理的资源需求。当然，running队列也会从waiting和swapped队列拿数据过来做推理，至于转移，怎么拿，接下来我们会详细分析。
  • swapped: 可以理解为失败者集散地，都是不满足条件（gpu blocks资源不足或某些推理参数超过阈值），被从running队列中踢出去的，等条件满足时，还会被从新加入running队列做推理。

上面提到了三个队列间的交互，它们的交互依据就是vllm的调度原则。
vllm调度的原则可以总结如下，了解了下面的处理逻辑，就能理解三个队列的用途及它们之间相互转移数据的规则：

  • 先来的请求先被服务（First-Come-First-Serve, FCFS）
  • 如有抢占的需要，后来的请求先被抢占（preemption）

FCFS大家都很好理解，我们来看下对preemption的处理：

抢占发生在推理阶段，vllm核心是最大限度优化吞吐量，推理过程中gpu显存不足或推理的tokens和seqs数量超过设定阈值，都会发生抢占，即暂时中断一些任务的执行，释放gpu上与它们相关的kv-cache,等资源充足，再恢复它们的执行。针对preemption有两种处理方式：

① 如果parallel sampling=1，直接释放所有physical blocks，将任务重新放回wait队列（放到队列头部，下一次最先取它），重新从prefill阶段开始做推理。
② 如果parallel sampling>1, 如果将它们直接丢掉，那未免过于浪费, 先把处理的好的blocks交换到CPU上，等gpu显存充足，再把这些blocks从CPU加载回来。

上面提到超过推理参数超过阈值也会导致抢占，这里的阈值指每次允许推理的最大seqs数量和最大tokens数量。在vllm 0.5.4版本中，由一个类budget的对象管理，每次推理前都要重新构建一个budget类对象，统计seqs和tokens数量是否越界。

至此，我们可以得出结论，判断一个seq_group是否被抢占的因素有三个：gpu blocks数量是否充足，当前调度能处理的seqs和tokens是否超过数量阈值。

3.3 vllm推理流程

有了上面的知识储备，我们就能理解一个seq从开始到结束，整个生命周期内的历程：
在这里插入图片描述

  • ① 每条prompt处理成Sequence对象，然后Sequence包装成seq_group，这条seq_group会存入waiting队列。此时只有一条seq，就是prompt，连预填充prefill都没做。status为waiting
  • ② 调度器选中这条seq_group做推理，图中我们展示两种情况，4输出和1输出，因此会产生4条seq和1条seq, 其中4 seq共享prompt，status为running。
  • ③ 推理一段时间后，gpu blocks资源不足或tokens或seqs数量超出阈值，发生抢占现象。多输出的seq_group相关kv blocks会被swap out到CPU上；而单输出的seq_group则会把相关的（prefill和decode）kv blocks释放，将seq_group重新放回waiting队列，就像什么都没发生过，幸运的是会被放到waiting队列最前面。
  • ④ 系统资源充足了，被swapped的seq_group会从CPU上swap_in gpu继续推理；单输出seq_group则从prefill开始重新奋斗。
  • ⑤⑥ 多输出必定会出现某条seq先推理结束，此时还活跃的seq数减1， 变为3个，当某条seq推理完结，会被标记为finished, 以后不再调用资源处理它，只有等seq_group中所有seq都推理结束，该seq_group才算推理完成。

注意：并不是每个seq_group都会经历抢占，如果系统资源充足，会跳过抢占的，此时的执行次序为：①②⑤⑥ 或 ①②⑤

四 推理代码解析

经过漫长的知识铺垫，我们终于能看到vllm的核心代码了
我们先略过初始化阶段，直接从generate开始，遇到重要知识点再现场分析

outputs = llm.generate(
        # 当tokenizer.apply_chat_templat中 tokenize为 False 时激活prompts
        prompts=[text,text2,text3],

        # 当tokenizer.apply_chat_templat中 tokenize为 True 时激活prompt_token_ids,与prompts二选一
        # prompt_token_ids=[text,text2,text3],

        sampling_params=sample_params
)

vllm推理代码真的很简洁，只要一行代码就行，背后逻辑就极其复杂了，而且每个版本都有很大改动。

  • vllm/entrypoints/llm.py
    def generate(
            self,
            prompts: Union[Union[PromptInputs, Sequence[PromptInputs]], Optional[Union[str, List[str]]]] = None,
            sampling_params: Optional[Union[SamplingParams, Sequence[SamplingParams]]] = None,
            prompt_token_ids: Optional[Union[List[int], List[List[int]]]] = None,
            use_tqdm: bool = True,
            lora_request: Optional[Union[List[LoRARequest], LoRARequest]] = None,
            prompt_adapter_request: Optional[PromptAdapterRequest] = None,
            guided_options_request: Optional[Union[LLMGuidedOptions, GuidedDecodingRequest]] = None
    ) -> List[RequestOutput]:

        if self.llm_engine.model_config.embedding_mode:
            raise ValueError(
                    "LLM.generate() is only supported for generation models (XForCausalLM)."
            )

        # cast 表面看是类型转换,但实践上没做任何事,直接原样返回, 那么这个调用的意义是什么 ?
        if prompt_token_ids is not None:
            inputs = self._convert_v1_inputs(
                    prompts=cast(Optional[Union[str, List[str]]], prompts),
                    prompt_token_ids=prompt_token_ids,
            )
        else:
            inputs = cast(Union[PromptInputs, Sequence[PromptInputs]], prompts)

        if isinstance(guided_options_request, dict):
            if len(guided_options_request) > 1:
                raise ValueError(
                        "You can only use one guided decoding but multiple is "
                        f"specified: {guided_options_request}")
            guided_options_request = GuidedDecodingRequest(**guided_options_request)

        if sampling_params is None:
            # Use default sampling params.
            sampling_params = SamplingParams()

        # 校验入参，并将一个batchsize中的每条prompt处理成Sequence对象，然后Sequence包装成SequenceGroup组，
        # 1. prompt->seq->seq_group, 2. 将seq_group加入合适gpu维护的scheduler的waiting队列,等待处理
        self._validate_and_add_requests(
                inputs=inputs,
                params=sampling_params,
                lora_request=lora_request,
                prompt_adapter_request=prompt_adapter_request,
                guided_options=guided_options_request)

        # 首先从scheduler的waiting队列取数据，加入到running队列，再从running队列
        # 中取数据推理，若物理blocks不够用，从running转入swap队列
        outputs = self._run_engine(use_tqdm=use_tqdm)

        return LLMEngine.validate_outputs(outputs, RequestOutput)

generate 代码中最重要的模块有两个，_validate_and_add_requests（数据预处理），_run_engine（实际推理）。

4.1 数据预处理

  • vllm/entrypoints/llm.py
    def _validate_and_add_requests(
            self,
            inputs: Union[PromptInputs, Sequence[PromptInputs]],
            params: Union[SamplingParams, Sequence[SamplingParams], PoolingParams, Sequence[PoolingParams]],
            lora_request: Optional[Union[Sequence[LoRARequest], LoRARequest]],
            prompt_adapter_request: Optional[PromptAdapterRequest],
            guided_options: Optional[GuidedDecodingRequest] = None,
    ) -> None:
        # 如果输入是一条prompt,而不是list,会在此处自动转换为list
        if isinstance(inputs, (str, dict)):
            # Convert a single prompt to a list.
            inputs = [inputs]
           
		...
		
        # Add requests to the engine.
        # 遍历每一条prompt，1个prompt算1个request，需要有1个全局唯一的request_id
        for i, request_inputs in enumerate(inputs):
            self._add_request(
                    request_inputs,
                    params[i] if isinstance(params, Sequence) else params,
                    lora_request=lora_request[i] if isinstance(lora_request, Sequence) else lora_request,
                    prompt_adapter_request=prompt_adapter_request)
                    
    # ==========================================================================================================
    def _add_request(
            self,
            inputs: PromptInputs,
            params: Union[SamplingParams, PoolingParams],
            lora_request: Optional[Union[List[LoRARequest], LoRARequest]] = None,
            prompt_adapter_request: Optional[PromptAdapterRequest] = None
    ) -> None:
        # 每个prompt赋1个全局唯一的request_id
        request_id = str(next(self.request_counter))
        self.llm_engine.add_request(
                request_id,
                inputs,
                params,
                lora_request=lora_request,
                prompt_adapter_request=prompt_adapter_request)
  • vllm/engine/llm_engine.py
  # ==========================================================================================================
    def add_request(
            self,
            request_id: str,  # 每个请求的唯一id,在vLLM内部，1条prompt算1个请求，会附给1个请求id
            inputs: PromptInputs,  # prompt
            params: Union[SamplingParams, PoolingParams],  # 用于采样的参数（温度、topk等）
            arrival_time: Optional[float] = None,  # 请求到达的时间。如果是None，则用当前系统时间
            lora_request: Optional[LoRARequest] = None,  # 如果是用lora模型做推理，相关的lora请求
            trace_headers: Optional[Mapping[str, str]] = None,
            prompt_adapter_request: Optional[PromptAdapterRequest] = None,

    ) -> None:
        if lora_request is not None and not self.lora_config:
            raise ValueError(f"Got lora_request {lora_request} but LoRA is not enabled!")
        # 设置该请求的到达时间
        if arrival_time is None:
            arrival_time = time.time()

        # processed_inputs:dict,= {'prompts':xxx,'prompts_token_ids':xxx,'multi_modal_data':None}
        processed_inputs = self.process_model_inputs(
                request_id=request_id,
                inputs=inputs,
                lora_request=lora_request,
                prompt_adapter_request=prompt_adapter_request)
        # 1. prompt->seq->seq_group, 2. 将seq_group加入合适gpu维护的scheduler的waiting队列,等待处理
        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,
        )

我们看到，经过疯狂套娃后，最终实际干活的是self.process_model_inputs和self._add_processed_request这两个方法，其他全都是中间商~

  • vllm/engine/llm_engine.py
    def process_model_inputs(
            self,
            request_id: str,
            inputs: PromptInputs,
            lora_request: Optional[LoRARequest] = None,
            prompt_adapter_request: Optional[PromptAdapterRequest] = None,
    ) -> LLMInputs:
        if isinstance(inputs, str):
            inputs = {"prompt": inputs}

        if "prompt_token_ids" not in inputs:
            # 这个函数就是为了拿到self.tokenizer
            tokenizer = self.get_tokenizer_group("prompts must be None if skip_tokenizer_init is True")
            # 文字prompt编码成token_id
            prompt_token_ids = tokenizer.encode(request_id=request_id,
                                                prompt=inputs["prompt"],
                                                lora_request=lora_request)
        # 如果入参前已经做好token_ids,直接取出来用
        else:
            prompt_token_ids = inputs["prompt_token_ids"]

        # 使用未合并的lora才会走进入这个判断分支
        if prompt_adapter_request:
            prompt_token_ids = [0] * prompt_adapter_request.prompt_adapter_num_virtual_tokens + prompt_token_ids

        # LLMInputs继承自TypedDict,将入参转换为字典
        # llm_inputs = {'prompts':xxx,'prompts_token_ids':xxx,'multi_modal_data':None}
        llm_inputs = LLMInputs(prompt_token_ids=prompt_token_ids,
                               prompt=inputs.get("prompt"),
                               multi_modal_data=inputs.get("multi_modal_data"))

        # todo 使用functools.partial高阶用法,返回的是一个固定llm_inputs参数的函数，真的好用吗？
        # 目前觉得这个函数会拖慢速度,因为每个prompt都要经过这里获得模型架构的操作
        # 目前这个函数,经过多层调用后, 最后原样返回,没对llm_inputs做任何操作
        return self.input_processor(llm_inputs)

process_model_inputs功能是把输入的prompt转换为token_id, 结果以字典形式输出：
llm_inputs = {‘prompts’:xxx,‘prompts_token_ids’:xxx,‘multi_modal_data’:None}

   # ==========================================================================================================
    
    def _add_processed_request(
            self,
            request_id: str,
            processed_inputs: LLMInputs,
            params: Union[SamplingParams, PoolingParams],
            arrival_time: float,
            lora_request: Optional[LoRARequest],
            prompt_adapter_request: Optional[PromptAdapterRequest],
            trace_headers: Optional[Mapping[str, str]] = None,
    ) -> None:
        # Create the sequences.
        block_size = self.cache_config.block_size
        # self.seq_counter是在类中初始化，所以可以为每条seq生成不重复的id，
        # seq_id与request_id是两个独立的变量
        seq_id = next(self.seq_counter)
        eos_token_id = self._get_eos_token_id(lora_request)
        # seq 包含当前prompt的各种信息:token_id,status(waiting,...), 占用blocks数量(逻辑,物理数量相同)
        seq = Sequence(seq_id, processed_inputs, block_size, eos_token_id,
                       lora_request, prompt_adapter_request)
        # 将seq和采样参数合并为seq_group
        # --------------------------------------------------------------------------------------------------------------
        if isinstance(params, SamplingParams):
            seq_group = self._create_sequence_group_with_sampling(
                    request_id,
                    seq,
                    params,
                    arrival_time=arrival_time,
                    lora_request=lora_request,
                    trace_headers=trace_headers,
                    prompt_adapter_request=prompt_adapter_request)
        elif isinstance(params, PoolingParams):
            seq_group = self._create_sequence_group_with_pooling(
                    request_id,
                    seq,
                    params,
                    arrival_time=arrival_time,
                    lora_request=lora_request,
                    prompt_adapter_request=prompt_adapter_request)
        else:
            raise ValueError("Either SamplingParams or PoolingParams must be provided.")
        # --------------------------------------------------------------------------------------------------------------

        # Add the sequence group to the scheduler with least unfinished seqs.
        # 获得当前每个gpu上还没推理结束的seq_group数量: len(self.waiting) + len(self.running) + len(self.swapped)
        costs = [scheduler.get_num_unfinished_seq_groups() for scheduler in self.scheduler]
        # 找出工作量最少的调度器
        min_cost_scheduler = self.scheduler[costs.index(min(costs))]
        # 将当前seq_group加入这个调度器中(根据self.scheduler初始过程可知,每个gpu维护一个调度器,
        # 这条代码的意思就是当前seq_group由工作量最少的gpu负责推理)
        min_cost_scheduler.add_seq_group(seq_group)

用户输入的prompt经过_validate_and_add_requests处理后，会封装为seq_group，然后将seq_group加入合适gpu维护的scheduler的waiting队列, 等待处理。
seq和seq_group是vllm推理的基本单元，是两个非常重要的概念，我们来直观感受下它们的数据格式：

seq:
在这里插入图片描述

seq_group:
初始阶段seqs_dict都只有一个元素，经过prefill后，会扩展为n个元素（n是输出outputs数量）
在这里插入图片描述

4.2 推理流程

    def _run_engine(
            self, *, use_tqdm: bool
    ) -> List[Union[RequestOutput, EmbeddingRequestOutput]]:
        # Initialize tqdm.
        if use_tqdm:
            num_requests = self.llm_engine.get_num_unfinished_requests()
            pbar = tqdm(
                    total=num_requests,
                    desc="Processed prompts",
                    dynamic_ncols=True,
                    postfix=f"est. speed input: {0:.2f} toks/s, output: {0:.2f} toks/s",
            )
        # Run the engine.
        outputs: List[Union[RequestOutput, EmbeddingRequestOutput]] = []
        total_in_toks = 0
        total_out_toks = 0

        # 如果当前调度器中还有没完成推理的请求（调度器中waiting/running/swapped任一队列非空）
        while self.llm_engine.has_unfinished_requests():
            # 执行1次推理调度（step），决定哪些请求的数据可以参与到这次推理中，step输出本次推理结果
            step_outputs = self.llm_engine.step()
            # 一次step推理后，如果有请求已经完成了推理，将推理结果装进outputs中，
            for output in step_outputs:
                if output.finished:
                    outputs.append(output)
                    if use_tqdm:
                        if isinstance(output, RequestOutput):
                            # Calculate tokens only for RequestOutput
                            total_in_toks += len(output.prompt_token_ids)
                            in_spd = total_in_toks / pbar.format_dict["elapsed"]
                            total_out_toks += sum(len(stp.token_ids) for stp in output.outputs)
                            out_spd = total_out_toks / pbar.format_dict["elapsed"]
                            pbar.postfix = (
                                f"est. speed input: {in_spd:.2f} toks/s, output: {out_spd:.2f} toks/s"
                            )
                        pbar.update(1)
        if use_tqdm:
            pbar.close()
        # Sort the outputs by request ID.
        # This is necessary because some requests may be finished earlier than
        # its previous requests.
        return sorted(outputs, key=lambda x: int(x.request_id))

整个推理engine中，最重要的是self.llm_engine.step()，封装了所有的调度，推理和后处理代码。

    def step(self) -> List[Union[RequestOutput, EmbeddingRequestOutput]]:
        # 多GPU并行推理时走AsyncLLMEngine分支。如果进入当前LLMEngine,性能会下降，这里会抛出异常。
        if self.parallel_config.pipeline_parallel_size > 1:
            raise NotImplementedError(
                    "Pipeline parallelism is only supported through AsyncLLMEngine "
                    "as performance will be severely degraded otherwise.")

        # 上述if判断表明，只有一个GPU可用。因此self.scheduler也只有一个元素，是当前GPU的调度
        # 该函数调用改变调度的内部状态(self.running、self.swapped 和 self.waiting)
        seq_group_metadata_list, scheduler_outputs = self.scheduler[0].schedule()
        
		...
		
        return request_outputs

step中使用的调度代码如下：

    def schedule(self) -> Tuple[List[SequenceGroupMetadata], SchedulerOutputs]:
        # Schedule sequence groups.
        # This function call changes the internal states of the scheduler
        # such as self.running, self.swapped, and self.waiting.
        # 该函数调用改变调度的内部状态(self.running、self.swapped 和 self.waiting）
        scheduler_outputs = self._schedule()
        ...

        return seq_group_metadata_list, scheduler_outputs

调度系统是vllm代码的核心，接下来，我们花单独一篇文章，详细解读self._schedule()的内部逻辑。

# 自然语言处理 # 语言模型 # 深度学习 # 人工智能 # gpt-3
Logo
开放原子开发者工作坊

开放原子开发者工作坊旨在鼓励更多人参与开源活动，与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动，如meetup、训练营等，主打技术交流，干货满满，真诚地邀请各位开发者共同参与！

更多推荐

第二届开放原子大赛首批创新成果集结武汉，诚邀广大开发者共鉴开源技术盛宴

第二届开放原子大赛首批创新成果集结武汉，诚邀广大开发者共鉴开源技术盛宴

avatar 开放原子开发者工作坊

诚邀报名 | 开源基础设施能力建设分论坛：打造开源生态的“心脏”

诚邀报名 | 开源基础设施能力建设分论坛：打造开源生态的“心脏”

avatar 开放原子开发者工作坊

诚邀报名 | 编程语言分论坛：AI时代的技术革新与开源实践

诚邀报名 | 编程语言分论坛：AI时代的技术革新与开源实践

avatar 开放原子开发者工作坊