开发大模型API接口开发思路

第一步、调用开源模型

开源大模型

from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.generation import GenerationConfig

model_path = os.path.join(current_path, "models/Qwen-7B-Chat")

# Model names: "Qwen/Qwen-7B-Chat", "Qwen/Qwen-14B-Chat"
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)

# use bf16
# model = AutoModelForCausalLM.from_pretrained(model_path, device_map="auto", trust_remote_code=True, bf16=True).eval()
# use fp16
# model = AutoModelForCausalLM.from_pretrained(model_path, device_map="auto", trust_remote_code=True, fp16=True).eval()
# use cpu only
# model = AutoModelForCausalLM.from_pretrained(model_path, device_map="cpu", trust_remote_code=True).eval()
# use auto mode, automatically select precision based on the device.
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    device_map="auto",
    trust_remote_code=True
).eval()

# Specify hyperparameters for generation. But if you use transformers>=4.32.0, there is no need to do this.
# model.generation_config = GenerationConfig.from_pretrained("Qwen/Qwen-7B-Chat", trust_remote_code=True)

# 1st dialogue turn
response, history = model.chat(tokenizer, "你好", history=None)
print(response)

# 2nd dialogue turn
response, history = model.chat(tokenizer, "给我讲一个年轻人奋斗创业最终取得成功的故事。", history=history)
print(response)

# 3rd dialogue turn
response, history = model.chat(tokenizer, "给这个故事起一个标题", history=history)
print(response)

参考： qwen官方仓库

调用精调后的模型

from transformers import AutoTokenizer
from peft import AutoPeftModelForCausalLM

model_path = os.path.join(current_path, "models/Qwen-finetune")

tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
model = AutoPeftModelForCausalLM.from_pretrained(model_path, device_map="auto", trust_remote_code=True, fp16=True).eval()

# 1st dialogue turn
response, history = model.chat(tokenizer, "你好", history=None)
print(response)

第二步、开发API

# -- coding: utf-8 --
import os
import tornado.httpserver  # tornado      HTTP Server implementation
import tornado.ioloop
import tornado.web
from logic import *
from conf import *
from my_logger import *


class HealthHandler(tornado.web.RequestHandler):
    def get(self):
        # 正确发送响应
        self.write('health')

    def post(self):
        # 正确发送响应
        self.write('health')


if __name__ == "__main__":
    print('---- ----  Service is being started! ---- ----')
    logger.info('---- ----  Service is being started! ---- ----')

    try:
        # 创建一个 Tornado Web 应用程序实例，其中包含一个路由，路径为 /server/，对应的处理程序是 MainHandler。
        app = tornado.web.Application(
            handlers=[
                (r"/health/", HealthHandler)  # 新添加的路由
            ]
        )
        # 创建一个 Tornado HTTP 服务器实例，并将之前创建的 Web 应用程序传递给构造函数。
        http_server = tornado.httpserver.HTTPServer(app)
        # 设置 HTTP 服务器监听的端口号。
        http_server.listen(port)
        print('---- ----  Listening to the port! ---- ----')
        logger.info('---- ----  Listening to the port! ---- ----')
        # 启动 Tornado 事件循环，以便处理传入的请求。
        tornado.ioloop.IOLoop.instance().start()
    # 如果在启动过程中出现任何异常，将打印异常并记录到日志。
    except Exception as e:
        print(e)
        logger.error('main start error: {}'.format(e))

这里使用 Tornado 框架作为服务器，并使用 Tornado 的 WebSocketHandler 来处理客户端的连接和消息。

只需要自己定义Handler（继承tornado.web.RequestHandler类），传参给web应用实例tornado.web.Application，就是可以自定义的大模型接口

第三步、API测试

import requests
import time
import json

base_url = "http://0.0.0.0:8501"

def test_qwen_api_server():

    header = {"appId": "coon","token": "92db43ea-b1ae-005056a542f0","requestId":"2021062100000931546240","requestTime":"2021-03-16 23:15:00"}

    prompt = "你是谁？"
    
    data = {
        "prompt": prompt
    }
    
    # 接口返回时间计时
    start_time = time.time()
    response = requests.post(f"{base_url}/server/", json=data, headers=header)
    end_time = time.time()
    toping = data["prompt"]
    print(f"Response Time for prompt '{toping[:30]}...': {end_time - start_time} seconds")
    
    # 将字符串转换为 JSON 对象  
    json_data = json.loads(response.text)
    # 输出 JSON 对象  
    print(json_data)  

    # 将字符串转换为 JSON 对象  
    json_data = json.loads(response.text)
    # 将 JSON 对象写入文件
    with open('response_data.json', 'a', encoding='utf-8') as file:
        json.dump(json_data, file, ensure_ascii=False, indent=4)
        file.write('\n')  # 在每个响应后添加换行符，以便区分不同的响应

test_qwen_api_server()