ai_qback/gpt/chatgpt.py

# -*- coding:utf-8 -*-
if __name__ == '__main__':
    import os

    os.chdir("..")

import requests
import random
import json
import time
from tools.loglog import logger, simple_logger, log_err_e
from tools.new_mysql import MySQLUploader
from typing import Optional, Dict, Any, Union
import httpx
import asyncio

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 get_article_gpt_pydantic(question, real_ip="localhost", demo_name="无", model="gpt-4.1", max_tokens=3500, temperature: float = 0, n=1,
                             check_fucn=None, sys_prompt=None, task_id=0, exercise_id=0):
    """
    异步获取文章
    :param question: 问题
    :param real_ip: 真实IP
    :param demo_name: 项目名称
    :param model: 模型名称
    :param max_tokens: 最大token数
    :param temperature: 温度
    :param n: 生成数量
    :param check_fucn: 校验函数
    :param sys_prompt: 系统提示
    :param task_id: 任务id
    :param exercise_id: 学案id

    :return: 文章内容
    """

    d2 = {"model": model, "messages": [], "max_tokens": max_tokens, "temperature": temperature, "n": n,
          "response_format": {"type": "json_schema", 'json_schema': {'name': 'Article', 'schema': {'$defs': {'Candidate': {
              'properties': {'label': {'description': 'ABCD序号的一种', 'title': '序号', 'type': 'string'},
                             'text': {'description': '英文，ABCD选项的文本', 'title': '选项文本', 'type': 'string'},
                             'isRight': {'description': '1是正确，0是错误', 'title': '是否是正确答案', 'type': 'integer'}},
              'required': ['label', 'text', 'isRight'], 'title': 'Candidate', 'type': 'object'}, 'DifficultSentence': {
              'properties': {'english': {'description': '文章中的一句难句', 'title': '英语难句', 'type': 'string'},
                             'chinese': {'description': '对英语难句的翻译', 'title': '中文难句', 'type': 'string'}},
              'required': ['english', 'chinese'], 'title': 'DifficultSentence', 'type': 'object'}, 'Question': {
              'properties': {'trunk': {'description': '用英语给出的选择题题目', 'title': '选择题题目', 'type': 'string'},
                             'analysis': {'description': '中文，选择题答案的分析思路', 'title': '选择题分析', 'type': 'string'},
                             'candidates': {'items': {'$ref': '#/$defs/Candidate'}, 'title': '选项对象', 'type': 'array'}},
              'required': ['trunk', 'analysis', 'candidates'], 'title': 'Question', 'type': 'object'}}, 'properties': {
              'difficultSentences': {'description': '挑选一句难句对象', 'items': {'$ref': '#/$defs/DifficultSentence'}, 'title': '难句对象',
                                     'type': 'array'},
              'usedMeanIds': {'items': {'type': 'integer'}, 'title': '用到的词义id', 'type': 'array'},
              'questions': {'description': '针对英语文章的选择题', 'items': {'$ref': '#/$defs/Question'}, 'title': '问题对象',
                            'type': 'array'}, 'englishArticle': {'description': '', 'title': '英语文章', 'type': 'string'},
              'chineseArticle': {'description': '', 'title': '中文翻译', 'type': 'string'}}, 'required': ['difficultSentences',
                                                                                                          'usedMeanIds', 'questions',
                                                                                                          'englishArticle',
                                                                                                          'chineseArticle'],
                                                                                                   'title': 'Article', 'type': 'object'}}}
          }
    if sys_prompt:
        d2['messages'].append({"role": "system", "content": sys_prompt})
    d2['messages'].append({"role": "user", "content": question})

    for num_count in range(3):
        try:
            response = requests.post('http://170.106.108.95/v1/chat/completions', json=d2)
            r_json = response.json()
            simple_logger.info(f"问题日志task_id:{task_id},exercise_id:{exercise_id}：\n{question}\n回答日志：\n{r_json}")
            return r_json

            #

            #





        except httpx.HTTPError as e:
            logger.error(f"HTTP请求错误: {str(e)}")
            if num_count < 2:
                time.sleep(10)
            else:
                raise
        except Exception as e:
            log_err_e(e, "其他错误")

            if num_count < 2:
                time.sleep(10)
            else:
                raise

    logger.critical("get_article_gpt_pydantic 严重错误，3次后都失败了")
    raise Exception("获取文章失败，已达到最大重试次数")


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__':

    def get_article():
        """可以在这里测试提示词的好坏"""

        sys_prompt = "你是一个专业的英语老师，擅长根据用户提供的词汇生成对应的英语文章和中文翻译和4个配套选择题。注意:生成的文章用到提供的单词，其词义一定要是提供的中文词义，不要使用同个单词，与提供词义不同的单词。例如：单词might的词义可能有[1.可能 2.强大力量]两种词义，不要混用同个单词的词义，一定要按照提供词义来使用。"
        q = """下面我会为你提供一组数据，[单词组]（里面包含词义id，英语单词，中文词义），请根据这些单词的中文词义，生成一篇带中文翻译的考场英语文章，英语文章和中文翻译要有[标题]。注意这个单词有多个词义时，生成的英语文章一定要用提供的中文词义。并挑选一句复杂的句子和其中文翻译，放入difficultSentences。英语文章，放入"englishArticle"中。中文翻译，放入"chineseArticle"中。最终文中使用到的单词id放入"usedMeanIds"中。4个选择题，放入questions字段。questions结构下有4个选择题对象，其中trunk是[英语]问题文本，analysis是[中文]的问题分析，candidates是4个ABCD选项，内部有label是指选项序号A B C D ，text是[英语]选项文本，isRight是否正确答案1是正确0是错误。
    
要求:
1.必须用提供的这个词义的单词，其他单词使用最简单最容易没有难度的单词。文章整体非常简洁，通俗易懂，适合初学者，刚入门，单词全是最常见的，语句通顺即可。选择题难度尽可能简单，参考中国小学生水平
2.优先保证文章语句通顺，意思不要太生硬。不要为了使用特定的单词，造成文章语义前后不搭，允许不使用个别词义。
3.文章中使用提供单词，一定要和提供单词的中文词义匹配，尤其是一词多义时，务必使用提供单词的词义。必须要用提供单词的词义。如果用到的词义与提供单词词义不一致，请不要使用这个单词。
4.生成的文章要求70词左右，可以用\\n\\n字符分段，一般1-2个段落左右。第一段是文章标题。
5.允许不使用[单词组]的个别单词，优先保证文章整体意思通顺连贯和故事完整。
6.注意回复字段的中英文，englishArticle是英文，chineseArticle是中文，其中trunk是英文，analysis是中文，text是英文。

提供[单词组]：[260  important 重要的]; [287  off 离开, 从…下去]; [74  get 变得]; [251  change 零钱]; [219  put 放]; [212  feel 触, 摸]; [239  few 很少的, 几乎没有的]; [262  still 仍然]; [283  country 国家]; [270  part 部分, 部件];
"""
        gpt_resp = get_article_gpt_pydantic(question=q, temperature=1.2, sys_prompt=sys_prompt, model="gpt-4.1", task_id=888,
                                            exercise_id=999, n=4)

        return_json = {"articles": []}
        for choice in gpt_resp["choices"]:
            single_article_dict = json.loads(choice["message"]["content"])
            return_json["articles"].append(single_article_dict)
        return return_json