From 706c67159ca965f90e3c8d6088df6d5e6a2e2b3d Mon Sep 17 00:00:00 2001 From: "majiahui@haimaqingfan.com" Date: Fri, 14 Feb 2025 14:36:17 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E6=A8=A1=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- mistral_api.py | 131 ------------------ mistral_model_predict_vllm_1.py | 106 +++++++++++++++ mistral_model_predict_vllm_4.py | 106 +++++++++++++++ model_api.py | 135 +++++++++++++++++++ openbuddy_llama3_1_model_predict_vllm_1.py | 202 ++++++++++++++++++++++++++++ openbuddy_llama3_1_model_predict_vllm_2.py | 202 ++++++++++++++++++++++++++++ openbuddy_llama3_1_model_predict_vllm_3.py | 202 ++++++++++++++++++++++++++++ openbuddy_qwen2_5_model_predict_vllm_1.py | 202 ++++++++++++++++++++++++++++ openbuddy_qwen2_5_model_predict_vllm_2.py | 202 ++++++++++++++++++++++++++++ openbuddy_qwen2_5_model_predict_vllm_3.py | 202 ++++++++++++++++++++++++++++ qwen2_5_Instruct_model_predict_vllm_1.py | 205 +++++++++++++++++++++++++++++ qwen2_5_Instruct_model_predict_vllm_2.py | 205 +++++++++++++++++++++++++++++ qwen2_5_Instruct_model_predict_vllm_3.py | 205 +++++++++++++++++++++++++++++ run_api_gunicorn.sh | 2 +- run_model_1.sh | 1 + run_model_4.sh | 1 + run_model_openbuddy_llama3_1_1.sh | 1 + run_model_openbuddy_llama3_1_2.sh | 1 + run_model_openbuddy_llama3_1_3.sh | 1 + run_model_openbuddy_qwen_1.sh | 1 + run_model_openbuddy_qwen_2.sh | 1 + run_model_openbuddy_qwen_3.sh | 1 + run_model_qwen_Instruct1.sh | 1 + run_model_qwen_Instruct2.sh | 1 + run_model_qwen_Instruct3.sh | 1 + 25 files changed, 2186 insertions(+), 132 deletions(-) delete mode 100644 mistral_api.py create mode 100644 mistral_model_predict_vllm_1.py create mode 100644 mistral_model_predict_vllm_4.py create mode 100644 model_api.py create mode 100644 openbuddy_llama3_1_model_predict_vllm_1.py create mode 100644 openbuddy_llama3_1_model_predict_vllm_2.py create mode 100644 openbuddy_llama3_1_model_predict_vllm_3.py create mode 100644 openbuddy_qwen2_5_model_predict_vllm_1.py create mode 100644 openbuddy_qwen2_5_model_predict_vllm_2.py create mode 100644 openbuddy_qwen2_5_model_predict_vllm_3.py create mode 100644 qwen2_5_Instruct_model_predict_vllm_1.py create mode 100644 qwen2_5_Instruct_model_predict_vllm_2.py create mode 100644 qwen2_5_Instruct_model_predict_vllm_3.py create mode 100644 run_model_1.sh create mode 100644 run_model_4.sh create mode 100644 run_model_openbuddy_llama3_1_1.sh create mode 100644 run_model_openbuddy_llama3_1_2.sh create mode 100644 run_model_openbuddy_llama3_1_3.sh create mode 100644 run_model_openbuddy_qwen_1.sh create mode 100644 run_model_openbuddy_qwen_2.sh create mode 100644 run_model_openbuddy_qwen_3.sh create mode 100644 run_model_qwen_Instruct1.sh create mode 100644 run_model_qwen_Instruct2.sh create mode 100644 run_model_qwen_Instruct3.sh diff --git a/mistral_api.py b/mistral_api.py deleted file mode 100644 index 4708308..0000000 --- a/mistral_api.py +++ /dev/null @@ -1,131 +0,0 @@ -from flask import Flask, jsonify -from flask import request -from transformers import pipeline -import redis -import uuid -import json -from threading import Thread -from vllm import LLM, SamplingParams -import time -import threading -import time -import concurrent.futures -import requests -import socket - -app = Flask(__name__) -app.config["JSON_AS_ASCII"] = False -pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") -redis_ = redis.Redis(connection_pool=pool, decode_responses=True) - -db_key_query = 'query' -db_key_querying = 'querying' -db_key_queryset = 'queryset' -db_key_result = 'result' -db_key_error = 'error' - -def smtp_f(name): - # 在下面的代码行中使用断点来调试脚本。 - import smtplib - from email.mime.text import MIMEText - from email.header import Header - - sender = '838878981@qq.com' # 发送邮箱 - receivers = ['838878981@qq.com'] # 接收邮箱 - auth_code = "jfqtutaiwrtdbcge" # 授权码 - - message = MIMEText('基础大模型出现错误，紧急', 'plain', 'utf-8') - message['From'] = Header("Sender<%s>" % sender) # 发送者 - message['To'] = Header("Receiver<%s>" % receivers[0]) # 接收者 - - subject = name - message['Subject'] = Header(subject, 'utf-8') - - try: - server = smtplib.SMTP_SSL('smtp.qq.com', 465) - server.login(sender, auth_code) - server.sendmail(sender, receivers, message.as_string()) - print("邮件发送成功") - server.close() - except smtplib.SMTPException: - print("Error: 无法发送邮件") - - -@app.route("/predict", methods=["POST"]) -def predict(): - text = request.json["texts"] # 获取用户query中的文本 例如"I love you" - id_ = str(uuid.uuid1()) # 为query生成唯一标识 - print("uuid: ", uuid) - d = {'id': id_, 'text': text} # 绑定文本和query id - try: - load_request_path = './request_data_logs/{}.json'.format(id_) - with open(load_request_path, 'w', encoding='utf8') as f2: - # ensure_ascii=False才能输入中文，否则是Unicode字符 - # indent=2 JSON数据的缩进，美观 - json.dump(d, f2, ensure_ascii=False, indent=4) - redis_.rpush(db_key_query, json.dumps({"id": id_, "path": load_request_path})) # 加入redis - redis_.sadd(db_key_querying, id_) - redis_.sadd(db_key_queryset, id_) - return_text = {"texts": {'id': id_, }, "probabilities": None, "status_code": 200} - except: - return_text = {"texts": {'id': id_, }, "probabilities": None, "status_code": 400} - smtp_f("vllm-main-paper") - return jsonify(return_text) # 返回结果 - - -@app.route("/search", methods=["POST"]) -def search(): - id_ = request.json['id'] # 获取用户query中的文本 例如"I love you" - result = redis_.get(id_) # 获取该query的模型结果 - try: - if result is not None: - result_path = result.decode('UTF-8') - with open(result_path, encoding='utf8') as f1: - # 加载文件的对象 - result_dict = json.load(f1) - code = result_dict["status_code"] - texts = result_dict["texts"] - probabilities = result_dict["probabilities"] - if str(code) == 400: - redis_.rpush(db_key_error, json.dumps({"id": id_})) - return False - result_text = {'code': code, 'text': texts, 'probabilities': probabilities} - else: - querying_list = list(redis_.smembers(db_key_querying)) - querying_set = set() - for i in querying_list: - querying_set.add(i.decode()) - - querying_bool = False - if id_ in querying_set: - querying_bool = True - - query_list_json = redis_.lrange(db_key_query, 0, -1) - query_set_ids = set() - for i in query_list_json: - data_dict = json.loads(i) - query_id = data_dict['id'] - query_set_ids.add(query_id) - - query_bool = False - if id_ in query_set_ids: - query_bool = True - - if querying_bool == True and query_bool == True: - result_text = {'code': "201", 'text': "", 'probabilities': None} - elif querying_bool == True and query_bool == False: - result_text = {'code': "202", 'text': "", 'probabilities': None} - else: - result_text = {'code': "203", 'text': "", 'probabilities': None} - load_request_path = './request_data_logs_203/{}.json'.format(id_) - with open(load_request_path, 'w', encoding='utf8') as f2: - # ensure_ascii=False才能输入中文，否则是Unicode字符 - # indent=2 JSON数据的缩进，美观 - json.dump(result_text, f2, ensure_ascii=False, indent=4) - except: - smtp_f("vllm-main") - result_text = {'code': "400", 'text': "", 'probabilities': None} - return jsonify(result_text) # 返回结果 - -if __name__ == "__main__": - app.run(debug=False, host='0.0.0.0', port=12001) diff --git a/mistral_model_predict_vllm_1.py b/mistral_model_predict_vllm_1.py new file mode 100644 index 0000000..453a6c5 --- /dev/null +++ b/mistral_model_predict_vllm_1.py @@ -0,0 +1,106 @@ +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "1" +from transformers import pipeline +import redis +import uuid +import json +from threading import Thread +from vllm import LLM, SamplingParams +import time +import threading +import time +import concurrent.futures +import requests +import socket + + +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_result = 'result' +batch_size = 32 + +# sampling_params = SamplingParams(temperature=0.95, top_p=0.7,presence_penalty=0.9,stop="", max_tokens=4096) +sampling_params = SamplingParams(temperature=0.95, top_p=0.7,stop="", presence_penalty=1.1, max_tokens=8192) +models_path = "/home/majiahui/project/LLaMA-Factory-main/lora_openbuddy_mistral_7b_v20_3-32k_paper_model_10" +llm = LLM(model=models_path, tokenizer_mode="slow", max_model_len=8192) + + +class log: + def __init__(self): + pass + + def log(*args, **kwargs): + format = '%Y/%m/%d-%H:%M:%S' + format_h = '%Y-%m-%d' + value = time.localtime(int(time.time())) + dt = time.strftime(format, value) + dt_log_file = time.strftime(format_h, value) + log_file = 'log_file/access-%s' % dt_log_file + ".log" + if not os.path.exists(log_file): + with open(os.path.join(log_file), 'w', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + else: + with open(os.path.join(log_file), 'a+', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + + +def classify(batch_size): # 调用模型，设置最大batch_size + while True: + texts = [] + query_ids = [] + if redis_.llen(db_key_query) == 0: # 若队列中没有元素就继续获取 + time.sleep(2) + continue + + # for i in range(min(redis_.llen(db_key_query), batch_size)): + while True: + query = redis_.lpop(db_key_query) # 获取query的text + if query == None: + break + + query = query.decode('UTF-8') + data_dict_path = json.loads(query) + + path = data_dict_path['path'] + with open(path, encoding='utf8') as f1: + # 加载文件的对象 + data_dict = json.load(f1) + # query_ids.append(json.loads(query)['id']) + # texts.append(json.loads(query)['text']) # 拼接若干text 为batch + query_id = data_dict['id'] + text = data_dict["text"] + query_ids.append(query_id) + texts.append(text) + if len(texts) == batch_size: + break + outputs = llm.generate(texts, sampling_params) # 调用模型 + + generated_text_list = [""] * len(texts) + print("outputs", len(outputs)) + for i, output in enumerate(outputs): + index = output.request_id + generated_text = output.outputs[0].text + generated_text_list[int(index)] = generated_text + + for (id_, output) in zip(query_ids, generated_text_list): + + return_text = {"texts": output, "probabilities": None, "status_code": 200} + load_result_path = "./new_data_logs/{}.json".format(id_) + with open(load_result_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(return_text, f2, ensure_ascii=False, indent=4) + redis_.set(id_, load_result_path, 86400) + # redis_.set(id_, load_result_path, 30) + redis_.srem(db_key_querying, id_) + log.log('start at', + 'query_id:{},load_result_path:{},return_text:{}'.format( + id_, load_result_path, return_text)) + + +if __name__ == '__main__': + t = Thread(target=classify, args=(batch_size,)) + t.start() diff --git a/mistral_model_predict_vllm_4.py b/mistral_model_predict_vllm_4.py new file mode 100644 index 0000000..4fa7823 --- /dev/null +++ b/mistral_model_predict_vllm_4.py @@ -0,0 +1,106 @@ +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "4" +from transformers import pipeline +import redis +import uuid +import json +from threading import Thread +from vllm import LLM, SamplingParams +import time +import threading +import time +import concurrent.futures +import requests +import socket + + +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=4, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_result = 'result' +batch_size = 24 + +# sampling_params = SamplingParams(temperature=0.95, top_p=0.7,presence_penalty=0.9,stop="", max_tokens=4096) +sampling_params = SamplingParams(temperature=0.95, top_p=0.7,stop="", presence_penalty=1.1, max_tokens=4096) +models_path = "/home/majiahui/project/LLaMA-Factory-main/lora_openbuddy_zephyr_paper_model_190000" +llm = LLM(model=models_path, tokenizer_mode="slow") + + +class log: + def __init__(self): + pass + + def log(*args, **kwargs): + format = '%Y/%m/%d-%H:%M:%S' + format_h = '%Y-%m-%d' + value = time.localtime(int(time.time())) + dt = time.strftime(format, value) + dt_log_file = time.strftime(format_h, value) + log_file = 'log_file/access-%s' % dt_log_file + ".log" + if not os.path.exists(log_file): + with open(os.path.join(log_file), 'w', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + else: + with open(os.path.join(log_file), 'a+', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + + +def classify(batch_size): # 调用模型，设置最大batch_size + while True: + texts = [] + query_ids = [] + if redis_.llen(db_key_query) == 0: # 若队列中没有元素就继续获取 + time.sleep(2) + continue + + # for i in range(min(redis_.llen(db_key_query), batch_size)): + while True: + query = redis_.lpop(db_key_query) # 获取query的text + if query == None: + break + + query = query.decode('UTF-8') + data_dict_path = json.loads(query) + + path = data_dict_path['path'] + with open(path, encoding='utf8') as f1: + # 加载文件的对象 + data_dict = json.load(f1) + # query_ids.append(json.loads(query)['id']) + # texts.append(json.loads(query)['text']) # 拼接若干text 为batch + query_id = data_dict['id'] + text = data_dict["text"] + query_ids.append(query_id) + texts.append(text) + if len(texts) == batch_size: + break + outputs = llm.generate(texts, sampling_params) # 调用模型 + + generated_text_list = [""] * len(texts) + print("outputs", len(outputs)) + for i, output in enumerate(outputs): + index = output.request_id + generated_text = output.outputs[0].text + generated_text_list[int(index)] = generated_text + + for (id_, output) in zip(query_ids, generated_text_list): + + return_text = {"texts": output, "probabilities": None, "status_code": 200} + load_result_path = "./new_data_logs/{}.json".format(id_) + with open(load_result_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(return_text, f2, ensure_ascii=False, indent=4) + redis_.set(id_, load_result_path, 300) + # redis_.set(id_, load_result_path, 30) + redis_.srem(db_key_querying, id_) + log.log('start at', + 'query_id:{},load_result_path:{},return_text:{}'.format( + id_, load_result_path, return_text)) + + +if __name__ == '__main__': + t = Thread(target=classify, args=(batch_size,)) + t.start() diff --git a/model_api.py b/model_api.py new file mode 100644 index 0000000..b1f4359 --- /dev/null +++ b/model_api.py @@ -0,0 +1,135 @@ +from flask import Flask, jsonify +from flask import request +from transformers import pipeline +import redis +import uuid +import json +from threading import Thread +from vllm import LLM, SamplingParams +import time +import threading +import time +import concurrent.futures +import requests +import socket + +app = Flask(__name__) +app.config["JSON_AS_ASCII"] = False +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_queryset = 'queryset' +db_key_result = 'result' +db_key_error = 'error' + +def smtp_f(name): + # 在下面的代码行中使用断点来调试脚本。 + import smtplib + from email.mime.text import MIMEText + from email.header import Header + + sender = '838878981@qq.com' # 发送邮箱 + receivers = ['838878981@qq.com'] # 接收邮箱 + auth_code = "jfqtutaiwrtdbcge" # 授权码 + + message = MIMEText('基础大模型出现错误，紧急', 'plain', 'utf-8') + message['From'] = Header("Sender<%s>" % sender) # 发送者 + message['To'] = Header("Receiver<%s>" % receivers[0]) # 接收者 + + subject = name + message['Subject'] = Header(subject, 'utf-8') + + try: + server = smtplib.SMTP_SSL('smtp.qq.com', 465) + server.login(sender, auth_code) + server.sendmail(sender, receivers, message.as_string()) + print("邮件发送成功") + server.close() + except smtplib.SMTPException: + print("Error: 无法发送邮件") + + +@app.route("/predict", methods=["POST"]) +def predict(): + content = request.json["content"] # 获取用户query中的文本 例如"I love you" + model = request.json["model"] + top_p = request.json["top_p"] + temperature = request.json["temperature"] + id_ = str(uuid.uuid1()) # 为query生成唯一标识 + print("uuid: ", uuid) + d = {'id': id_, 'text': content, 'model': model, 'top_p': top_p,'temperature': temperature} # 绑定文本和query id + print(d) + try: + load_request_path = './request_data_logs/{}.json'.format(id_) + with open(load_request_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(d, f2, ensure_ascii=False, indent=4) + redis_.rpush(db_key_query, json.dumps({"id": id_, "path": load_request_path})) # 加入redis + redis_.sadd(db_key_querying, id_) + redis_.sadd(db_key_queryset, id_) + return_text = {"texts": {'id': id_, }, "probabilities": None, "status_code": 200} + except: + return_text = {"texts": {'id': id_, }, "probabilities": None, "status_code": 400} + smtp_f("vllm-main-paper") + return jsonify(return_text) # 返回结果 + + +@app.route("/search", methods=["POST"]) +def search(): + id_ = request.json['id'] # 获取用户query中的文本 例如"I love you" + result = redis_.get(id_) # 获取该query的模型结果 + try: + if result is not None: + result_path = result.decode('UTF-8') + with open(result_path, encoding='utf8') as f1: + # 加载文件的对象 + result_dict = json.load(f1) + code = result_dict["status_code"] + texts = result_dict["texts"] + probabilities = result_dict["probabilities"] + if str(code) == 400: + redis_.rpush(db_key_error, json.dumps({"id": id_})) + return False + result_text = {'code': code, 'text': texts, 'probabilities': probabilities} + else: + querying_list = list(redis_.smembers(db_key_querying)) + querying_set = set() + for i in querying_list: + querying_set.add(i.decode()) + + querying_bool = False + if id_ in querying_set: + querying_bool = True + + query_list_json = redis_.lrange(db_key_query, 0, -1) + query_set_ids = set() + for i in query_list_json: + data_dict = json.loads(i) + query_id = data_dict['id'] + query_set_ids.add(query_id) + + query_bool = False + if id_ in query_set_ids: + query_bool = True + + if querying_bool == True and query_bool == True: + result_text = {'code': "201", 'text': "", 'probabilities': None} + elif querying_bool == True and query_bool == False: + result_text = {'code': "202", 'text': "", 'probabilities': None} + else: + result_text = {'code': "203", 'text': "", 'probabilities': None} + load_request_path = './request_data_logs_203/{}.json'.format(id_) + with open(load_request_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(result_text, f2, ensure_ascii=False, indent=4) + except: + smtp_f("vllm-main") + result_text = {'code': "400", 'text': "", 'probabilities': None} + return jsonify(result_text) # 返回结果 + +if __name__ == "__main__": + app.run(debug=False, host='0.0.0.0', port=12001) diff --git a/openbuddy_llama3_1_model_predict_vllm_1.py b/openbuddy_llama3_1_model_predict_vllm_1.py new file mode 100644 index 0000000..b3c85c6 --- /dev/null +++ b/openbuddy_llama3_1_model_predict_vllm_1.py @@ -0,0 +1,202 @@ +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "1" +import argparse +from typing import List, Tuple +from threading import Thread +from vllm import EngineArgs, LLMEngine, RequestOutput, SamplingParams +# from vllm.utils import FlexibleArgumentParser +from flask import Flask, jsonify +from flask import request +import redis +import time +import json + +# http接口服务 +# app = FastAPI() +app = Flask(__name__) +app.config["JSON_AS_ASCII"] = False + +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_result = 'result' +batch_size = 15 + +class log: + def __init__(self): + pass + + def log(*args, **kwargs): + format = '%Y/%m/%d-%H:%M:%S' + format_h = '%Y-%m-%d' + value = time.localtime(int(time.time())) + dt = time.strftime(format, value) + dt_log_file = time.strftime(format_h, value) + log_file = 'log_file/access-%s' % dt_log_file + ".log" + if not os.path.exists(log_file): + with open(os.path.join(log_file), 'w', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + else: + with open(os.path.join(log_file), 'a+', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + + +def initialize_engine() -> LLMEngine: + """Initialize the LLMEngine from the command line arguments.""" + # model_dir = "/home/majiahui/project/models-llm/Qwen-0_5B-Chat" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper" + model_dir = "/home/majiahui/project/models-llm/openbuddy-llama3.1-8b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper_1" + args = EngineArgs(model_dir) + args.max_num_seqs = 16 # batch最大20条样本 + args.gpu_memory_utilization = 0.8 + args.max_model_len=8192 + # 加载模型 + return LLMEngine.from_engine_args(args) + +engine = initialize_engine() + + +def create_test_prompts(prompt_texts, query_ids, sampling_params_list) -> List[Tuple[str,str, SamplingParams]]: + """Create a list of test prompts with their sampling parameters.""" + + return_list = [] + + for i,j,k in zip(prompt_texts, query_ids, sampling_params_list): + return_list.append((i, j, k)) + return return_list + + +def process_requests(engine: LLMEngine, + test_prompts: List[Tuple[str, str, SamplingParams]]): + """Continuously process a list of prompts and handle the outputs.""" + + return_list = [] + while test_prompts or engine.has_unfinished_requests(): + if test_prompts: + prompt, query_id, sampling_params = test_prompts.pop(0) + engine.add_request(str(query_id), prompt, sampling_params) + + request_outputs: List[RequestOutput] = engine.step() + + for request_output in request_outputs: + if request_output.finished: + return_list.append(request_output) + return return_list + + +def main(prompt_texts, query_ids, sampling_params_list): + """Main function that sets up and runs the prompt processing.""" + + test_prompts = create_test_prompts(prompt_texts, query_ids, sampling_params_list) + return process_requests(engine, test_prompts) + + +# chat对话接口 +# @app.route("/predict/", methods=["POST"]) +# def chat(): +# # request = 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个停止句，遇到任何一个都会停止 +# +# query = request.json['query'] +# +# +# # 构造prompt +# # prompt_text, prompt_tokens = _build_prompt(generation_config, tokenizer, query, history=history, system=system) +# +# prompt_text = f"<|im_start|>user\n{query}\n<|im_end|>\n<|im_start|>assistant\n" +# +# +# return_output = main(prompt_text, sampling_params) +# return_info = { +# "request_id": return_output.request_id, +# "text": return_output.outputs[0].text +# } +# +# return jsonify(return_info) + +def classify(batch_size): # 调用模型，设置最大batch_size + while True: + texts = [] + query_ids = [] + sampling_params_list = [] + if redis_.llen(db_key_query) == 0: # 若队列中没有元素就继续获取 + time.sleep(2) + continue + + # for i in range(min(redis_.llen(db_key_query), batch_size)): + while True: + query = redis_.lpop(db_key_query) # 获取query的text + if query == None: + break + + query = query.decode('UTF-8') + data_dict_path = json.loads(query) + + path = data_dict_path['path'] + with open(path, encoding='utf8') as f1: + # 加载文件的对象 + data_dict = json.load(f1) + # query_ids.append(json.loads(query)['id']) + # texts.append(json.loads(query)['text']) # 拼接若干text 为batch + query_id = data_dict['id'] + print("query_id", query_id) + text = data_dict["text"] + model = data_dict["model"] + top_p = data_dict["top_p"] + temperature = data_dict["temperature"] + presence_penalty = 0.8 + max_tokens = 8192 + query_ids.append(query_id) + texts.append(text) + # sampling_params = SamplingParams(temperature=0.3, top_p=0.5, stop="<|end|>", presence_penalty=1.1, max_tokens=8192) + sampling_params_list.append(SamplingParams( + temperature=temperature, + top_p=top_p, + stop="<|end|>", + presence_penalty=presence_penalty, + max_tokens=max_tokens + )) + if len(texts) == batch_size: + break + + print("texts", len(texts)) + print("query_ids", len(query_ids)) + print("sampling_params_list", len(sampling_params_list)) + outputs = main(texts, query_ids, sampling_params_list) + + print("预测完成") + generated_text_dict = {} + print("outputs", len(outputs)) + for i, output in enumerate(outputs): + index = output.request_id + print(index) + generated_text = output.outputs[0].text + generated_text_dict[index] = generated_text + + print(generated_text_dict) + for id_, output in generated_text_dict.items(): + + return_text = {"texts": output, "probabilities": None, "status_code": 200} + load_result_path = "./new_data_logs/{}.json".format(id_) + with open(load_result_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(return_text, f2, ensure_ascii=False, indent=4) + redis_.set(id_, load_result_path, 86400) + # redis_.set(id_, load_result_path, 30) + redis_.srem(db_key_querying, id_) + log.log('start at', + 'query_id:{},load_result_path:{},return_text:{}'.format( + id_, load_result_path, return_text)) + + +if __name__ == '__main__': + t = Thread(target=classify, args=(batch_size,)) + t.start() diff --git a/openbuddy_llama3_1_model_predict_vllm_2.py b/openbuddy_llama3_1_model_predict_vllm_2.py new file mode 100644 index 0000000..ae98c74 --- /dev/null +++ b/openbuddy_llama3_1_model_predict_vllm_2.py @@ -0,0 +1,202 @@ +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "2" +import argparse +from typing import List, Tuple +from threading import Thread +from vllm import EngineArgs, LLMEngine, RequestOutput, SamplingParams +# from vllm.utils import FlexibleArgumentParser +from flask import Flask, jsonify +from flask import request +import redis +import time +import json + +# http接口服务 +# app = FastAPI() +app = Flask(__name__) +app.config["JSON_AS_ASCII"] = False + +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_result = 'result' +batch_size = 15 + +class log: + def __init__(self): + pass + + def log(*args, **kwargs): + format = '%Y/%m/%d-%H:%M:%S' + format_h = '%Y-%m-%d' + value = time.localtime(int(time.time())) + dt = time.strftime(format, value) + dt_log_file = time.strftime(format_h, value) + log_file = 'log_file/access-%s' % dt_log_file + ".log" + if not os.path.exists(log_file): + with open(os.path.join(log_file), 'w', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + else: + with open(os.path.join(log_file), 'a+', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + + +def initialize_engine() -> LLMEngine: + """Initialize the LLMEngine from the command line arguments.""" + # model_dir = "/home/majiahui/project/models-llm/Qwen-0_5B-Chat" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper" + model_dir = "/home/majiahui/project/models-llm/openbuddy-llama3.1-8b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper_1" + args = EngineArgs(model_dir) + args.max_num_seqs = 16 # batch最大20条样本 + args.gpu_memory_utilization = 0.8 + args.max_model_len=8192 + # 加载模型 + return LLMEngine.from_engine_args(args) + +engine = initialize_engine() + + +def create_test_prompts(prompt_texts, query_ids, sampling_params_list) -> List[Tuple[str,str, SamplingParams]]: + """Create a list of test prompts with their sampling parameters.""" + + return_list = [] + + for i,j,k in zip(prompt_texts, query_ids, sampling_params_list): + return_list.append((i, j, k)) + return return_list + + +def process_requests(engine: LLMEngine, + test_prompts: List[Tuple[str, str, SamplingParams]]): + """Continuously process a list of prompts and handle the outputs.""" + + return_list = [] + while test_prompts or engine.has_unfinished_requests(): + if test_prompts: + prompt, query_id, sampling_params = test_prompts.pop(0) + engine.add_request(str(query_id), prompt, sampling_params) + + request_outputs: List[RequestOutput] = engine.step() + + for request_output in request_outputs: + if request_output.finished: + return_list.append(request_output) + return return_list + + +def main(prompt_texts, query_ids, sampling_params_list): + """Main function that sets up and runs the prompt processing.""" + + test_prompts = create_test_prompts(prompt_texts, query_ids, sampling_params_list) + return process_requests(engine, test_prompts) + + +# chat对话接口 +# @app.route("/predict/", methods=["POST"]) +# def chat(): +# # request = 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个停止句，遇到任何一个都会停止 +# +# query = request.json['query'] +# +# +# # 构造prompt +# # prompt_text, prompt_tokens = _build_prompt(generation_config, tokenizer, query, history=history, system=system) +# +# prompt_text = f"<|im_start|>user\n{query}\n<|im_end|>\n<|im_start|>assistant\n" +# +# +# return_output = main(prompt_text, sampling_params) +# return_info = { +# "request_id": return_output.request_id, +# "text": return_output.outputs[0].text +# } +# +# return jsonify(return_info) + +def classify(batch_size): # 调用模型，设置最大batch_size + while True: + texts = [] + query_ids = [] + sampling_params_list = [] + if redis_.llen(db_key_query) == 0: # 若队列中没有元素就继续获取 + time.sleep(2) + continue + + # for i in range(min(redis_.llen(db_key_query), batch_size)): + while True: + query = redis_.lpop(db_key_query) # 获取query的text + if query == None: + break + + query = query.decode('UTF-8') + data_dict_path = json.loads(query) + + path = data_dict_path['path'] + with open(path, encoding='utf8') as f1: + # 加载文件的对象 + data_dict = json.load(f1) + # query_ids.append(json.loads(query)['id']) + # texts.append(json.loads(query)['text']) # 拼接若干text 为batch + query_id = data_dict['id'] + print("query_id", query_id) + text = data_dict["text"] + model = data_dict["model"] + top_p = data_dict["top_p"] + temperature = data_dict["temperature"] + presence_penalty = 0.8 + max_tokens = 8192 + query_ids.append(query_id) + texts.append(text) + # sampling_params = SamplingParams(temperature=0.3, top_p=0.5, stop="<|end|>", presence_penalty=1.1, max_tokens=8192) + sampling_params_list.append(SamplingParams( + temperature=temperature, + top_p=top_p, + stop="<|end|>", + presence_penalty=presence_penalty, + max_tokens=max_tokens + )) + if len(texts) == batch_size: + break + + print("texts", len(texts)) + print("query_ids", len(query_ids)) + print("sampling_params_list", len(sampling_params_list)) + outputs = main(texts, query_ids, sampling_params_list) + + print("预测完成") + generated_text_dict = {} + print("outputs", len(outputs)) + for i, output in enumerate(outputs): + index = output.request_id + print(index) + generated_text = output.outputs[0].text + generated_text_dict[index] = generated_text + + print(generated_text_dict) + for id_, output in generated_text_dict.items(): + + return_text = {"texts": output, "probabilities": None, "status_code": 200} + load_result_path = "./new_data_logs/{}.json".format(id_) + with open(load_result_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(return_text, f2, ensure_ascii=False, indent=4) + redis_.set(id_, load_result_path, 86400) + # redis_.set(id_, load_result_path, 30) + redis_.srem(db_key_querying, id_) + log.log('start at', + 'query_id:{},load_result_path:{},return_text:{}'.format( + id_, load_result_path, return_text)) + + +if __name__ == '__main__': + t = Thread(target=classify, args=(batch_size,)) + t.start() diff --git a/openbuddy_llama3_1_model_predict_vllm_3.py b/openbuddy_llama3_1_model_predict_vllm_3.py new file mode 100644 index 0000000..ffe17fe --- /dev/null +++ b/openbuddy_llama3_1_model_predict_vllm_3.py @@ -0,0 +1,202 @@ +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "3" +import argparse +from typing import List, Tuple +from threading import Thread +from vllm import EngineArgs, LLMEngine, RequestOutput, SamplingParams +# from vllm.utils import FlexibleArgumentParser +from flask import Flask, jsonify +from flask import request +import redis +import time +import json + +# http接口服务 +# app = FastAPI() +app = Flask(__name__) +app.config["JSON_AS_ASCII"] = False + +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_result = 'result' +batch_size = 15 + +class log: + def __init__(self): + pass + + def log(*args, **kwargs): + format = '%Y/%m/%d-%H:%M:%S' + format_h = '%Y-%m-%d' + value = time.localtime(int(time.time())) + dt = time.strftime(format, value) + dt_log_file = time.strftime(format_h, value) + log_file = 'log_file/access-%s' % dt_log_file + ".log" + if not os.path.exists(log_file): + with open(os.path.join(log_file), 'w', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + else: + with open(os.path.join(log_file), 'a+', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + + +def initialize_engine() -> LLMEngine: + """Initialize the LLMEngine from the command line arguments.""" + # model_dir = "/home/majiahui/project/models-llm/Qwen-0_5B-Chat" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper" + model_dir = "/home/majiahui/project/models-llm/openbuddy-llama3.1-8b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper_1" + args = EngineArgs(model_dir) + args.max_num_seqs = 16 # batch最大20条样本 + args.gpu_memory_utilization = 0.8 + args.max_model_len=8192 + # 加载模型 + return LLMEngine.from_engine_args(args) + +engine = initialize_engine() + + +def create_test_prompts(prompt_texts, query_ids, sampling_params_list) -> List[Tuple[str,str, SamplingParams]]: + """Create a list of test prompts with their sampling parameters.""" + + return_list = [] + + for i,j,k in zip(prompt_texts, query_ids, sampling_params_list): + return_list.append((i, j, k)) + return return_list + + +def process_requests(engine: LLMEngine, + test_prompts: List[Tuple[str, str, SamplingParams]]): + """Continuously process a list of prompts and handle the outputs.""" + + return_list = [] + while test_prompts or engine.has_unfinished_requests(): + if test_prompts: + prompt, query_id, sampling_params = test_prompts.pop(0) + engine.add_request(str(query_id), prompt, sampling_params) + + request_outputs: List[RequestOutput] = engine.step() + + for request_output in request_outputs: + if request_output.finished: + return_list.append(request_output) + return return_list + + +def main(prompt_texts, query_ids, sampling_params_list): + """Main function that sets up and runs the prompt processing.""" + + test_prompts = create_test_prompts(prompt_texts, query_ids, sampling_params_list) + return process_requests(engine, test_prompts) + + +# chat对话接口 +# @app.route("/predict/", methods=["POST"]) +# def chat(): +# # request = 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个停止句，遇到任何一个都会停止 +# +# query = request.json['query'] +# +# +# # 构造prompt +# # prompt_text, prompt_tokens = _build_prompt(generation_config, tokenizer, query, history=history, system=system) +# +# prompt_text = f"<|im_start|>user\n{query}\n<|im_end|>\n<|im_start|>assistant\n" +# +# +# return_output = main(prompt_text, sampling_params) +# return_info = { +# "request_id": return_output.request_id, +# "text": return_output.outputs[0].text +# } +# +# return jsonify(return_info) + +def classify(batch_size): # 调用模型，设置最大batch_size + while True: + texts = [] + query_ids = [] + sampling_params_list = [] + if redis_.llen(db_key_query) == 0: # 若队列中没有元素就继续获取 + time.sleep(2) + continue + + # for i in range(min(redis_.llen(db_key_query), batch_size)): + while True: + query = redis_.lpop(db_key_query) # 获取query的text + if query == None: + break + + query = query.decode('UTF-8') + data_dict_path = json.loads(query) + + path = data_dict_path['path'] + with open(path, encoding='utf8') as f1: + # 加载文件的对象 + data_dict = json.load(f1) + # query_ids.append(json.loads(query)['id']) + # texts.append(json.loads(query)['text']) # 拼接若干text 为batch + query_id = data_dict['id'] + print("query_id", query_id) + text = data_dict["text"] + model = data_dict["model"] + top_p = data_dict["top_p"] + temperature = data_dict["temperature"] + presence_penalty = 0.8 + max_tokens = 8192 + query_ids.append(query_id) + texts.append(text) + # sampling_params = SamplingParams(temperature=0.3, top_p=0.5, stop="<|end|>", presence_penalty=1.1, max_tokens=8192) + sampling_params_list.append(SamplingParams( + temperature=temperature, + top_p=top_p, + stop="<|end|>", + presence_penalty=presence_penalty, + max_tokens=max_tokens + )) + if len(texts) == batch_size: + break + + print("texts", len(texts)) + print("query_ids", len(query_ids)) + print("sampling_params_list", len(sampling_params_list)) + outputs = main(texts, query_ids, sampling_params_list) + + print("预测完成") + generated_text_dict = {} + print("outputs", len(outputs)) + for i, output in enumerate(outputs): + index = output.request_id + print(index) + generated_text = output.outputs[0].text + generated_text_dict[index] = generated_text + + print(generated_text_dict) + for id_, output in generated_text_dict.items(): + + return_text = {"texts": output, "probabilities": None, "status_code": 200} + load_result_path = "./new_data_logs/{}.json".format(id_) + with open(load_result_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(return_text, f2, ensure_ascii=False, indent=4) + redis_.set(id_, load_result_path, 86400) + # redis_.set(id_, load_result_path, 30) + redis_.srem(db_key_querying, id_) + log.log('start at', + 'query_id:{},load_result_path:{},return_text:{}'.format( + id_, load_result_path, return_text)) + + +if __name__ == '__main__': + t = Thread(target=classify, args=(batch_size,)) + t.start() diff --git a/openbuddy_qwen2_5_model_predict_vllm_1.py b/openbuddy_qwen2_5_model_predict_vllm_1.py new file mode 100644 index 0000000..944f07b --- /dev/null +++ b/openbuddy_qwen2_5_model_predict_vllm_1.py @@ -0,0 +1,202 @@ +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "1" +import argparse +from typing import List, Tuple +from threading import Thread +from vllm import EngineArgs, LLMEngine, RequestOutput, SamplingParams +# from vllm.utils import FlexibleArgumentParser +from flask import Flask, jsonify +from flask import request +import redis +import time +import json + +# http接口服务 +# app = FastAPI() +app = Flask(__name__) +app.config["JSON_AS_ASCII"] = False + +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_result = 'result' +batch_size = 15 + +class log: + def __init__(self): + pass + + def log(*args, **kwargs): + format = '%Y/%m/%d-%H:%M:%S' + format_h = '%Y-%m-%d' + value = time.localtime(int(time.time())) + dt = time.strftime(format, value) + dt_log_file = time.strftime(format_h, value) + log_file = 'log_file/access-%s' % dt_log_file + ".log" + if not os.path.exists(log_file): + with open(os.path.join(log_file), 'w', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + else: + with open(os.path.join(log_file), 'a+', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + + +def initialize_engine() -> LLMEngine: + """Initialize the LLMEngine from the command line arguments.""" + # model_dir = "/home/majiahui/project/models-llm/Qwen-0_5B-Chat" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper" + model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper_2" + args = EngineArgs(model_dir) + args.max_num_seqs = 16 # batch最大20条样本 + args.gpu_memory_utilization = 0.8 + args.max_model_len=8192 + # 加载模型 + return LLMEngine.from_engine_args(args) + +engine = initialize_engine() + + +def create_test_prompts(prompt_texts, query_ids, sampling_params_list) -> List[Tuple[str,str, SamplingParams]]: + """Create a list of test prompts with their sampling parameters.""" + + return_list = [] + + for i,j,k in zip(prompt_texts, query_ids, sampling_params_list): + return_list.append((i, j, k)) + return return_list + + +def process_requests(engine: LLMEngine, + test_prompts: List[Tuple[str, str, SamplingParams]]): + """Continuously process a list of prompts and handle the outputs.""" + + return_list = [] + while test_prompts or engine.has_unfinished_requests(): + if test_prompts: + prompt, query_id, sampling_params = test_prompts.pop(0) + engine.add_request(str(query_id), prompt, sampling_params) + + request_outputs: List[RequestOutput] = engine.step() + + for request_output in request_outputs: + if request_output.finished: + return_list.append(request_output) + return return_list + + +def main(prompt_texts, query_ids, sampling_params_list): + """Main function that sets up and runs the prompt processing.""" + + test_prompts = create_test_prompts(prompt_texts, query_ids, sampling_params_list) + return process_requests(engine, test_prompts) + + +# chat对话接口 +# @app.route("/predict/", methods=["POST"]) +# def chat(): +# # request = 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个停止句，遇到任何一个都会停止 +# +# query = request.json['query'] +# +# +# # 构造prompt +# # prompt_text, prompt_tokens = _build_prompt(generation_config, tokenizer, query, history=history, system=system) +# +# prompt_text = f"<|im_start|>user\n{query}\n<|im_end|>\n<|im_start|>assistant\n" +# +# +# return_output = main(prompt_text, sampling_params) +# return_info = { +# "request_id": return_output.request_id, +# "text": return_output.outputs[0].text +# } +# +# return jsonify(return_info) + +def classify(batch_size): # 调用模型，设置最大batch_size + while True: + texts = [] + query_ids = [] + sampling_params_list = [] + if redis_.llen(db_key_query) == 0: # 若队列中没有元素就继续获取 + time.sleep(2) + continue + + # for i in range(min(redis_.llen(db_key_query), batch_size)): + while True: + query = redis_.lpop(db_key_query) # 获取query的text + if query == None: + break + + query = query.decode('UTF-8') + data_dict_path = json.loads(query) + + path = data_dict_path['path'] + with open(path, encoding='utf8') as f1: + # 加载文件的对象 + data_dict = json.load(f1) + # query_ids.append(json.loads(query)['id']) + # texts.append(json.loads(query)['text']) # 拼接若干text 为batch + query_id = data_dict['id'] + print("query_id", query_id) + text = data_dict["text"] + model = data_dict["model"] + top_p = data_dict["top_p"] + temperature = data_dict["temperature"] + presence_penalty = 0.8 + max_tokens = 8192 + query_ids.append(query_id) + texts.append(text) + # sampling_params = SamplingParams(temperature=0.3, top_p=0.5, stop="<|end|>", presence_penalty=1.1, max_tokens=8192) + sampling_params_list.append(SamplingParams( + temperature=temperature, + top_p=top_p, + stop="<|end|>", + presence_penalty=presence_penalty, + max_tokens=max_tokens + )) + if len(texts) == batch_size: + break + + print("texts", len(texts)) + print("query_ids", len(query_ids)) + print("sampling_params_list", len(sampling_params_list)) + outputs = main(texts, query_ids, sampling_params_list) + + print("预测完成") + generated_text_dict = {} + print("outputs", len(outputs)) + for i, output in enumerate(outputs): + index = output.request_id + print(index) + generated_text = output.outputs[0].text + generated_text_dict[index] = generated_text + + print(generated_text_dict) + for id_, output in generated_text_dict.items(): + + return_text = {"texts": output, "probabilities": None, "status_code": 200} + load_result_path = "./new_data_logs/{}.json".format(id_) + with open(load_result_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(return_text, f2, ensure_ascii=False, indent=4) + redis_.set(id_, load_result_path, 86400) + # redis_.set(id_, load_result_path, 30) + redis_.srem(db_key_querying, id_) + log.log('start at', + 'query_id:{},load_result_path:{},return_text:{}'.format( + id_, load_result_path, return_text)) + + +if __name__ == '__main__': + t = Thread(target=classify, args=(batch_size,)) + t.start() diff --git a/openbuddy_qwen2_5_model_predict_vllm_2.py b/openbuddy_qwen2_5_model_predict_vllm_2.py new file mode 100644 index 0000000..465a753 --- /dev/null +++ b/openbuddy_qwen2_5_model_predict_vllm_2.py @@ -0,0 +1,202 @@ +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "2" +import argparse +from typing import List, Tuple +from threading import Thread +from vllm import EngineArgs, LLMEngine, RequestOutput, SamplingParams +# from vllm.utils import FlexibleArgumentParser +from flask import Flask, jsonify +from flask import request +import redis +import time +import json + +# http接口服务 +# app = FastAPI() +app = Flask(__name__) +app.config["JSON_AS_ASCII"] = False + +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_result = 'result' +batch_size = 15 + +class log: + def __init__(self): + pass + + def log(*args, **kwargs): + format = '%Y/%m/%d-%H:%M:%S' + format_h = '%Y-%m-%d' + value = time.localtime(int(time.time())) + dt = time.strftime(format, value) + dt_log_file = time.strftime(format_h, value) + log_file = 'log_file/access-%s' % dt_log_file + ".log" + if not os.path.exists(log_file): + with open(os.path.join(log_file), 'w', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + else: + with open(os.path.join(log_file), 'a+', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + + +def initialize_engine() -> LLMEngine: + """Initialize the LLMEngine from the command line arguments.""" + # model_dir = "/home/majiahui/project/models-llm/Qwen-0_5B-Chat" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper" + model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper_2" + args = EngineArgs(model_dir) + args.max_num_seqs = 16 # batch最大20条样本 + args.gpu_memory_utilization = 0.8 + args.max_model_len=8192 + # 加载模型 + return LLMEngine.from_engine_args(args) + +engine = initialize_engine() + + +def create_test_prompts(prompt_texts, query_ids, sampling_params_list) -> List[Tuple[str,str, SamplingParams]]: + """Create a list of test prompts with their sampling parameters.""" + + return_list = [] + + for i,j,k in zip(prompt_texts, query_ids, sampling_params_list): + return_list.append((i, j, k)) + return return_list + + +def process_requests(engine: LLMEngine, + test_prompts: List[Tuple[str, str, SamplingParams]]): + """Continuously process a list of prompts and handle the outputs.""" + + return_list = [] + while test_prompts or engine.has_unfinished_requests(): + if test_prompts: + prompt, query_id, sampling_params = test_prompts.pop(0) + engine.add_request(str(query_id), prompt, sampling_params) + + request_outputs: List[RequestOutput] = engine.step() + + for request_output in request_outputs: + if request_output.finished: + return_list.append(request_output) + return return_list + + +def main(prompt_texts, query_ids, sampling_params_list): + """Main function that sets up and runs the prompt processing.""" + + test_prompts = create_test_prompts(prompt_texts, query_ids, sampling_params_list) + return process_requests(engine, test_prompts) + + +# chat对话接口 +# @app.route("/predict/", methods=["POST"]) +# def chat(): +# # request = 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个停止句，遇到任何一个都会停止 +# +# query = request.json['query'] +# +# +# # 构造prompt +# # prompt_text, prompt_tokens = _build_prompt(generation_config, tokenizer, query, history=history, system=system) +# +# prompt_text = f"<|im_start|>user\n{query}\n<|im_end|>\n<|im_start|>assistant\n" +# +# +# return_output = main(prompt_text, sampling_params) +# return_info = { +# "request_id": return_output.request_id, +# "text": return_output.outputs[0].text +# } +# +# return jsonify(return_info) + +def classify(batch_size): # 调用模型，设置最大batch_size + while True: + texts = [] + query_ids = [] + sampling_params_list = [] + if redis_.llen(db_key_query) == 0: # 若队列中没有元素就继续获取 + time.sleep(2) + continue + + # for i in range(min(redis_.llen(db_key_query), batch_size)): + while True: + query = redis_.lpop(db_key_query) # 获取query的text + if query == None: + break + + query = query.decode('UTF-8') + data_dict_path = json.loads(query) + + path = data_dict_path['path'] + with open(path, encoding='utf8') as f1: + # 加载文件的对象 + data_dict = json.load(f1) + # query_ids.append(json.loads(query)['id']) + # texts.append(json.loads(query)['text']) # 拼接若干text 为batch + query_id = data_dict['id'] + print("query_id", query_id) + text = data_dict["text"] + model = data_dict["model"] + top_p = data_dict["top_p"] + temperature = data_dict["temperature"] + presence_penalty = 0.8 + max_tokens = 8192 + query_ids.append(query_id) + texts.append(text) + # sampling_params = SamplingParams(temperature=0.3, top_p=0.5, stop="<|end|>", presence_penalty=1.1, max_tokens=8192) + sampling_params_list.append(SamplingParams( + temperature=temperature, + top_p=top_p, + stop="<|end|>", + presence_penalty=presence_penalty, + max_tokens=max_tokens + )) + if len(texts) == batch_size: + break + + print("texts", len(texts)) + print("query_ids", len(query_ids)) + print("sampling_params_list", len(sampling_params_list)) + outputs = main(texts, query_ids, sampling_params_list) + + print("预测完成") + generated_text_dict = {} + print("outputs", len(outputs)) + for i, output in enumerate(outputs): + index = output.request_id + print(index) + generated_text = output.outputs[0].text + generated_text_dict[index] = generated_text + + print(generated_text_dict) + for id_, output in generated_text_dict.items(): + + return_text = {"texts": output, "probabilities": None, "status_code": 200} + load_result_path = "./new_data_logs/{}.json".format(id_) + with open(load_result_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(return_text, f2, ensure_ascii=False, indent=4) + redis_.set(id_, load_result_path, 86400) + # redis_.set(id_, load_result_path, 30) + redis_.srem(db_key_querying, id_) + log.log('start at', + 'query_id:{},load_result_path:{},return_text:{}'.format( + id_, load_result_path, return_text)) + + +if __name__ == '__main__': + t = Thread(target=classify, args=(batch_size,)) + t.start() diff --git a/openbuddy_qwen2_5_model_predict_vllm_3.py b/openbuddy_qwen2_5_model_predict_vllm_3.py new file mode 100644 index 0000000..078c82f --- /dev/null +++ b/openbuddy_qwen2_5_model_predict_vllm_3.py @@ -0,0 +1,202 @@ +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "3" +import argparse +from typing import List, Tuple +from threading import Thread +from vllm import EngineArgs, LLMEngine, RequestOutput, SamplingParams +# from vllm.utils import FlexibleArgumentParser +from flask import Flask, jsonify +from flask import request +import redis +import time +import json + +# http接口服务 +# app = FastAPI() +app = Flask(__name__) +app.config["JSON_AS_ASCII"] = False + +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_result = 'result' +batch_size = 15 + +class log: + def __init__(self): + pass + + def log(*args, **kwargs): + format = '%Y/%m/%d-%H:%M:%S' + format_h = '%Y-%m-%d' + value = time.localtime(int(time.time())) + dt = time.strftime(format, value) + dt_log_file = time.strftime(format_h, value) + log_file = 'log_file/access-%s' % dt_log_file + ".log" + if not os.path.exists(log_file): + with open(os.path.join(log_file), 'w', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + else: + with open(os.path.join(log_file), 'a+', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + + +def initialize_engine() -> LLMEngine: + """Initialize the LLMEngine from the command line arguments.""" + # model_dir = "/home/majiahui/project/models-llm/Qwen-0_5B-Chat" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper" + model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper_2" + args = EngineArgs(model_dir) + args.max_num_seqs = 16 # batch最大20条样本 + args.gpu_memory_utilization = 0.8 + args.max_model_len=8192 + # 加载模型 + return LLMEngine.from_engine_args(args) + +engine = initialize_engine() + + +def create_test_prompts(prompt_texts, query_ids, sampling_params_list) -> List[Tuple[str,str, SamplingParams]]: + """Create a list of test prompts with their sampling parameters.""" + + return_list = [] + + for i,j,k in zip(prompt_texts, query_ids, sampling_params_list): + return_list.append((i, j, k)) + return return_list + + +def process_requests(engine: LLMEngine, + test_prompts: List[Tuple[str, str, SamplingParams]]): + """Continuously process a list of prompts and handle the outputs.""" + + return_list = [] + while test_prompts or engine.has_unfinished_requests(): + if test_prompts: + prompt, query_id, sampling_params = test_prompts.pop(0) + engine.add_request(str(query_id), prompt, sampling_params) + + request_outputs: List[RequestOutput] = engine.step() + + for request_output in request_outputs: + if request_output.finished: + return_list.append(request_output) + return return_list + + +def main(prompt_texts, query_ids, sampling_params_list): + """Main function that sets up and runs the prompt processing.""" + + test_prompts = create_test_prompts(prompt_texts, query_ids, sampling_params_list) + return process_requests(engine, test_prompts) + + +# chat对话接口 +# @app.route("/predict/", methods=["POST"]) +# def chat(): +# # request = 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个停止句，遇到任何一个都会停止 +# +# query = request.json['query'] +# +# +# # 构造prompt +# # prompt_text, prompt_tokens = _build_prompt(generation_config, tokenizer, query, history=history, system=system) +# +# prompt_text = f"<|im_start|>user\n{query}\n<|im_end|>\n<|im_start|>assistant\n" +# +# +# return_output = main(prompt_text, sampling_params) +# return_info = { +# "request_id": return_output.request_id, +# "text": return_output.outputs[0].text +# } +# +# return jsonify(return_info) + +def classify(batch_size): # 调用模型，设置最大batch_size + while True: + texts = [] + query_ids = [] + sampling_params_list = [] + if redis_.llen(db_key_query) == 0: # 若队列中没有元素就继续获取 + time.sleep(2) + continue + + # for i in range(min(redis_.llen(db_key_query), batch_size)): + while True: + query = redis_.lpop(db_key_query) # 获取query的text + if query == None: + break + + query = query.decode('UTF-8') + data_dict_path = json.loads(query) + + path = data_dict_path['path'] + with open(path, encoding='utf8') as f1: + # 加载文件的对象 + data_dict = json.load(f1) + # query_ids.append(json.loads(query)['id']) + # texts.append(json.loads(query)['text']) # 拼接若干text 为batch + query_id = data_dict['id'] + print("query_id", query_id) + text = data_dict["text"] + model = data_dict["model"] + top_p = data_dict["top_p"] + temperature = data_dict["temperature"] + presence_penalty = 0.8 + max_tokens = 8192 + query_ids.append(query_id) + texts.append(text) + # sampling_params = SamplingParams(temperature=0.3, top_p=0.5, stop="<|end|>", presence_penalty=1.1, max_tokens=8192) + sampling_params_list.append(SamplingParams( + temperature=temperature, + top_p=top_p, + stop="<|end|>", + presence_penalty=presence_penalty, + max_tokens=max_tokens + )) + if len(texts) == batch_size: + break + + print("texts", len(texts)) + print("query_ids", len(query_ids)) + print("sampling_params_list", len(sampling_params_list)) + outputs = main(texts, query_ids, sampling_params_list) + + print("预测完成") + generated_text_dict = {} + print("outputs", len(outputs)) + for i, output in enumerate(outputs): + index = output.request_id + print(index) + generated_text = output.outputs[0].text + generated_text_dict[index] = generated_text + + print(generated_text_dict) + for id_, output in generated_text_dict.items(): + + return_text = {"texts": output, "probabilities": None, "status_code": 200} + load_result_path = "./new_data_logs/{}.json".format(id_) + with open(load_result_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(return_text, f2, ensure_ascii=False, indent=4) + redis_.set(id_, load_result_path, 86400) + # redis_.set(id_, load_result_path, 30) + redis_.srem(db_key_querying, id_) + log.log('start at', + 'query_id:{},load_result_path:{},return_text:{}'.format( + id_, load_result_path, return_text)) + + +if __name__ == '__main__': + t = Thread(target=classify, args=(batch_size,)) + t.start() diff --git a/qwen2_5_Instruct_model_predict_vllm_1.py b/qwen2_5_Instruct_model_predict_vllm_1.py new file mode 100644 index 0000000..5d26fbd --- /dev/null +++ b/qwen2_5_Instruct_model_predict_vllm_1.py @@ -0,0 +1,205 @@ +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "1" +import argparse +from typing import List, Tuple +from threading import Thread +from vllm import EngineArgs, LLMEngine, RequestOutput, SamplingParams +# from vllm.utils import FlexibleArgumentParser +from flask import Flask, jsonify +from flask import request +import redis +import time +import json + +# http接口服务 +# app = FastAPI() +app = Flask(__name__) +app.config["JSON_AS_ASCII"] = False + +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_result = 'result' +batch_size = 15 + +class log: + def __init__(self): + pass + + def log(*args, **kwargs): + format = '%Y/%m/%d-%H:%M:%S' + format_h = '%Y-%m-%d' + value = time.localtime(int(time.time())) + dt = time.strftime(format, value) + dt_log_file = time.strftime(format_h, value) + log_file = 'log_file/access-%s' % dt_log_file + ".log" + if not os.path.exists(log_file): + with open(os.path.join(log_file), 'w', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + else: + with open(os.path.join(log_file), 'a+', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + + +def initialize_engine() -> LLMEngine: + """Initialize the LLMEngine from the command line arguments.""" + # model_dir = "/home/majiahui/project/models-llm/Qwen-0_5B-Chat" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper_2" + # model_dir = "/home/majiahui/project/models-llm/Qwen2.5-7B-Instruct-1M" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b-v23.1-200k" + model_dir = "/home/majiahui/project/models-llm/qwen2_5_7B_train_11_prompt_4_gpt_xiaobiaot_real_paper_1" + args = EngineArgs(model_dir) + args.max_num_seqs = 16 # batch最大20条样本 + args.gpu_memory_utilization = 0.8 + args.max_model_len=8192 + # 加载模型 + return LLMEngine.from_engine_args(args) + +engine = initialize_engine() + + +def create_test_prompts(prompt_texts, query_ids, sampling_params_list) -> List[Tuple[str,str, SamplingParams]]: + """Create a list of test prompts with their sampling parameters.""" + + return_list = [] + + for i,j,k in zip(prompt_texts, query_ids, sampling_params_list): + return_list.append((i, j, k)) + return return_list + + +def process_requests(engine: LLMEngine, + test_prompts: List[Tuple[str, str, SamplingParams]]): + """Continuously process a list of prompts and handle the outputs.""" + + return_list = [] + while test_prompts or engine.has_unfinished_requests(): + if test_prompts: + prompt, query_id, sampling_params = test_prompts.pop(0) + engine.add_request(str(query_id), prompt, sampling_params) + + request_outputs: List[RequestOutput] = engine.step() + + for request_output in request_outputs: + if request_output.finished: + return_list.append(request_output) + return return_list + + +def main(prompt_texts, query_ids, sampling_params_list): + """Main function that sets up and runs the prompt processing.""" + + test_prompts = create_test_prompts(prompt_texts, query_ids, sampling_params_list) + return process_requests(engine, test_prompts) + + +# chat对话接口 +# @app.route("/predict/", methods=["POST"]) +# def chat(): +# # request = 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个停止句，遇到任何一个都会停止 +# +# query = request.json['query'] +# +# +# # 构造prompt +# # prompt_text, prompt_tokens = _build_prompt(generation_config, tokenizer, query, history=history, system=system) +# +# prompt_text = f"<|im_start|>user\n{query}\n<|im_end|>\n<|im_start|>assistant\n" +# +# +# return_output = main(prompt_text, sampling_params) +# return_info = { +# "request_id": return_output.request_id, +# "text": return_output.outputs[0].text +# } +# +# return jsonify(return_info) + +def classify(batch_size): # 调用模型，设置最大batch_size + while True: + texts = [] + query_ids = [] + sampling_params_list = [] + if redis_.llen(db_key_query) == 0: # 若队列中没有元素就继续获取 + time.sleep(2) + continue + + # for i in range(min(redis_.llen(db_key_query), batch_size)): + while True: + query = redis_.lpop(db_key_query) # 获取query的text + if query == None: + break + + query = query.decode('UTF-8') + data_dict_path = json.loads(query) + + path = data_dict_path['path'] + with open(path, encoding='utf8') as f1: + # 加载文件的对象 + data_dict = json.load(f1) + # query_ids.append(json.loads(query)['id']) + # texts.append(json.loads(query)['text']) # 拼接若干text 为batch + query_id = data_dict['id'] + print("query_id", query_id) + text = data_dict["text"] + model = data_dict["model"] + top_p = data_dict["top_p"] + temperature = data_dict["temperature"] + presence_penalty = 1.1 + max_tokens = 8192 + query_ids.append(query_id) + texts.append(text) + # sampling_params = SamplingParams(temperature=0.3, top_p=0.5, stop="<|end|>", presence_penalty=1.1, max_tokens=8192) + sampling_params_list.append(SamplingParams( + temperature=temperature, + top_p=top_p, + stop="<|end|>", + presence_penalty=presence_penalty, + max_tokens=max_tokens + )) + if len(texts) == batch_size: + break + + print("texts", len(texts)) + print("query_ids", len(query_ids)) + print("sampling_params_list", len(sampling_params_list)) + outputs = main(texts, query_ids, sampling_params_list) + + print("预测完成") + generated_text_dict = {} + print("outputs", len(outputs)) + for i, output in enumerate(outputs): + index = output.request_id + print(index) + generated_text = output.outputs[0].text + generated_text_dict[index] = generated_text + + print(generated_text_dict) + for id_, output in generated_text_dict.items(): + + return_text = {"texts": output, "probabilities": None, "status_code": 200} + load_result_path = "./new_data_logs/{}.json".format(id_) + with open(load_result_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(return_text, f2, ensure_ascii=False, indent=4) + redis_.set(id_, load_result_path, 86400) + # redis_.set(id_, load_result_path, 30) + redis_.srem(db_key_querying, id_) + log.log('start at', + 'query_id:{},load_result_path:{},return_text:{}'.format( + id_, load_result_path, return_text)) + + +if __name__ == '__main__': + t = Thread(target=classify, args=(batch_size,)) + t.start() diff --git a/qwen2_5_Instruct_model_predict_vllm_2.py b/qwen2_5_Instruct_model_predict_vllm_2.py new file mode 100644 index 0000000..65279a2 --- /dev/null +++ b/qwen2_5_Instruct_model_predict_vllm_2.py @@ -0,0 +1,205 @@ +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "2" +import argparse +from typing import List, Tuple +from threading import Thread +from vllm import EngineArgs, LLMEngine, RequestOutput, SamplingParams +# from vllm.utils import FlexibleArgumentParser +from flask import Flask, jsonify +from flask import request +import redis +import time +import json + +# http接口服务 +# app = FastAPI() +app = Flask(__name__) +app.config["JSON_AS_ASCII"] = False + +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_result = 'result' +batch_size = 15 + +class log: + def __init__(self): + pass + + def log(*args, **kwargs): + format = '%Y/%m/%d-%H:%M:%S' + format_h = '%Y-%m-%d' + value = time.localtime(int(time.time())) + dt = time.strftime(format, value) + dt_log_file = time.strftime(format_h, value) + log_file = 'log_file/access-%s' % dt_log_file + ".log" + if not os.path.exists(log_file): + with open(os.path.join(log_file), 'w', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + else: + with open(os.path.join(log_file), 'a+', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + + +def initialize_engine() -> LLMEngine: + """Initialize the LLMEngine from the command line arguments.""" + # model_dir = "/home/majiahui/project/models-llm/Qwen-0_5B-Chat" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper_2" + # model_dir = "/home/majiahui/project/models-llm/Qwen2.5-7B-Instruct-1M" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b-v23.1-200k" + model_dir = "/home/majiahui/project/models-llm/qwen2_5_7B_train_11_prompt_4_gpt_xiaobiaot_real_paper_1" + args = EngineArgs(model_dir) + args.max_num_seqs = 16 # batch最大20条样本 + args.gpu_memory_utilization = 0.8 + args.max_model_len=8192 + # 加载模型 + return LLMEngine.from_engine_args(args) + +engine = initialize_engine() + + +def create_test_prompts(prompt_texts, query_ids, sampling_params_list) -> List[Tuple[str,str, SamplingParams]]: + """Create a list of test prompts with their sampling parameters.""" + + return_list = [] + + for i,j,k in zip(prompt_texts, query_ids, sampling_params_list): + return_list.append((i, j, k)) + return return_list + + +def process_requests(engine: LLMEngine, + test_prompts: List[Tuple[str, str, SamplingParams]]): + """Continuously process a list of prompts and handle the outputs.""" + + return_list = [] + while test_prompts or engine.has_unfinished_requests(): + if test_prompts: + prompt, query_id, sampling_params = test_prompts.pop(0) + engine.add_request(str(query_id), prompt, sampling_params) + + request_outputs: List[RequestOutput] = engine.step() + + for request_output in request_outputs: + if request_output.finished: + return_list.append(request_output) + return return_list + + +def main(prompt_texts, query_ids, sampling_params_list): + """Main function that sets up and runs the prompt processing.""" + + test_prompts = create_test_prompts(prompt_texts, query_ids, sampling_params_list) + return process_requests(engine, test_prompts) + + +# chat对话接口 +# @app.route("/predict/", methods=["POST"]) +# def chat(): +# # request = 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个停止句，遇到任何一个都会停止 +# +# query = request.json['query'] +# +# +# # 构造prompt +# # prompt_text, prompt_tokens = _build_prompt(generation_config, tokenizer, query, history=history, system=system) +# +# prompt_text = f"<|im_start|>user\n{query}\n<|im_end|>\n<|im_start|>assistant\n" +# +# +# return_output = main(prompt_text, sampling_params) +# return_info = { +# "request_id": return_output.request_id, +# "text": return_output.outputs[0].text +# } +# +# return jsonify(return_info) + +def classify(batch_size): # 调用模型，设置最大batch_size + while True: + texts = [] + query_ids = [] + sampling_params_list = [] + if redis_.llen(db_key_query) == 0: # 若队列中没有元素就继续获取 + time.sleep(2) + continue + + # for i in range(min(redis_.llen(db_key_query), batch_size)): + while True: + query = redis_.lpop(db_key_query) # 获取query的text + if query == None: + break + + query = query.decode('UTF-8') + data_dict_path = json.loads(query) + + path = data_dict_path['path'] + with open(path, encoding='utf8') as f1: + # 加载文件的对象 + data_dict = json.load(f1) + # query_ids.append(json.loads(query)['id']) + # texts.append(json.loads(query)['text']) # 拼接若干text 为batch + query_id = data_dict['id'] + print("query_id", query_id) + text = data_dict["text"] + model = data_dict["model"] + top_p = data_dict["top_p"] + temperature = data_dict["temperature"] + presence_penalty = 1.1 + max_tokens = 8192 + query_ids.append(query_id) + texts.append(text) + # sampling_params = SamplingParams(temperature=0.3, top_p=0.5, stop="<|end|>", presence_penalty=1.1, max_tokens=8192) + sampling_params_list.append(SamplingParams( + temperature=temperature, + top_p=top_p, + stop="<|end|>", + presence_penalty=presence_penalty, + max_tokens=max_tokens + )) + if len(texts) == batch_size: + break + + print("texts", len(texts)) + print("query_ids", len(query_ids)) + print("sampling_params_list", len(sampling_params_list)) + outputs = main(texts, query_ids, sampling_params_list) + + print("预测完成") + generated_text_dict = {} + print("outputs", len(outputs)) + for i, output in enumerate(outputs): + index = output.request_id + print(index) + generated_text = output.outputs[0].text + generated_text_dict[index] = generated_text + + print(generated_text_dict) + for id_, output in generated_text_dict.items(): + + return_text = {"texts": output, "probabilities": None, "status_code": 200} + load_result_path = "./new_data_logs/{}.json".format(id_) + with open(load_result_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(return_text, f2, ensure_ascii=False, indent=4) + redis_.set(id_, load_result_path, 86400) + # redis_.set(id_, load_result_path, 30) + redis_.srem(db_key_querying, id_) + log.log('start at', + 'query_id:{},load_result_path:{},return_text:{}'.format( + id_, load_result_path, return_text)) + + +if __name__ == '__main__': + t = Thread(target=classify, args=(batch_size,)) + t.start() diff --git a/qwen2_5_Instruct_model_predict_vllm_3.py b/qwen2_5_Instruct_model_predict_vllm_3.py new file mode 100644 index 0000000..27a8a43 --- /dev/null +++ b/qwen2_5_Instruct_model_predict_vllm_3.py @@ -0,0 +1,205 @@ +import os +os.environ["CUDA_VISIBLE_DEVICES"] = "3" +import argparse +from typing import List, Tuple +from threading import Thread +from vllm import EngineArgs, LLMEngine, RequestOutput, SamplingParams +# from vllm.utils import FlexibleArgumentParser +from flask import Flask, jsonify +from flask import request +import redis +import time +import json + +# http接口服务 +# app = FastAPI() +app = Flask(__name__) +app.config["JSON_AS_ASCII"] = False + +pool = redis.ConnectionPool(host='localhost', port=63179, max_connections=50,db=3, password="zhicheng123*") +redis_ = redis.Redis(connection_pool=pool, decode_responses=True) + +db_key_query = 'query' +db_key_querying = 'querying' +db_key_result = 'result' +batch_size = 15 + +class log: + def __init__(self): + pass + + def log(*args, **kwargs): + format = '%Y/%m/%d-%H:%M:%S' + format_h = '%Y-%m-%d' + value = time.localtime(int(time.time())) + dt = time.strftime(format, value) + dt_log_file = time.strftime(format_h, value) + log_file = 'log_file/access-%s' % dt_log_file + ".log" + if not os.path.exists(log_file): + with open(os.path.join(log_file), 'w', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + else: + with open(os.path.join(log_file), 'a+', encoding='utf-8') as f: + print(dt, *args, file=f, **kwargs) + + +def initialize_engine() -> LLMEngine: + """Initialize the LLMEngine from the command line arguments.""" + # model_dir = "/home/majiahui/project/models-llm/Qwen-0_5B-Chat" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b_train_11_prompt_mistral_gpt_xiaobiaot_real_paper_2" + # model_dir = "/home/majiahui/project/models-llm/Qwen2.5-7B-Instruct-1M" + # model_dir = "/home/majiahui/project/models-llm/openbuddy-qwen2.5llamaify-7b-v23.1-200k" + model_dir = "/home/majiahui/project/models-llm/qwen2_5_7B_train_11_prompt_4_gpt_xiaobiaot_real_paper_1" + args = EngineArgs(model_dir) + args.max_num_seqs = 16 # batch最大20条样本 + args.gpu_memory_utilization = 0.8 + args.max_model_len=8192 + # 加载模型 + return LLMEngine.from_engine_args(args) + +engine = initialize_engine() + + +def create_test_prompts(prompt_texts, query_ids, sampling_params_list) -> List[Tuple[str,str, SamplingParams]]: + """Create a list of test prompts with their sampling parameters.""" + + return_list = [] + + for i,j,k in zip(prompt_texts, query_ids, sampling_params_list): + return_list.append((i, j, k)) + return return_list + + +def process_requests(engine: LLMEngine, + test_prompts: List[Tuple[str, str, SamplingParams]]): + """Continuously process a list of prompts and handle the outputs.""" + + return_list = [] + while test_prompts or engine.has_unfinished_requests(): + if test_prompts: + prompt, query_id, sampling_params = test_prompts.pop(0) + engine.add_request(str(query_id), prompt, sampling_params) + + request_outputs: List[RequestOutput] = engine.step() + + for request_output in request_outputs: + if request_output.finished: + return_list.append(request_output) + return return_list + + +def main(prompt_texts, query_ids, sampling_params_list): + """Main function that sets up and runs the prompt processing.""" + + test_prompts = create_test_prompts(prompt_texts, query_ids, sampling_params_list) + return process_requests(engine, test_prompts) + + +# chat对话接口 +# @app.route("/predict/", methods=["POST"]) +# def chat(): +# # request = 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个停止句，遇到任何一个都会停止 +# +# query = request.json['query'] +# +# +# # 构造prompt +# # prompt_text, prompt_tokens = _build_prompt(generation_config, tokenizer, query, history=history, system=system) +# +# prompt_text = f"<|im_start|>user\n{query}\n<|im_end|>\n<|im_start|>assistant\n" +# +# +# return_output = main(prompt_text, sampling_params) +# return_info = { +# "request_id": return_output.request_id, +# "text": return_output.outputs[0].text +# } +# +# return jsonify(return_info) + +def classify(batch_size): # 调用模型，设置最大batch_size + while True: + texts = [] + query_ids = [] + sampling_params_list = [] + if redis_.llen(db_key_query) == 0: # 若队列中没有元素就继续获取 + time.sleep(2) + continue + + # for i in range(min(redis_.llen(db_key_query), batch_size)): + while True: + query = redis_.lpop(db_key_query) # 获取query的text + if query == None: + break + + query = query.decode('UTF-8') + data_dict_path = json.loads(query) + + path = data_dict_path['path'] + with open(path, encoding='utf8') as f1: + # 加载文件的对象 + data_dict = json.load(f1) + # query_ids.append(json.loads(query)['id']) + # texts.append(json.loads(query)['text']) # 拼接若干text 为batch + query_id = data_dict['id'] + print("query_id", query_id) + text = data_dict["text"] + model = data_dict["model"] + top_p = data_dict["top_p"] + temperature = data_dict["temperature"] + presence_penalty = 1.1 + max_tokens = 8192 + query_ids.append(query_id) + texts.append(text) + # sampling_params = SamplingParams(temperature=0.3, top_p=0.5, stop="<|end|>", presence_penalty=1.1, max_tokens=8192) + sampling_params_list.append(SamplingParams( + temperature=temperature, + top_p=top_p, + stop="<|end|>", + presence_penalty=presence_penalty, + max_tokens=max_tokens + )) + if len(texts) == batch_size: + break + + print("texts", len(texts)) + print("query_ids", len(query_ids)) + print("sampling_params_list", len(sampling_params_list)) + outputs = main(texts, query_ids, sampling_params_list) + + print("预测完成") + generated_text_dict = {} + print("outputs", len(outputs)) + for i, output in enumerate(outputs): + index = output.request_id + print(index) + generated_text = output.outputs[0].text + generated_text_dict[index] = generated_text + + print(generated_text_dict) + for id_, output in generated_text_dict.items(): + + return_text = {"texts": output, "probabilities": None, "status_code": 200} + load_result_path = "./new_data_logs/{}.json".format(id_) + with open(load_result_path, 'w', encoding='utf8') as f2: + # ensure_ascii=False才能输入中文，否则是Unicode字符 + # indent=2 JSON数据的缩进，美观 + json.dump(return_text, f2, ensure_ascii=False, indent=4) + redis_.set(id_, load_result_path, 86400) + # redis_.set(id_, load_result_path, 30) + redis_.srem(db_key_querying, id_) + log.log('start at', + 'query_id:{},load_result_path:{},return_text:{}'.format( + id_, load_result_path, return_text)) + + +if __name__ == '__main__': + t = Thread(target=classify, args=(batch_size,)) + t.start() diff --git a/run_api_gunicorn.sh b/run_api_gunicorn.sh index 4060eb5..dc56c02 100644 --- a/run_api_gunicorn.sh +++ b/run_api_gunicorn.sh @@ -1 +1 @@ -gunicorn mistral_api:app -c gunicorn_config.py +gunicorn model_api:app -c gunicorn_config.py diff --git a/run_model_1.sh b/run_model_1.sh new file mode 100644 index 0000000..68b978b --- /dev/null +++ b/run_model_1.sh @@ -0,0 +1 @@ +nohup python mistral_model_predict_vllm_1.py > myout_mis_model_1.file 2>&1 & diff --git a/run_model_4.sh b/run_model_4.sh new file mode 100644 index 0000000..3a30332 --- /dev/null +++ b/run_model_4.sh @@ -0,0 +1 @@ +nohup python mistral_model_predict_vllm_4.py > myout_mis_model_4.file 2>&1 & diff --git a/run_model_openbuddy_llama3_1_1.sh b/run_model_openbuddy_llama3_1_1.sh new file mode 100644 index 0000000..9bbc9aa --- /dev/null +++ b/run_model_openbuddy_llama3_1_1.sh @@ -0,0 +1 @@ +nohup python python openbuddy_llama3_1_model_predict_vllm_1.py > myout_model_openbuddy_llama3_1_1.file 2>&1 & diff --git a/run_model_openbuddy_llama3_1_2.sh b/run_model_openbuddy_llama3_1_2.sh new file mode 100644 index 0000000..bc2c523 --- /dev/null +++ b/run_model_openbuddy_llama3_1_2.sh @@ -0,0 +1 @@ +nohup python openbuddy_llama3_1_model_predict_vllm_2.py > myout_model_openbuddy_llama3_1_2.file 2>&1 & \ No newline at end of file diff --git a/run_model_openbuddy_llama3_1_3.sh b/run_model_openbuddy_llama3_1_3.sh new file mode 100644 index 0000000..f334527 --- /dev/null +++ b/run_model_openbuddy_llama3_1_3.sh @@ -0,0 +1 @@ +nohup python openbuddy_llama3_1_model_predict_vllm_3.py > myout_model_openbuddy_llama3_1_3.file 2>&1 & \ No newline at end of file diff --git a/run_model_openbuddy_qwen_1.sh b/run_model_openbuddy_qwen_1.sh new file mode 100644 index 0000000..28e22cc --- /dev/null +++ b/run_model_openbuddy_qwen_1.sh @@ -0,0 +1 @@ +nohup python openbuddy_qwen2_5_model_predict_vllm_1.py > myout_model_openbuddy_qwen_1.file 2>&1 & diff --git a/run_model_openbuddy_qwen_2.sh b/run_model_openbuddy_qwen_2.sh new file mode 100644 index 0000000..459cfe4 --- /dev/null +++ b/run_model_openbuddy_qwen_2.sh @@ -0,0 +1 @@ +nohup python openbuddy_qwen2_5_model_predict_vllm_2.py > myout_model_openbuddy_qwen_2.file 2>&1 & \ No newline at end of file diff --git a/run_model_openbuddy_qwen_3.sh b/run_model_openbuddy_qwen_3.sh new file mode 100644 index 0000000..2039531 --- /dev/null +++ b/run_model_openbuddy_qwen_3.sh @@ -0,0 +1 @@ +nohup python openbuddy_qwen2_5_model_predict_vllm_3.py > myout_model_openbuddy_qwen_3.file 2>&1 & \ No newline at end of file diff --git a/run_model_qwen_Instruct1.sh b/run_model_qwen_Instruct1.sh new file mode 100644 index 0000000..37b67c5 --- /dev/null +++ b/run_model_qwen_Instruct1.sh @@ -0,0 +1 @@ +nohup python qwen2_5_Instruct_model_predict_vllm_1.py > myout_model_qwen_1.file 2>&1 & diff --git a/run_model_qwen_Instruct2.sh b/run_model_qwen_Instruct2.sh new file mode 100644 index 0000000..888c8b8 --- /dev/null +++ b/run_model_qwen_Instruct2.sh @@ -0,0 +1 @@ +nohup python qwen2_5_Instruct_model_predict_vllm_2.py > myout_model_qwen_2.file 2>&1 & \ No newline at end of file diff --git a/run_model_qwen_Instruct3.sh b/run_model_qwen_Instruct3.sh new file mode 100644 index 0000000..53a5cb2 --- /dev/null +++ b/run_model_qwen_Instruct3.sh @@ -0,0 +1 @@ +nohup python qwen2_5_Instruct_model_predict_vllm_3.py > myout_model_qwen_3.file 2>&1 & \ No newline at end of file