# -*- coding:utf-8 -*- if __name__ == '__main__': import os os.chdir("..") import requests import random import time from tools.loglog import logger,simple_logger from tools.new_mysql import MySQLUploader from typing import Optional, Dict, Any,Union m = MySQLUploader() def get_openai_model(model_text:str): """模糊获得模型名""" if "3.5" in model_text or "3.5-turbo" in model_text or "3.5turbo" in model_text: model = "gpt-3.5-turbo" elif "4o" in model_text or "gpt4o" in model_text: model = "gpt-4o" elif "4turbo" in model_text or "4-turbo" in model_text: model = "gpt-4-turbo" else: model = "gpt-4o" return model def insert_ip_token(ip,demo_name,gpt_content,prompt_tokens,completion_tokens,total_tokens): sql = "insert into consumer_token (ip,demo_name,gpt_content,prompt_tokens,completion_tokens,total_tokens) values (%s,%s,%s,%s,%s,%s)" m.execute_(sql,(ip,demo_name,str(gpt_content),prompt_tokens,completion_tokens,total_tokens)) def get_answer_from_gpt(question,real_ip="localhost",demo_name="无",model="gpt-4o",max_tokens=3500,temperature:float=0, json_resp:Union[Dict[Any, Any],bool]=False,n=1,check_fucn=None,sys_prompt=None): model = get_openai_model(model) d2 = {"model": model,"messages": [],"max_tokens": max_tokens,"temperature": temperature,'n': n} if sys_prompt: d2['messages'].append({"role": "system", "content": sys_prompt}) d2['messages'].append({"role": "user", "content": question}) if json_resp is True: d2["response_format"] = {"type": "json_object"} elif json_resp is False: pass else: d2["response_format"] = json_resp for num_count in range(3): try: response = requests.post(f'http://170.106.108.95/v1/chat/completions', json=d2) r_json = response.json() if r2:= r_json.get("choices",None): if n>1: gpt_res = [] for i in r2: gpt_res.append(i["message"]["content"]) else: gpt_res= r2[0]["message"]["content"] gpt_content = str(gpt_res) prompt_tokens = r_json["usage"]["prompt_tokens"] completion_tokens = r_json["usage"]["completion_tokens"] total_tokens = r_json["usage"]["total_tokens"] insert_ip_token(real_ip,demo_name,gpt_content,prompt_tokens,completion_tokens,total_tokens) simple_logger.info(f"问题日志:\n{question}\n回答日志:\n{gpt_res}") if not check_fucn: return gpt_res check_result = check_fucn(str(gpt_res)) if check_result: return gpt_res else: raise Exception(f"第{num_count+1}次共3次,GPT的校验没有通过,校验函数:{check_fucn.__name__}") elif r_json.get("message") == "IP address blocked": print("IP address blocked") raise Exception("IP address blocked") else: print(f"小错误:{question[:10]}") logger.error(response.text) except Exception as e: logger.info(f"小报错忽略{e}") time.sleep(10) logger.critical("get_answer_from_gpt 严重错误,3次后都失败了") def parse_gpt_phon_to_tuplelist(text:str) -> list: """解析gpt返回的音标数据""" result = [] if not text: return [] for i in text.split("\n"): ii = i.split("***") if len(ii)>=3: result.append((ii[0].strip(),ii[1].strip(),ii[2].strip())) return result if __name__ == '__main__': question = "hello" print(get_answer_from_gpt(question,temperature=0.8))