You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
161 lines
7.8 KiB
161 lines
7.8 KiB
# coding:utf8
|
|
import os, sys
|
|
import io
|
|
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf8')
|
|
cur_dir = os.path.dirname(os.path.abspath(__file__)) or os.getcwd()
|
|
par_dir = os.path.abspath(os.path.join(cur_dir, os.path.pardir))
|
|
sys.path.append(cur_dir)
|
|
sys.path.append(par_dir)
|
|
import json
|
|
from django.http import HttpResponse
|
|
from text_analysis.tools import to_kafka
|
|
from text_analysis.cusException import promptLen_Exception
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from log_util.set_logger import set_logger
|
|
from openai import OpenAI
|
|
logging = set_logger('logs/results.log')
|
|
import traceback
|
|
import queue
|
|
import requests
|
|
from text_analysis.tools.tool import get_content,parse_gptResult
|
|
import uuid
|
|
import time
|
|
from kazoo.client import KazooClient
|
|
from kazoo.protocol.states import EventType
|
|
# openai_api_key = "EMPTY"
|
|
# openai_api_base = "http://10.0.32.225:9000/v1"
|
|
# client = OpenAI(api_key=openai_api_key,base_url=openai_api_base)
|
|
task_queue = queue.Queue()
|
|
stop_dict={}
|
|
|
|
@csrf_exempt
|
|
def QwenModel(request):
|
|
if request.method == 'POST':
|
|
try:
|
|
# txt=request.body.encode("utf-8")
|
|
raw_data = json.loads(request.body)
|
|
task_queue.put(raw_data)
|
|
return HttpResponse(json.dumps({"code": 1, "msg": "请求正常！"}, ensure_ascii=False))
|
|
except:
|
|
logging.error(traceback.format_exc())
|
|
return HttpResponse(json.dumps({"code": 0, "msg": "请求json格式不正确！"}, ensure_ascii=False))
|
|
else:
|
|
return HttpResponse(json.dumps({"code": 0, "msg": "请求方式错误，改为post请求"}, ensure_ascii=False))
|
|
|
|
|
|
def Qwen():
|
|
while True:
|
|
if task_queue.qsize() > 0:
|
|
try:
|
|
logging.info("取任务队列长度{}".format(task_queue.qsize()))
|
|
raw_data = task_queue.get()
|
|
output = raw_data["output"]
|
|
res_tmp = {key: "" for key in output}
|
|
if "id" in res_tmp.keys():
|
|
res_tmp["id"] = str(uuid.uuid4())
|
|
task_id=raw_data["scenes_id"]
|
|
task_version=raw_data["version"]
|
|
if "data" not in raw_data.keys():
|
|
logging.info("任务缺少data—{}".format(raw_data))
|
|
raw_data["result"] = {"successCode": "0", "errorLog": "", "results": "", "status": 2,"message": "未配置data内容"}
|
|
res_tmp_json = json.dumps(res_tmp, ensure_ascii=False)
|
|
raw_data["result"]["results"] = res_tmp_json
|
|
to_kafka.send_kafka(raw_data, logging)
|
|
else:
|
|
logging.info("任务数据为：{}".format(raw_data))
|
|
logging.info("当前version信息为：{}".format(stop_dict))
|
|
if task_id in stop_dict.keys() and task_version!=stop_dict[task_id]["version"]:
|
|
logging.info("已暂停任务，过滤掉。{}".format(raw_data))
|
|
continue
|
|
data = get_content(raw_data, logging)
|
|
if len(data["prompt"])>=10000:
|
|
raise promptLen_Exception
|
|
# logging.info("请求信息为{},解析后模型请求为{}".format(raw_data,data))
|
|
url="http://10.0.32.225:9000/v1/chat/completions"
|
|
headers = {
|
|
"Content-Type": "application/json;charset=UTF-8"
|
|
}
|
|
payload = json.dumps({
|
|
"model":"Qwen2-72B-Instruct-GPTQ-Int4",
|
|
"messages": [{"role": "user", "content": data["prompt"]}],
|
|
"temperature": float(data["temperature"]),
|
|
"top_p": float(data["top_p"]),
|
|
"n": int(data["n"])
|
|
})
|
|
# logging.info("prompt为{}".format(data["prompt"]))
|
|
response = requests.request("POST", url, headers=headers, data=payload,timeout=180)
|
|
logging.info("Prompt为：{}—Qwen返回值：{}-{}".format(data["prompt"],response,response.text))
|
|
d = json.loads(response.text)
|
|
if response.status_code==400 and "maximum context length" in d["message"]:
|
|
raise promptLen_Exception
|
|
result = d['choices'][0]['message']['content']
|
|
if result:
|
|
#添加 0是文本，1是json格式
|
|
fieldType = raw_data["input"]['fieldType']
|
|
if fieldType == 0:
|
|
res_tmp["content"] = result
|
|
res_tmp_json = json.dumps(res_tmp, ensure_ascii=False)
|
|
raw_data["result"] = {"successCode": "1", "errorLog": "", "results": res_tmp_json,"status":1,"message":"成功"}
|
|
else:
|
|
res=parse_gptResult(res_tmp,result)
|
|
if res:
|
|
res_tmp_json = json.dumps(res, ensure_ascii=False)
|
|
raw_data["result"] = {"successCode": "1", "errorLog": "", "results": res_tmp_json,"status":1,"message":"成功"}
|
|
else:
|
|
res_tmp_json = json.dumps(res_tmp, ensure_ascii=False)
|
|
raw_data["result"] = {"successCode": "0", "errorLog": "GPT返回结果不符合预期", "results": res_tmp_json,"status":2,"message":"GPT返回结果不符合预期"}
|
|
else:
|
|
logging.info("模型返回值正常但为空-{}".format(raw_data))
|
|
res_tmp_json = json.dumps(res_tmp, ensure_ascii=False)
|
|
raw_data["result"] = {"successCode": "1", "errorLog": "", "results": res_tmp_json, "status": 1,"message": "成功"}
|
|
|
|
logging.info(raw_data)
|
|
to_kafka.send_kafka(raw_data, logging)
|
|
except promptLen_Exception:
|
|
logging.info("文本长度超过模型限制-{}".format(raw_data))
|
|
res_tmp_json = json.dumps(res_tmp, ensure_ascii=False)
|
|
raw_data["result"] = {"successCode": "0", "errorLog": "", "results": res_tmp_json, "status": 2,"message": "文本长度超过模型限制"}
|
|
to_kafka.send_kafka(raw_data, logging)
|
|
|
|
except:
|
|
res_tmp_json = json.dumps(res_tmp, ensure_ascii=False)
|
|
raw_data["result"] = {"successCode": "0", "errorLog": traceback.format_exc(), "results": res_tmp_json,"status":2,"message":"异常"}
|
|
logging.info("调用Qwen失败{}-{}".format(raw_data, traceback.format_exc()))
|
|
to_kafka.send_kafka(raw_data, logging)
|
|
|
|
else:
|
|
logging.info("暂无任务，进入休眠--")
|
|
time.sleep(10)
|
|
|
|
def zk_monitoring():
|
|
try:
|
|
#线上环境
|
|
zk = KazooClient(hosts='172.18.1.146:2181,172.18.1.147:2181,172.18.1.148:2181')
|
|
#测试环境
|
|
# zk = KazooClient(hosts='172.16.12.55:2181,172.16.12.56:2181,172.16.12.57:2181')
|
|
zk.start()
|
|
# 设置监听器
|
|
@zk.DataWatch("/analyze")
|
|
def watch_node(data, stat, event):
|
|
if event is not None and event.type == EventType.CHANGED:
|
|
data, stat = zk.get("/analyze")
|
|
logging.info("执行删除操作：{}".format(data))
|
|
d = json.loads(data)
|
|
id = d["scenes_id"]
|
|
stop_dict[id] = {}
|
|
stop_dict[id]["version"] = d["version"]
|
|
stop_dict[id]["operation"] = d["operation"]
|
|
# 保持程序运行以监听节点变化
|
|
try:
|
|
while True:
|
|
time.sleep(1)
|
|
except:
|
|
logging.info("Stopping...")
|
|
# 关闭连接
|
|
zk.stop()
|
|
zk.close()
|
|
except:
|
|
logging.error(traceback.format_exc())
|
|
|
|
|