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 syssys.path.append('/mnt/e/PyCharm/insteresting/vllm-0.5.4/')from vllm_module import LLM, SamplingParams
# from vllm import LLM, SamplingParams
from transformers import AutoTokenizeros.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 时激活promptsprompts=[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. 原理如下图所示,下面我们来逐一解析。
- logical kv blocks:逻辑表, 不实际存储kv-cache,可以理解为C++语言中的指针,prefill和decode生成的kv-cache的"地址指针"存储在logical kv blocks, 逻辑表对"指针"的存储是连续的。不过在新版vllm中,logical kv这个东西已经删除了,当然逻辑块只是形式上消失了,实际上它依然隐藏在Sequence类的各个属性中,解释起来比较复杂,我们在以后的代码分析中再详解。
- physical kv blocks:可理解为实际存储token的物理显存,vllm中一个块默认为16(可以装16个token的k/v值),图中展示每个block大小为4。每个block内部是连续的,但block之间是不连续的,那么如何才能与logical保持对齐呢?这就需要block table了
- 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 时激活promptsprompts=[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,会在此处自动转换为listif 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_idfor 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_idrequest_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个请求idinputs: PromptInputs, # promptparams: 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.tokenizertokenizer = self.get_tokenizer_group("prompts must be None if skip_tokenizer_init is True")# 文字prompt编码成token_idprompt_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 = 0total_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 RequestOutputtotal_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()的内部逻辑。