人工智能平台 PAI：Llama2-WebUI基于EAS的一键部署

背景信息

Llama2是由Meta开源的一个大语言模型，模型的参数范围从70亿到700亿不等，包括7B、13B、70B不同的版本。相比Llama1，Llama2的训练数据增加了40%，用了2万亿个Tokens进行训练。此外，Llama2的上下文长度是Llama1的两倍，达到4096。Llama2-Chat是基于Llama2在聊天对话场景微调的版本，采用了SFT（监督微调）和RLHF（人类反馈强化学习）进行迭代优化，以更好地和人类偏好保持一致并提高安全性。微调数据包括公开可用的指令数据集和超过一百万个人工标注样本。Llama2-Chat可用于类似助理的聊天，适用于各种自然语言生成任务。本文将以llama2-13b-chat基础模型为例，为您介绍如何在EAS中一键部署基于开源模型Llama2的ChatLLM-WebUI应用，并进行模型推理，同时提供常见问题的解决方法。

部署EAS服务

1. 进入PAI-EAS 模型在线服务页面。

   1. 登录PAI控制台。

   2. 在左侧导航栏单击工作空间列表，在工作空间列表页面中单击待操作的工作空间名称，进入对应工作空间内。

   3. 在工作空间页面的左侧导航栏选择模型部署>模型在线服务（EAS），进入PAI-EAS 模型在线服务页面。

2. 在PAI-EAS 模型在线服务页面，单击部署服务，在自定义部署区域中，单击立刻部署。

3. 在新建服务页面，配置以下关键参数。

   参数	描述
   服务名称	自定义服务名称。本案例使用的示例值为：chatllm_llama2_13b。
   部署方式	选择镜像部署AI-Web应用。
   镜像选择	在PAI平台镜像列表中选择chat-llm-webui；镜像版本选择2.0。 说明 由于版本迭代迅速，部署时镜像版本选择最高版本即可。
   运行命令	服务运行命令配置如下，本案例使用13b的模型进行部署。 如果使用13B的模型进行部署：python webui/webui_server.py --port=8000 --model-path=meta-llama/Llama-2-13b-chat-hf --precision=fp16。 如果使用7B的模型进行部署：python webui/webui_server.py --port=8000 --model-path=meta-llama/Llama-2-7b-chat-hf。 端口配置为8000。
   资源组种类	选择公共资源组。
   资源配置方法	选择常规资源配置。
   资源配置选择	必须选择GPU类型，实例规格推荐使用ecs.gn6e-c12g1.3xlarge。 13B的模型务必使用gn6e及更高规格的机型。 7B的模型可以使用A10或GU30机型。
   额外系统盘	配置为50 GB。

   

4. 单击部署，持续时间大约为5分钟，即可完成模型部署。

   当模型状态为运行中时，服务部署成功。

启动WebUI进行模型推理

1. 单击目标服务服务方式列下的查看Web应用。

2. 在WebUI页面进行模型推理验证。

   在对话框下方的输入文本框中输入对话内容，例如请提供一个理财学习计划，单击Send，即可开始对话。

常见问题及解决方法

如何使用API进行模型推理？

1. 获取服务访问地址和Token。

   1. 进入PAI-EAS 模型在线服务页面，详情请参见部署EAS服务。

   2. 在该页面中，单击目标服务名称进入服务详情页面。

   3. 在基本信息区域单击查看调用信息，在公网地址调用页签获取服务Token和访问地址。

2. 启动API进行模型推理。

使用HTTP方式调用服务

• 非流式调用

  客户端使用标准的HTTP格式，使用curl命令调用时，支持发送以下两种类型的请求：

  • 发送String类型的请求

    curl $host -H 'Authorization: $authorization' --data-binary @chatllm_data.txt -v

    其中：$authorization需替换为服务Token，$host：需替换为服务访问地址，chatllm_data.txt：该文件为包含问题的纯文本文件。

  • 发送结构化类型的请求

    curl $host -H 'Authorization: $authorization' -H "Content-type: application/json" --data-binary @chatllm_data.json -v -H "Connection: close"

    使用chatllm_data.json文件来设置推理参数，chatllm_data.json文件的内容格式如下：

    {
      "max_new_tokens": 4096,
      "use_stream_chat": false,
      "prompt": "How to install it?",
      "system_prompt": "Act like you are programmer with 5+ years of experience.",
      "history": [
        [
          "Can you tell me what's the bladellm?",
          "BladeLLM is an framework for LLM serving, integrated with acceleration techniques like quantization, ai compilation, etc. , and supporting popular LLMs like OPT, Bloom, LLaMA, etc."
        ]
      ],
      "temperature": 0.8,
      "top_k": 10,
      "top_p": 0.8,
      "do_sample": true,
      "use_cache": true
    }

    参数说明如下，请酌情添加或删除。

    参数	描述	默认值
    max_new_tokens	生成输出token的最大长度，单位为个。	2048
    use_stream_chat	是否使用流式输出形式。	true
    prompt	用户的Prompt。	""
    system_prompt	系统Prompt。	""
    history	对话的历史记录，类型为List[Tuple(str, str)]。	[()]
    temperature	用于调节模型输出结果的随机性，值越大随机性越强，0值为固定输出。Float类型，区间为0~1。	0.95
    top_k	从生成结果中选择候选输出的数量。	30
    top_p	从生成结果中按百分比选择输出结果。Float类型，区间为0~1。	0.8
    do_sample	开启输出采样。	true
    use_cache	开启kv cache。	true

  您也可以基于Python的requests包实现自己的客户端，示例代码如下：

  import argparse
  import json
  from typing import Iterable, List
  
  import requests
  
  def post_http_request(prompt: str,
                        system_prompt: str,
                        history: list,
                        host: str,
                        authorization: str,
                        max_new_tokens: int = 2048,
                        temperature: float = 0.95,
                        top_k: int = 1,
                        top_p: float = 0.8,
                        langchain: bool = False,
                        use_stream_chat: bool = False) -> requests.Response:
      headers = {
          "User-Agent": "Test Client",
          "Authorization": f"{authorization}"
      }
      if not history:
          history = [
              (
                  "San Francisco is a",
                  "city located in the state of California in the United States. \
                  It is known for its iconic landmarks, such as the Golden Gate Bridge \
                  and Alcatraz Island, as well as its vibrant culture, diverse population, \
                  and tech industry. The city is also home to many famous companies and \
                  startups, including Google, Apple, and Twitter."
              )
          ]
      pload = {
          "prompt": prompt,
          "system_prompt": system_prompt,
          "top_k": top_k,
          "top_p": top_p,
          "temperature": temperature,
          "max_new_tokens": max_new_tokens,
          "use_stream_chat": use_stream_chat,
          "history": history
      }
      if langchain:
          pload["langchain"] = langchain
      response = requests.post(host, headers=headers,
                               json=pload, stream=use_stream_chat)
      return response
  
  def get_response(response: requests.Response) -> List[str]:
      data = json.loads(response.content)
      output = data["response"]
      history = data["history"]
      return output, history
  
  if __name__ == "__main__":
      parser = argparse.ArgumentParser()
      parser.add_argument("--top-k", type=int, default=4)
      parser.add_argument("--top-p", type=float, default=0.8)
      parser.add_argument("--max-new-tokens", type=int, default=2048)
      parser.add_argument("--temperature", type=float, default=0.95)
      parser.add_argument("--prompt", type=str, default="How can I get there?")
      parser.add_argument("--langchain", action="store_true")
  
      args = parser.parse_args()
  
      prompt = args.prompt
      top_k = args.top_k
      top_p = args.top_p
      use_stream_chat = False
      temperature = args.temperature
      langchain = args.langchain
      max_new_tokens = args.max_new_tokens
  
      host = "EAS服务公网地址"
      authorization = "EAS服务公网Token"
  
      print(f"Prompt: {prompt!r}\n", flush=True)
      # 在客户端请求中可设置语言模型的system prompt。
      system_prompt = "Act like you are programmer with \
                  5+ years of experience."
  
      # 客户端请求中可设置对话的历史信息，客户端维护当前用户的对话记录，用于实现多轮对话。通常情况下可以使用上一轮对话返回的histroy信息，history格式为List[Tuple(str, str)]。
      history = []
      response = post_http_request(
          prompt, system_prompt, history,
          host, authorization,
          max_new_tokens, temperature, top_k, top_p,
          langchain=langchain, use_stream_chat=use_stream_chat)
      output, history = get_response(response)
      print(f" --- output: {output} \n --- history: {history}", flush=True)
  
  # 服务端返回JSON格式的响应结果，包含推理结果与对话历史。
  def get_response(response: requests.Response) -> List[str]:
      data = json.loads(response.content)
      output = data["response"]
      history = data["history"]
      return output, history

  其中：

  • host：配置为服务访问地址。

  • authorization：配置为服务Token。

• 流式调用

  流式调用使用HTTP SSE方式，其他设置方式与非流式相同，代码参考如下：

  import argparse
  import json
  from typing import Iterable, List
  
  import requests
  
  
  def clear_line(n: int = 1) -> None:
      LINE_UP = '\033[1A'
      LINE_CLEAR = '\x1b[2K'
      for _ in range(n):
          print(LINE_UP, end=LINE_CLEAR, flush=True)
  
  
  def post_http_request(prompt: str,
                        system_prompt: str,
                        history: list,
                        host: str,
                        authorization: str,
                        max_new_tokens: int = 2048,
                        temperature: float = 0.95,
                        top_k: int = 1,
                        top_p: float = 0.8,
                        langchain: bool = False,
                        use_stream_chat: bool = False) -> requests.Response:
      headers = {
          "User-Agent": "Test Client",
          "Authorization": f"{authorization}"
      }
      if not history:
          history = [
              (
                  "San Francisco is a",
                  "city located in the state of California in the United States. \
                  It is known for its iconic landmarks, such as the Golden Gate Bridge \
                  and Alcatraz Island, as well as its vibrant culture, diverse population, \
                  and tech industry. The city is also home to many famous companies and \
                  startups, including Google, Apple, and Twitter."
              )
          ]
      pload = {
          "prompt": prompt,
          "system_prompt": system_prompt,
          "top_k": top_k,
          "top_p": top_p,
          "temperature": temperature,
          "max_new_tokens": max_new_tokens,
          "use_stream_chat": use_stream_chat,
          "history": history
      }
      if langchain:
          pload["langchain"] = langchain
      response = requests.post(host, headers=headers,
                               json=pload, stream=use_stream_chat)
      return response
  
  
  def get_streaming_response(response: requests.Response) -> Iterable[List[str]]:
      for chunk in response.iter_lines(chunk_size=8192,
                                       decode_unicode=False,
                                       delimiter=b"\0"):
          if chunk:
              data = json.loads(chunk.decode("utf-8"))
              output = data["response"]
              history = data["history"]
              yield output, history
  
  
  if __name__ == "__main__":
      parser = argparse.ArgumentParser()
      parser.add_argument("--top-k", type=int, default=4)
      parser.add_argument("--top-p", type=float, default=0.8)
      parser.add_argument("--max-new-tokens", type=int, default=2048)
      parser.add_argument("--temperature", type=float, default=0.95)
      parser.add_argument("--prompt", type=str, default="How can I get there?")
      parser.add_argument("--langchain", action="store_true")
      args = parser.parse_args()
  
      prompt = args.prompt
      top_k = args.top_k
      top_p = args.top_p
      use_stream_chat = True
      temperature = args.temperature
      langchain = args.langchain
      max_new_tokens = args.max_new_tokens
  
      host = ""
      authorization = ""
  
      print(f"Prompt: {prompt!r}\n", flush=True)
      system_prompt = "Act like you are programmer with \
                  5+ years of experience."
      history = []
      response = post_http_request(
          prompt, system_prompt, history,
          host, authorization,
          max_new_tokens, temperature, top_k, top_p,
          langchain=langchain, use_stream_chat=use_stream_chat)
  
      for h, history in get_streaming_response(response):
          print(
              f" --- stream line: {h} \n --- history: {history}", flush=True)
  

  其中：

  • host：配置为服务访问地址。

  • authorization：配置为服务Token。

使用WebSocket方式调用服务

为了更好地维护用户对话信息，您也可以使用WebSocket方式保持与服务的连接完成单轮或多轮对话，代码示例如下：

import os
import time
import json
import struct
from multiprocessing import Process

import websocket

round = 5
questions = 0


def on_message_1(ws, message):
    if message == "<EOS>":
        print('pid-{} timestamp-({}) receives end message: {}'.format(os.getpid(),
              time.time(), message), flush=True)
        ws.send(struct.pack('!H', 1000), websocket.ABNF.OPCODE_CLOSE)
    else:
        print("{}".format(time.time()))
        print('pid-{} timestamp-({}) --- message received: {}'.format(os.getpid(),
              time.time(), message), flush=True)


def on_message_2(ws, message):
    global questions
    print('pid-{} --- message received: {}'.format(os.getpid(), message))
    # end the client-side streaming
    if message == "<EOS>":
        questions = questions + 1
        if questions == 5:
            ws.send(struct.pack('!H', 1000), websocket.ABNF.OPCODE_CLOSE)


def on_message_3(ws, message):
    print('pid-{} --- message received: {}'.format(os.getpid(), message))
    # end the client-side streaming
    ws.send(struct.pack('!H', 1000), websocket.ABNF.OPCODE_CLOSE)


def on_error(ws, error):
    print('error happened: ', str(error))


def on_close(ws, a, b):
    print("### closed ###", a, b)


def on_pong(ws, pong):
    print('pong:', pong)

# stream chat validation test
def on_open_1(ws):
    print('Opening Websocket connection to the server ... ')
    params_dict = {}
    params_dict['prompt'] = """Show me a golang code example: """
    params_dict['temperature'] = 0.9
    params_dict['top_p'] = 0.1
    params_dict['top_k'] = 30
    params_dict['max_new_tokens'] = 2048
    params_dict['do_sample'] = True
    raw_req = json.dumps(params_dict, ensure_ascii=False).encode('utf8')
    # raw_req = f"""To open a Websocket connection to the server: """

    ws.send(raw_req)
    # end the client-side streaming


# multi-round query validation test
def on_open_2(ws):
    global round
    print('Opening Websocket connection to the server ... ')
    params_dict = {"max_new_tokens": 6144}
    params_dict['temperature'] = 0.9
    params_dict['top_p'] = 0.1
    params_dict['top_k'] = 30
    params_dict['use_stream_chat'] = True
    params_dict['prompt'] = "您好！"
    params_dict = {
        "system_prompt":
        "Act like you are programmer with 5+ years of experience."
    }
    raw_req = json.dumps(params_dict, ensure_ascii=False).encode('utf8')
    ws.send(raw_req)
    params_dict['prompt'] = "请使用Python，编写一个排序算法"
    raw_req = json.dumps(params_dict, ensure_ascii=False).encode('utf8')
    ws.send(raw_req)
    params_dict['prompt'] = "请转写成java语言的实现"
    raw_req = json.dumps(params_dict, ensure_ascii=False).encode('utf8')
    ws.send(raw_req)
    params_dict['prompt'] = "请介绍一下你自己？"
    raw_req = json.dumps(params_dict, ensure_ascii=False).encode('utf8')
    ws.send(raw_req)
    params_dict['prompt'] = "请总结上述对话"
    raw_req = json.dumps(params_dict, ensure_ascii=False).encode('utf8')
    ws.send(raw_req)


# Langchain validation test.
def on_open_3(ws):
    global round
    print('Opening Websocket connection to the server ... ')

    params_dict = {}
    # params_dict['prompt'] = """To open a Websocket connection to the server: """
    params_dict['prompt'] = """Can you tell me what's the MNN?"""
    params_dict['temperature'] = 0.9
    params_dict['top_p'] = 0.1
    params_dict['top_k'] = 30
    params_dict['max_new_tokens'] = 2048
    params_dict['use_stream_chat'] = False
    params_dict['langchain'] = True
    raw_req = json.dumps(params_dict, ensure_ascii=False).encode('utf8')
    ws.send(raw_req)


authorization = ""
host = "ws://" + ""


def single_call(on_open_func, on_message_func, on_clonse_func=on_close):
    ws = websocket.WebSocketApp(
        host,
        on_open=on_open_func,
        on_message=on_message_func,
        on_error=on_error,
        on_pong=on_pong,
        on_close=on_clonse_func,
        header=[
            'Authorization: ' + authorization],
    )

    # setup ping interval to keep long connection.
    ws.run_forever(ping_interval=2)


if __name__ == "__main__":
    for i in range(5):
        p1 = Process(target=single_call, args=(on_open_1, on_message_1))
        p2 = Process(target=single_call, args=(on_open_2, on_message_2))
        p3 = Process(target=single_call, args=(on_open_3, on_message_3))

        p1.start()
        p2.start()
        p3.start()

        p1.join()
        p2.join()
        p3.join()

其中：

• authorization：配置为服务Token。

• host：配置为服务访问地址。并将访问地址中前端的http替换为ws。

• use_stream_chat：通过该请求参数来控制客户端是否为流式输出。默认值为True，表示服务端返回流式数据。

• 参考上述示例代码中的on_open_2函数的实现方法实现多轮对话。

相关文档

• 更多关于EAS的内容介绍，请参见EAS模型服务概述。

• 您也可以使用本地知识库检索功能，详情请参见PAI+向量检索快速搭建大模型知识库对话。