1.前言

        将微调后的模型部署到线上后，需要用到模型的推理加速，记录使用vLLM的推理加速流程。

        过程参考：GitHub - owenliang/qwen-vllm: 通义千问VLLM推理部署DEMO

2.加速流程

        在准备线上服务时，准备了两套方案。方案一：将大模型封装为一个单独的服务，提供访问接口，业务端调用大模型的接口；方案二：将大模型的加载包装成一个类，通过实例化类函方法，完成大模型的加载。

2.1 方案一

        将大模型作为一个服务端，单独封装为一个服务。在业务端通过http方法调用大模型。

2.2.1 服务端

       vllm_server.py内容如下：

import os 
from vllm import AsyncEngineArgs,AsyncLLMEngine
from vllm.sampling_params import SamplingParams
#from modelscope import AutoTokenizer, GenerationConfig,snapshot_download
from transformers import AutoTokenizer, AutoModel, AutoModelForCausalLM
from transformers.generation import GenerationConfig
import ray
ray.init(local_mode=True)
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, Response, StreamingResponse
import uvicorn
from prompt_utils import _build_prompt,remove_stop_words
import uuid
import json
#from modelscope import AutoTokenizer, GenerationConfig,snapshot_download
os.environ['CUDA_VISIBLE_DEVICES'] = '3'
# http接口服务
app=FastAPI()
if not ray.is_initialized():
    ray.init(local_model=True)
# vLLM参数
#model_dir="/data/shared/Qwen/Qwen-Chat/"
model_dir="/data/shared/Qwen/INT_4"
#model_dir = '/data/shared/Qwen/Qwen-14B-Chat-Int4/'
tensor_parallel_size=1
gpu_memory_utilization=0.5
quantization='gptq'
dtype='float16'

# vLLM模型加载
def load_vllm():
    global generation_config,tokenizer,stop_words_ids,engine    
    # 模型下载
    #snapshot_download(model_dir)
    # 模型基础配置
    generation_config=GenerationConfig.from_pretrained(model_dir,trust_remote_code=True)
    # 加载分词器
    tokenizer=AutoTokenizer.from_pretrained(model_dir,trust_remote_code=True)
    tokenizer.eos_token_id=generation_config.eos_token_id
    # 推理终止词
    stop_words_ids=[tokenizer.im_start_id,tokenizer.im_end_id,tokenizer.eos_token_id]
    # vLLM基础配置
    args=AsyncEngineArgs(model_dir)
    args.worker_use_ray=False
    args.engine_use_ray=False
    args.tokenizer=model_dir
    args.tensor_parallel_size=tensor_parallel_size
    args.trust_remote_code=True
    #args.quantization=quantization
    args.gpu_memory_utilization=gpu_memory_utilization
    args.dtype=dtype
    args.max_num_seqs=30    # batch最大20条样本
    # 加载模型
    os.environ['VLLM_USE_MODELSCOPE']='False'
    engine=AsyncLLMEngine.from_engine_args(args)
    return generation_config,tokenizer,stop_words_ids,engine

generation_config,tokenizer,stop_words_ids,engine=load_vllm()

# 用户停止句匹配
def match_user_stop_words(response_token_ids,user_stop_tokens):
    for stop_tokens in user_stop_tokens:
        if len(response_token_ids)<len(stop_tokens):
            continue 
        if response_token_ids[-len(stop_tokens):]==stop_tokens:
            return True  # 命中停止句, 返回True
    return False

# chat对话接口
@app.post("/chat")
async def chat(request: Request):
    request=await request.json()
    
    query=request.get('query',None)
    history=request.get('history',[])
    system=request.get('system','You are a helpful assistant.')
    stream=request.get("stream",False)
    user_stop_words=request.get("user_stop_words",[])    # list[str]，用户自定义停止句，例如：['Observation: ', 'Action: ']定义了2个停止句，遇到任何一个都会停止
    
    if query is None:
        return Response(status_code=502,content='query is empty')

    # 用户停止词
    user_stop_tokens=[]
    for words in user_stop_words:
        user_stop_tokens.append(tokenizer.encode(words))
    
    # 构造prompt
    prompt_text,prompt_tokens=_build_prompt(generation_config,tokenizer,query,history=history,system=system)
        
    # vLLM请求配置
    sampling_params=SamplingParams(stop_token_ids=stop_words_ids, 
                                    early_stopping=False,
                                    top_p=generation_config.top_p,
                                    top_k=-1 if generation_config.top_k == 0 else generation_config.top_k,
                                    temperature=generation_config.temperature,
                                    repetition_penalty=generation_config.repetition_penalty,
                                    max_tokens=generation_config.max_new_tokens)
    # vLLM异步推理（在独立线程中阻塞执行推理，主线程异步等待完成通知）
    request_id=str(uuid.uuid4().hex)
    results_iter=engine.generate(prompt=None,sampling_params=sampling_params,prompt_token_ids=prompt_tokens,request_id=request_id)
    
    # 流式返回，即迭代transformer的每一步推理结果并反复返回
    if stream:
        async def streaming_resp():
            async for result in results_iter:
                # 移除im_end,eos等系统停止词
                token_ids=remove_stop_words(result.outputs[0].token_ids,stop_words_ids)
                # 返回截止目前的tokens输出                
                text=tokenizer.decode(token_ids)
                yield (json.dumps({'text':text})+'\0').encode('utf-8')
                # 匹配用户停止词,终止推理
                if match_user_stop_words(token_ids,user_stop_tokens):
                    await engine.abort(request_id)   # 终止vllm后续推理
                    break
        return StreamingResponse(streaming_resp())

    # 整体一次性返回模式
    async for result in results_iter:
        # 移除im_end,eos等系统停止词
        token_ids=remove_stop_words(result.outputs[0].token_ids,stop_words_ids)
        # 返回截止目前的tokens输出                
        text=tokenizer.decode(token_ids)
        # 匹配用户停止词,终止推理
        if match_user_stop_words(token_ids,user_stop_tokens):
            await engine.abort(request_id)   # 终止vllm后续推理
            break

    ret={"text":text}
    print(ret)
    return JSONResponse(content=ret)

if __name__=='__main__':
    uvicorn.run(app,
                host='0.0.0.0',
                port=9522,
                log_level="debug")

prompt_utils.py 内容如下：

import copy 

# 按chatml格式构造千问的Prompt
def _build_prompt(
                generation_config,
                tokenizer,
                query,
                history=None,
                system=""):
    if history is None:
        history=[]

    # 包裹发言内容的token
    im_start,im_start_tokens='<|im_start|>',[tokenizer.im_start_id]
    im_end,im_end_tokens='<|im_end|>',[tokenizer.im_end_id]
    # 换行符token
    nl_tokens=tokenizer.encode("\n")

    # 用于编码system/user/assistant的一段发言, 格式{role}\n{content}
    def _tokenize_str(role,content): # 返回元组，下标0是文本，下标1是token ids
        return f"{role}\n{content}",tokenizer.encode(role)+nl_tokens+tokenizer.encode(content)
    
    # 剩余token数
    left_token_space=generation_config.max_window_size

    # prompt头部: system发言
    system_text_part,system_tokens_part=_tokenize_str("system", system) # system_tokens_part -->    system\nYou are a helpful assistant.
    system_text=f'{im_start}{system_text_part}{im_end}'
    system_tokens=im_start_tokens+system_tokens_part+im_end_tokens # <|im_start|>system\nYou are a helpful assistant.<|im_end|>
    left_token_space-=len(system_tokens)
    
    # prompt尾部: user发言和assistant引导
    query_text_part,query_tokens_part=_tokenize_str('user', query)
    query_tokens_prefix=nl_tokens+ im_start_tokens
    query_tokens_suffix=im_end_tokens+nl_tokens+im_start_tokens+tokenizer.encode('assistant')+nl_tokens
    if len(query_tokens_prefix)+len(query_tokens_part)+len(query_tokens_suffix)>left_token_space: # query太长截断
        query_token_len=left_token_space-len(query_tokens_prefix)-len(query_tokens_suffix)
        query_tokens_part=query_tokens_part[:query_token_len]
        query_text_part=tokenizer.decode(query_tokens_part)
    query_tokens=query_tokens_prefix+query_tokens_part+query_tokens_suffix
    query_text=f"\n{im_start}{query_text_part}{im_end}\n{im_start}assistant\n"
    left_token_space-=len(query_tokens)
    
    # prompt腰部: 历史user+assitant对话
    history_text,history_tokens='',[]
    for hist_query,hist_response in reversed(history):    # 优先采用最近的对话历史
        hist_query_text,hist_query_tokens_part=_tokenize_str("user",hist_query) # user\n历史提问
        hist_response_text,hist_response_tokens_part=_tokenize_str("assistant",hist_response) # assistant\n历史回答
        # 生成本轮对话
        cur_history_tokens=nl_tokens+im_start_tokens+hist_query_tokens_part+im_end_tokens+nl_tokens+im_start_tokens+hist_response_tokens_part+im_end_tokens
        cur_history_text=f"\n{im_start}{hist_query_text}{im_end}\n{im_start}{hist_response_text}{im_end}"
        # 储存多轮对话
        if len(cur_history_tokens)<=left_token_space:
            history_text=cur_history_text+history_text
            history_tokens=cur_history_tokens+history_tokens
            left_token_space-=len(cur_history_tokens)
        else:
            break 
            
    # 生成完整Prompt
    prompt_str=f'{system_text}{history_text}{query_text}'
    prompt_tokens=system_tokens+history_tokens+query_tokens
    return prompt_str,prompt_tokens

# 停用词清理
def remove_stop_words(token_ids,stop_words_ids):
    token_ids=copy.deepcopy(token_ids)
    while len(token_ids)>0:
        if token_ids[-1] in stop_words_ids:
            token_ids.pop(-1)
        else:
            break
    return token_ids

2.2.2 客户端

        在客户端通过http方式进行访问。

instruction = '你现在是一个信息抽取模型，请你抽出输入文本中的条件...输入文本为：'
prefix_url = 'http://00.00.00.00:0000/chat?'
query = instruction + query
response = requests.post(prefix_url,json={'query':query})
temp_response = response.json()['text']
#如果不对就要修正
attempts = 0
max_attempts = 3  # 设置最大尝试修正次数
while attempts <= max_attempts:
    try:
        json_data = json.loads(temp_response)
        print("大模型返回的json",json_data)
        return json_data,200
    except:
        print('修正次数', attempts + 1)
        if attempts + 1 == max_attempts:
           return "This is a LLM error!", 501
        corrected_str = check_correct(temp_response)
        temp_response = corrected_str
        attempts += 1

2.2 方案二

        方案二是将模型的加载封装成类函数，通过实例化类方法加载大模型。

2.2.1 vLLM类

        将模型加载封装成类，在vllm_chat.py文件中实现。

import os
from vllm import AsyncEngineArgs, AsyncLLMEngine
from vllm.sampling_params import SamplingParams
from transformers import AutoTokenizer, GenerationConfig
from fastapi import FastAPI
import uvicorn
from prompt_utils import _build_prompt, remove_stop_words
import uuid
import json

class VLLMChat:
    def __init__(self, model_dir, tensor_parallel_size, gpu_memory_utilization, quantization, dtype):
        os.environ['CUDA_VISIBLE_DEVICES'] = '3'

        # vLLM参数
        self.model_dir = model_dir
        self.tensor_parallel_size = tensor_parallel_size
        self.gpu_memory_utilization = gpu_memory_utilization
        self.quantization = quantization
        self.dtype = dtype

        # vLLM模型加载
        self.load_vllm()

    def load_vllm(self):
        global generation_config, tokenizer, stop_words_ids, engine
        # 模型基础配置
        generation_config = GenerationConfig.from_pretrained(self.model_dir, trust_remote_code=True)
        # 加载分词器
        tokenizer = AutoTokenizer.from_pretrained(self.model_dir, trust_remote_code=True)
        tokenizer.eos_token_id = generation_config.eos_token_id
        # 推理终止词
        stop_words_ids = [tokenizer.im_start_id, tokenizer.im_end_id, tokenizer.eos_token_id]
        # vLLM基础配置
        args = AsyncEngineArgs(self.model_dir)
        args.worker_use_ray = False
        args.engine_use_ray = False
        args.tokenizer = self.model_dir
        args.tensor_parallel_size = self.tensor_parallel_size
        args.trust_remote_code = True
        #args.quantization = self.quantization
        args.gpu_memory_utilization = self.gpu_memory_utilization
        args.dtype = self.dtype
        args.max_num_seqs = 20  # batch最大20条样本
        # 加载模型
        os.environ['VLLM_USE_MODELSCOPE'] = 'False'
        engine = AsyncLLMEngine.from_engine_args(args)

    # 用户停止句匹配
    def match_user_stop_words(self, response_token_ids, user_stop_tokens):
        for stop_tokens in user_stop_tokens:
            if len(response_token_ids) < len(stop_tokens):
                continue
            if response_token_ids[-len(stop_tokens):] == stop_tokens:
                return True  # 命中停止句, 返回True
        return False

    async def chat(self, query, user_stop_words, history, system):
        # 用户停止词
        user_stop_tokens = []
        for words in user_stop_words:
            user_stop_tokens.append(tokenizer.encode(words))

        # 构造prompt
        prompt_text, prompt_tokens = _build_prompt(generation_config, tokenizer, query, history=history, system=system)

        # vLLM请求配置
        sampling_params = SamplingParams(stop_token_ids=stop_words_ids,
                                         early_stopping=False,
                                         top_p=generation_config.top_p,
                                         top_k=-1 if generation_config.top_k == 0 else generation_config.top_k,
                                         temperature=generation_config.temperature,
                                         repetition_penalty=generation_config.repetition_penalty,
                                         max_tokens=generation_config.max_new_tokens)
        # vLLM异步推理（在独立线程中阻塞执行推理，主线程异步等待完成通知）
        request_id = str(uuid.uuid4().hex)
        results_iter = engine.generate(prompt=None, sampling_params=sampling_params, prompt_token_ids=prompt_tokens,
                                       request_id=request_id)

        # 整体一次性返回模式
        async for result in results_iter:
            # 移除im_end,eos等系统停止词
            token_ids = remove_stop_words(result.outputs[0].token_ids, stop_words_ids)
            # 返回截止目前的tokens输出
            text = tokenizer.decode(token_ids)
            # 匹配用户停止词,终止推理
            if self.match_user_stop_words(token_ids, user_stop_tokens):
                await engine.abort(request_id)  # 终止vllm后续推理
                break

        ret = {"text": text}
        return ret

2.2.2 业务端

async def generate_json(query):
    user_stop_words = []
    history = []
    system = 'You are a helpful assistant.'
    ''''
    实例化的模型
    '''
    result = await vllm_chat_instance.chat(query, user_stop_words, history, system)
    temp_response =  result['text']

3.镜像加速 

        在Qwen2提供的官方镜像中，配备了vllm的加速框架。可以使用官方发布的镜像加载模型、加速推理。

3.1 启动镜像

# docker_run.sh
model_path="/root/ssd/Qwen2.5-3B-Instruct"
app_path="/root/Work/LLM_service/Qwen2.5-7B/Codes"
docker run  -d -it --shm-size=2g --ulimit memlock=-1 --ulimit stack=67108864 --gpus "device=0" \
    -p 19519:19519 \
    -v "$model_path":/data/shared/Qwen/Qwen2.5-32B \
    -v "$app_path":/data/shared/Qwen/Codes \
    -d qwenllm/qwen:2-cu121

3.2 启动服务

#run_vllm.sh
python -m vllm.entrypoints.openai.api_server \
    --trust-remote-code \
    --model /data/shared/Qwen/Qwen2.5-3B \
    --gpu-memory-utilization 0.9 \
    --tensor-parallel-size 1 \
    --max-model-len 8192 \
    --port 19519

3.3 服务端访问

from openai import OpenAI
import time
import math
openai_api_key = "EMPTY"
openai_api_base = "http://xx.xx.xx.xx:xx/v1/"

client = OpenAI(
    api_key=openai_api_key,
    base_url=openai_api_base,
)
input_string = '阿尔罕布拉宫是西班牙摩尔人统治时期的宫殿建筑。'
instruction = "任务：1.请你识别文本中的槽位信息；要求：1.输出格式务必为json格式;2.抽取的槽位值必须在原文中出现，不允许臆想与编造；3.只返回你的答案，禁止任何解释与说明；待识别的文本为："
system = "你是一个槽位信息抽取模型，可以抽取出文本中的槽位信息"
        
messages=[
        {"role": "system", "content": "prompt"},
        {"role": "user", "content": instruction + input_string}
            ]
response = client.chat.completions.create(
        model="/data/shared/Qwen/Qwen2.5-3B",
        messages=messages,
        stream=False,
        max_tokens=128,
        temperature=0.1,
        presence_penalty=1.2,
        top_p=0.9
    )
print(response)