DeepGPU-LLM的API接口说明及示例

模型加载函数

DeepGPU-LLM提供统一的模型处理类deepgpu_model，具体接口代码如下。其中，该模型处理类包含模型加载函数__init__函数和from_pretrained函数。

加载函数介绍

class deepgpu_model(torch.nn.Module):
    def __init__(self, model_path: str, tensor_para_size: int, precision: int = 0,
                 kv_cache_quant_level: int = 0,
                 generation_config: typing.Optional[DeepGPUGenerationConfig] = None,
                 is_gemm_tuning: bool = False,
                 load_pretrained: bool = False,
                 page_size: int = 16,
                 gpu_utilization: float = 0.9,
                 max_batch_size : int = 128,
                 max_context_token_num: int = 8192,
                 session_len: int = 2048)


    @classmethod
    def from_pretrained(cls, model_path, tensor_para_size, precision = 0,
                 kv_cache_quant_level: int = 0,
                 generation_config: typing.Optional[DeepGPUGenerationConfig] = None,
                 page_size: int = 16,
                 gpu_utilization: float = 0.9,
                 max_batch_size : int = 128,
                 max_context_token_num: int = 8192,
                 session_len: int = 2048,
                 data_set = None)

• __init__函数：用于加载已转换好格式的模型。具体参数介绍如下：

  参数名	说明
  model_path	指向格式已转换好的模型目录。 说明 一般为x-gpu目录，其中x为多少个GPU，模型转换时自动生成的目录。
  tensor_para_size	使用的GPU数量，需要与模型转换设置的GPU数量保持一致。
  precision	权重量化级别。取值范围： 0：表示fp16精度。 1：表示int8量化。 3：表示int4量化。 5：表示fp8精度。 默认值：0。
  kv_cache_quant_level	kv_cache量化级别。取值范围： 0：表示无量化压缩。 1：表示K8_V8量化，推荐值。 2：表示K8_V4量化。 3：表示K4_V4量化。 默认值：0。
  generation_config	模型配置参数。更多信息，请参见模型参数类（DeepGPUGenerationConfig）。 说明 若不设置该参数，则会加载模型配置文件中的默认参数。
  is_gemm_tuning	是否优化最佳Kernel，常见GPU的最佳Kernel已优化好并存储于安装包内。取值范围： False：不优化最佳Kernel。 True：优化最佳Kernel。 默认值：False。
  load_pretrained	加载的模型是否为原始模型。取值范围： False：表示加载格式已转换好的模型。 True：表示需要加载格式未转换的原始模型，不可直接调用，主要是提供from_pretrained函数内部调用以实现从原始模型加载的功能。 默认值：False。
  page_size	page attention所使用的内存块大小。
  gpu_utilization	GPU显存容量使用比例。
  max_batch_size	能够同时处理的最大请求数。 说明 若资源不足以支持配置的值，系统也会根据GPU显存资源进行调整。
  max_context_token_num	上下文限制的最大Token数量，该值不能小于session_len值。
  session_len	对话长度限制，即输入和输出的Tokens数量。 若该值超过session_len默认值，则需要配置该参数。
  data_set	配置fp8量化使用的标定数据目录。 若不使用fp8精度，则不需要设置该参数。

• from_pretrained函数：直接加载从huggingface和modelscope平台上下载的开源模型，加载过程中支持在线自动转换模型。

  该函数是一个classmethod类型的函数，会在内部调用__init__创建一个deepgpu_model类，可以从原始模型目录加载模型，进行模型转换并进行初始化。关于from_pretrained函数的具体参数介绍，请参见上表中的__init__函数具体参数介绍。

常用类介绍

deepgpu_utils.py 通常提供一些实用的函数和工具，以支持DeepGPU-LLM的推理API接口，deepgpu_utils.py定义了一些常用类，例如，模型参数类（DeepGPUGenerationConfig）、并发流处理类（DeepGPUStreamer）以及请求ID处理类（RequestCounter）。这些类可以管理模型的运行环境、处理并发请求以及跟踪请求的执行情况，您可以通过以下代码将这些类导入至deepgpu_utils.py中。

from deepgpu_llm.deepgpu_utils import DeepGPUGenerationConfig, DeepGPUStreamer, RequestCounter

• 模型参数类（DeepGPUGenerationConfig）

  DeepGPUGenerationConfig类主要用于给模型传递推理运行相关参数，详见以下代码。具体参数可以不设置，DeepGPU-LLM会自动从模型配置文件里获取相关初始参数。

  class DeepGPUGenerationConfig():
      def __init__(self, **kwargs):
          self.max_new_tokens = kwargs.pop("max_new_tokens", 512)
          self.do_sample = kwargs.pop("do_sample", None)    
          self.num_beams = kwargs.pop("num_beams", None)  
          self.temperature = kwargs.pop("temperature", None)  
          self.top_k = kwargs.pop("top_k", None) 
          self.top_p = kwargs.pop("top_p", None) 
          self.repetition_penalty = kwargs.pop("repetition_penalty", None)    
          self.presence_penalty = kwargs.pop("presence_penalty", None) 
          self.len_penalty = kwargs.pop("len_penalty", None)
          self.beam_search_diversity_rate = kwargs.pop("beam_search_diversity_rate", None) 
          self.min_tokens = kwargs.pop("min_tokens", 0)

• 并发流处理类（DeepGPUStreamer）

  DeepGPUStreamer类用于处理大语言模型的流式输出和多请求输出，详见以下代码。

  class DeepGPUStreamer():
      def __init__(self, tokenizer: "AutoTokenizer", 
                  skip_prompt: bool = False, 
                  print_out: bool = True,
                  **decode_kwargs)
  
      def handle_str(self, value)
      def end(self)

  函数名	说明
  __init__()	初始化内部使用到的tokenizer和用于打印的字符串缓存区。
  handle_str()	接受一个token id，并转换为自然语言文本，然后放入字符串缓存区中，返回字符串缓存区中可以打印的字符串。
  end()	返回字符串缓存区中剩余的字符串。

• 请求ID处理类（RequestCounter）

  RequestCounter类主要用于多请求场景中的request_id管理，详见以下代码。

  class RequestCounter:
      def __init__(self, start: int = 0) -> None
      def __next__(self) -> int
      def reset(self) -> None
      def add(self, len: int) -> int
      def cur(self) -> int

  函数名	说明
  __init__()	对RequestCounter类进行初始化，默认值为0，您也可以通过设置start值对类进行初始化。
  __next__()	获取当前counter值，并对内部的counter累加1。
  reset()	内部的counter重置为0，以便从头开始生成请求ID。
  add()	返回内部原始counter值，然后将counter加上len作为新值。
  cur()	获取当前counter值，不进行其他操作。

DeepGPU-LLM实例代码

本文提供了一系列DeepGPU-LLM的参考代码，帮助您更快速地上手进行编程开发。

离线（offline）代码示例

• 简单Llama模型示例

  本示例以llama2-7b-chat模型为例，运行以下deepgpu-llm实例代码前，请确保您已经对lama2-7b-chat模型进行模型转换操作，同时，您也可以根据实际情况调整模型目录、GPU数量、量化精度等参数。

  import time
  from deepgpu_llm.deepgpu_model import deepgpu_model
  from deepgpu_llm.deepgpu_utils import DeepGPUGenerationConfig
  from transformers import LlamaTokenizer
  
  model_path = '/mnt/models_deepgpu/llama2-7b-chat'
  tokenizer = LlamaTokenizer.from_pretrained(model_path)
  
  model_path_conv = "/mnt/models_deepgpu/llama2-7b-chat/1-gpu"
  tensor_para_size = 1
  precision = 0  # 0:fp16 mode, 1:int8 mode, 3:int4 mode
  kv_cache_quant_level = 0 # 0:no quant, 1:K8_V8, 2:K8_v4, 3:K4_V4
  generation_config = DeepGPUGenerationConfig(max_new_tokens=512)
  model = deepgpu_model(model_path_conv, tensor_para_size,
                      precision, kv_cache_quant_level,
                      generation_config=generation_config)
  
  payload = "Hi, please introduce the alibaba?"
  start_ids = [tokenizer(payload, return_tensors="pt").input_ids]
  
  print(payload)
  
  for i in range(5):
      s = time.time()
      output = model.generate(start_ids, generation_config)
      e = time.time()
      print("---- time", e - s)
  
  tokens = output[0].tolist()
  for i in range(len(tokens)):
      print(tokenizer.decode(tokens[i][0]))

  运行上述实例代码后，显示结果如下所示：

  

• 流式输出Qwen模型示例

  本示例以单卡GPU上运行qwen-7b模型的实例代码为例，您可以调整代码中precision量化精度、kv_cache_quant_level量化级别、tp_sizeGPU数量，获得更快的性能和减少GPU显存占用。运行下列代码后将结果流式打印输出，并统计性能。

  import time
  from deepgpu_llm.deepgpu_model import deepgpu_model
  from deepgpu_llm.deepgpu_utils import DeepGPUGenerationConfig, DeepGPUStreamer
  from transformers import AutoTokenizer
  
  model_path = "/mnt/models_deepgpu/Qwen2-7B-Instruct"
  tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
  
  model_path_conv = "/mnt/models_deepgpu/Qwen2-7B-Instruct/1-gpu"
  tp_size = 1    # tensor parallel
  precision = 0  # 0:fp16 mode, 1:int8 mode, 3:int4 mode
  kv_cache_quant_level = 0 # 0:no quant, 1:K8_V8, 2:K8_v4, 3:K4_V4
  generation_config = DeepGPUGenerationConfig(max_new_tokens=512)
  model = deepgpu_model(model_path_conv, tp_size, precision, kv_cache_quant_level,
                     generation_config=generation_config)
  print("model init over!")
  
  
  payload = "<|im_start|>user\n 你好，请介绍下杭州的旅游景区？ <|im_end|>\n<|im_start|>assistant\n"
  start_ids = tokenizer(payload, return_tensors="pt").input_ids
  
  streamer = DeepGPUStreamer(tokenizer, **{'skip_special_tokens':True})
  
  total_len = 0
  response = ""
  print("response : ")
  start = time.time()
  for output in model.stream_generate([start_ids],
                      generation_config=generation_config,
                      skip_inputs=True):
      printable_str = streamer.handel_str(output)
      #response = response + printable_str
      total_len += 1
      print(printable_str, flush=True, end = "")
  end = time.time()
  print()
  print()
  print("time : ", (end - start))
  print("speed: ", total_len/(end - start), " tokens/s")

  运行上述实例代码，输出结果如下所示。问题回复结果是流式输出的，此时您可以体验到DeepGPU-LLM推理速度了。

  

  说明

  如需查看其他类型LLM模型的流式输出实例代码，您可以通过执行以下命令直接查看DeepGPU-LLM的安装目录，找到llama_cli、chatglm_cli、baichuan_cli和qwen_cli脚本，即可查看相应代码。

  pip show -f deepgpu-llm

• 多batch同时输入的示例

  针对于多batch场景，关键是处理输入和输出，各类模型输入输出的处理方法，请参见普通输出函数generate（一次性输出）。

  import time
  from deepgpu_llm.deepgpu_model import deepgpu_model
  from deepgpu_llm.deepgpu_utils import DeepGPUGenerationConfig
  from transformers import AutoTokenizer
  
  model_path = "/mnt/models_deepgpu/Qwen1.5-72B-Chat"
  tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
  
  model_path_conv = "/mnt/models_deepgpu/Qwen1.5-72B-Chat/4-gpu"
  tp_size = 4    # tensor parallel
  precision = 0  # 0:fp16 mode, 1:int8 mode, 3:int4 mode
  kv_cache_quant_level = 0 # 0:no quant, 1:K8_V8, 2:K8_v4, 3:K4_V4
  generation_config = DeepGPUGenerationConfig(max_new_tokens=512)
  model = deepgpu_model(model_path_conv, tp_size, precision, kv_cache_quant_level,
                     generation_config=generation_config)
  print("model init over!")
  
  prompt =["你好，你是谁？", "你好，请介绍下杭州的旅游景区？", "你好，请介绍下北京的旅游景区？", "你好，请介绍下新疆的旅游景区？", "你好，请介绍下西藏的旅游景区？"]
  batchsize = len(prompt)
  start_ids = []
  for bs in range(batchsize):
    payload = "<|im_start|>user\n " + prompt[bs] + "<|im_end|>\n<|im_start|>assistant\n"
    start_ids.append(tokenizer(payload, return_tensors="pt").input_ids)
  
  output = model.generate(start_ids, generation_config)
  
  tokens = output[0].tolist()
  for bs in range(batchsize):
    response = tokenizer.decode(tokens[bs][0], skip_special_tokens=True)
    print("response [", bs, "] : ", response.rstrip(tokenizer.decode(0)))

  运行上述代码，输出结果如下所示：

  

在线（serving）代码示例

• 服务端代码示例

  执行deepgpu fastapi_server.py脚本文件，实现接收文本生成请求，使用DeepGPU模型进行文本生成，并返回生成的文本。

  import argparse
  import json
  from typing import AsyncGenerator
  from fastapi import FastAPI, Request
  from fastapi.responses import JSONResponse, Response, StreamingResponse
  import uvicorn
  from deepgpu_llm.deepgpu_model import deepgpu_model
  from deepgpu_llm.deepgpu_utils import DeepGPUGenerationConfig,DeepGPUStreamer,RequestCounter
  from transformers import AutoTokenizer
  import time
  TIMEOUT_KEEP_ALIVE = 5  # seconds.
  TIMEOUT_TO_PREVENT_DEADLOCK = 1  # seconds.
  app = FastAPI()
  engine = None
  counter = RequestCounter()
  tokenizer = None
  
  @app.get("/health")
  async def health() -> Response:
      """Health check."""
      return Response(status_code=200)
  @app.post("/generate")
  async def generate(request: Request) -> Response:
      
      request_dict = await request.json()
      prompt = request_dict.pop("prompt")
      max_new_token = request_dict.pop("max_tokens")
      prompt = [tokenizer(prompt, return_tensors='pt').input_ids]
      stream = request_dict.pop("stream", True)
      streamer = DeepGPUStreamer(tokenizer=tokenizer)
      generation_config = DeepGPUGenerationConfig(max_new_tokens = max_new_token)
      results_generator = engine.generate_cb_async(prompt,generation_config=generation_config)
      # Streaming case
      async def stream_results() -> AsyncGenerator[bytes, None]:
          # await asyncio.sleep(0)
          async for request_output in results_generator:
              if(request_output==-1):
                  printable_str = streamer.end()
                  text_outputs = [
                      printable_str
                      ]
                  ret = {"text": text_outputs}
                  yield (json.dumps(ret) + "\0").encode("utf-8")
                  break
              else:
                  printable_str = streamer.handel_str(request_output)
                  
              text_outputs = [
                  printable_str
              ]
              ret = {"text": text_outputs}
              yield (json.dumps(ret) + "\0").encode("utf-8")
  
      if stream:
          return StreamingResponse(stream_results())
  
      # Non-streaming case
      total_str = ""
      async for request_output in results_generator:
          if(request_output == -1):
              total_str += streamer.end()
              break
          else:
              total_str += streamer.handel_str(request_output)
      ret = {"text": total_str}
      return JSONResponse(ret)
  
  
  if __name__ == "__main__":
      parser = argparse.ArgumentParser(formatter_class=argparse.RawTextHelpFormatter)
      parser.add_argument("--host", type=str, default=None)
      parser.add_argument("--port", type=int, default=8000)
      parser.add_argument('--model_dir', '-i', type=str, help='converted model dir', required=True)
      parser.add_argument('--tokenizer_dir', '-t', type=str, help='tokenizer dir', required=False)
      parser.add_argument('--tp_size', '-tp', type=int, help='tensor para size', required=True)
      args = parser.parse_args()
      tokenizer = AutoTokenizer.from_pretrained(args.tokenizer_dir, trust_remote_code=True)
      model_path = args.model_dir
      tp_size = args.tp_size
      precision = 0 # 0:fp16 mode, 1:int8 mode, 3:int4 mode
      engine = deepgpu_model(model_path, tp_size, precision,is_gemm_tuning=False)
  
      uvicorn.run(app,
                  host=args.host,
                  port=args.port,
                  log_level="debug",
                  timeout_keep_alive=TIMEOUT_KEEP_ALIVE)

  继续运行以下命令，启动服务。

  python3 fastapi_server.py -i /mnt/models_deepgpu/Qwen2-7B-Instruct/1-gpu -t /mnt/models_deepgpu/Qwen2-7B-Instruct -tp 1

  如果显示以下结果，表示该服务搭建成功。

  

  说明

  DeepGPU兼容vLLM的serving实现，您可以直接使用vLLM的官方代码benchmark_serving.py作为参考，仅需将其内部的import vllm更改为import deepgpu即可。更多信息，请参见更多示例（将vLLM切换为DeepGPU-LLM推理引擎）。

• 客户端python代码

  本示例以一个异步的Python脚本为例，实现向一个HTTP服务发送请求，获取文本生成的流式响应，并打印这些响应的功能。

  import requests
  import json
  import time
  import aiohttp
  import asyncio
  import random
  import os
  url = "http://0.0.0.0:8000/generate" # 普通服务模式
  # url = "http://0.0.0.0:8000/v1/completions" # openai接口普通服务模式
  headers = {"User-Agent": "Benchmark Client"} # 普通服务模式
  # headers = {
  # "Authorization": f"Bearer {os.environ.get('OPENAI_API_KEY')}" #openai接口普通服务模式
  # }
  async def stream_back(data):
      async with aiohttp.ClientSession() as session:
          async with session.post(url, headers=headers, data=json.dumps(data)) as response:
              if response.status == 200:
                  async for chunk, _ in response.content.iter_chunks():
                      chunk = chunk.rstrip(b'\x00')
                      decoded_chunk = chunk.decode('utf-8')
                      if(not decoded_chunk.strip()):
                          continue
                      try:
                          data = json.loads(decoded_chunk)
                          text_content = data.get('text', [])
                          for text in text_content:
                              if text:
                                  print(text,flush=True,end='')
                                  continue
                      except json.JSONDecodeError as e:
                              print("JSON decode error! error code:", e) 
                              
  
  async def fetch_stream(prompt: str):
      max_tokens = 512
      # open ai服务使用的data
      # data = { 
          # "model": '/root/models/Llama-2-7b-chat-hf/',
          # "prompt": prompt,
          # "temperature": 0.0,
          # "best_of": 1,
          # "max_tokens": 512,
          # "stream": True,
      # }
      # 普通服务使用的data
      print("[prompt]: ", prompt)
      data = {"prompt": prompt, "max_tokens": max_tokens, "stream": True}
      async with aiohttp.ClientSession() as session:
          start_time = time.time()
          await stream_back(data)
          # await none_stream_back(data) 
          end_time = time.time()
          latency = end_time - start_time
          print()
          print(f"Latency: {latency} seconds")
          
  
  async def main(prompt_num):
      prompt =["你好，你是谁？",
               "请介绍下杭州的旅游景区？",
               "请介绍下北京的旅游景区？",
               "请介绍下杭州的历史变迁？",
               "你知道BAT吗？",
               "考上大学后，需要如何填报志愿？",
               "介绍下浙江最好的几所大学？"]
      tasks = [fetch_stream(prompt[random.randint(0,len(prompt)-1)]) for _ in range(prompt_num)]
      await asyncio.gather(*tasks)
          
  if __name__ == "__main__":
      asyncio.run(main(1))

  运行上述代码后，显示结果如下所示：

  

• 客户端curl命令

  启动服务后，执行以下命令给服务端发送请求，实现一段文本的输出。

  curl -X POST "http://0.0.0.0:8000/generate" \
       -H "Content-Type: application/json" \
       -H "User-Agent: Benchmark Client" \
       -d '{"prompt": "请介绍下杭州的历史变迁？", "stream": false, "max_tokens": 128}'

  说明

  curl命令交互不支持流式输出，stream只能设置为false。

  运行以上命令后，显示结果如下所示：

  

更多示例（将vLLM切换为DeepGPU-LLM推理引擎）

DeepGPU-LLM提供了类似vLLM的接口，如果您需要在不改变现有vLLM代码结构的情况下，基于开源vLLM代码切换为DeepGPU-LLM推理引擎，实现DeepGPU-LLM带来的性能提升和功能扩展，请参考以下两种方式的推理示例：

兼容vLLM的DeepGPU-LLM推理示例（offline）

1. 获取vLLM的离线推理示例。

   offline_inference.py示例如下：

   from vllm import LLM, SamplingParams
   
   # Sample prompts.
   prompts = [
       "Hello, my name is",
       "The president of the United States is",
       "The capital of France is",
       "The future of AI is",
   ]
   # Create a sampling params object.
   sampling_params = SamplingParams(temperature=0.8, top_p=0.95)
   
   # Create an LLM.
   llm = LLM(model="facebook/opt-125m")
   # Generate texts from the prompts. The output is a list of RequestOutput objects
   # that contain the prompt, generated text, and other information.
   outputs = llm.generate(prompts, sampling_params)
   # Print the outputs.
   for output in outputs:
       prompt = output.prompt
       generated_text = output.outputs[0].text
       print(f"Prompt: {prompt!r}, Generated text: {generated_text!r}")

2. 基于vLLM示例将推理引擎更改为DeepGPU示例。

   修改以下代码完成推理引擎切换，即将导入包由vllm改为deepgpu_llm，模型文件改为转换后的DeepGPU模型文件。

   # from vllm import LLM, SamplingParams 更改为：
   from deepgpu_llm import LLM, SamplingParams
   # llm = LLM(model="facebook/opt-125m") 更改为
   llm = LLM("path/to/your/deepgpu_model")

兼容vLLM的DeepGPU-LLM多请求推理示例（serving）

1. 在线获取vLLM的推理示例。

   获取路径：api_server。

2. 基于vLLM示例将推理引擎更改为DeepGPU示例。

   修改以下代码，即将vllm在线代码更改为deepgpu_llm相关示例。

   # from vllm.engine.arg_utils import AsyncEngineArgs 更改为：
   from deepgpu_llm.engine.arg_utils import AsyncEngineArgs
   # from vllm.engine.async_llm_engine import AsyncLLMEngine 更改为：
   from deepgpu_llm.engine.async_llm_engine import AsyncLLMEngine
   # from vllm.sampling_params import SamplingParams 更改为：
   from deepgpu_llm.sampling_params import SamplingParams
   # from vllm.usage.usage_lib import UsageContext 更改为：
   from deepgpu_llm.usage.usage_lib import UsageContext
   # from vllm.utils import random_uuid 更改为：
   from deepgpu_llm.utils import random_uuid

3. 启动基于DeepGPU-LLM的API服务器，该服务器可以提供模型推理服务。

   # 拉起普通服务：
   python3 -m deepgpu_llm.entrypoints.api_server \
       --model <your model> \
       --trust-remote-code \
       --tensor-parallel-size tp_tpsize \
       --gpu-memory-utilization 0.95 
   
   # 拉起openai接口服务
   python3 -m deepgpu_llm.entrypoints.openai.api_server \
       --model <YOUR_MODEL> \
       --trust-remote-code \
       --tensor-parallel-size tp_tpsize \
       --gpu-memory-utilization 0.95 

4. 使用benchmark_serving.py脚本和ShareGPT数据集进行推理性能基准测试。

   python3 benchmark_serving.py \
       --backend vllm \
       --model /root/deepgpu/models/qwen1.5-7b \
       --tokenizer /root/deepgpu/models/qwen1.5-7b \
       --dataset-name sharegpt \
       --dataset-path ShareGPT_V3_unfiltered_cleaned_split.json \
       --request-rate 30 \
       --num-prompts 2000 \
       --port 8000