LQAgentPlatform/utils/tool_utils.py

import time
from math import log
import os
from dotenv import load_dotenv
from enums.common_enums import BusinessSceneEnum, ErrorCodeEnum, UserRoleEnum
from functools import wraps

from logger.loggering import server_logger
from utils.common import handler_err
from base.config import config_handler

# 获取当前文件的目录
current_dir = os.path.dirname(__file__)
# 构建到 .env 的相对路径
conf_file_path = os.path.join(current_dir , '../',  '.env')
#server_logger.info(f"当前目录: {conf_file_path}")
# 加载环境变量
load_dotenv(dotenv_path=conf_file_path)

def verify_param(param: dict):
    """
        验证请求参数
    """
    input_data = param.get("input")
    session_id = param.get("config").get("session_id")
    businessScene = param.get("businessScene")
    if input_data is None:
        raise ValueError(ErrorCodeEnum.INPUT_INFO_EMPTY.__str__)
    if session_id is None:
        raise ValueError(ErrorCodeEnum.SESSION_ID_EMPTY.__str__)
    # 是否可使用默认的通用模型查询 默认 False
    use_default_common_model_query = os.environ.get("USE_DEFAULT_COMMON_MODEL_QUERY" , False)
    server_logger.info(f"使用可默认的通用模型查询: {use_default_common_model_query}")
    
    if not use_default_common_model_query:
        if businessScene is None:
            raise ValueError(ErrorCodeEnum.BUSINSESS_SCENE_EMPTY.__str__)
        if not BusinessSceneEnum.get_item_by_code(param.get('businessScene')):
            raise ValueError(ErrorCodeEnum.BUSINSESS_SCENE_ERROR.__str__)



def get_system_prompt() -> str:
    """
        获取系统提示语
    """
    system_prompt = config_handler.get("system", "SYSTEM_PROMPT")
    server_logger.info(f"获取系统提示语: {system_prompt}")
    return str(system_prompt)



def get_business_scene_prompt(business_scene):
    """
        获取业务场景的提示语
    """
    # 默认公共查询提示语
    business_scene_enum = BusinessSceneEnum.COMMON_MODEL_QUERY
    prompt_file = business_scene_enum.prompt_file
     # 是否可使用默认的通用模型查询 默认 False
    use_default_common_model_query = os.environ.get("USE_DEFAULT_COMMON_MODEL_QUERY" , False)
    if not business_scene is None:
        business_scene_enum = BusinessSceneEnum.get_item_by_code(business_scene)
        if not business_scene_enum:
            raise ValueError("未找到枚举值")
        if business_scene_enum.prompt_file is None:
            raise ValueError("业务场景不存在")
        prompt_file = business_scene_enum.prompt_file
    
    prompt_file = os.path.join(current_dir , '../', 'config', 'prompt' , prompt_file)
    server_logger.info(f"获取业务场景提示语: {prompt_file}")
    if not os.path.exists(prompt_file):
        raise ValueError("业务场景不存在")
    
    try:
        with open(prompt_file, 'r', encoding='utf-8') as f:
          return business_scene_enum , '\n'.join(f.readlines())
    except Exception as e:
        handler_err(server_logger, e,err_name="get_business_scene_prompt")
        server_logger.error(f"获取业务场景提示语失败: {e}")
        raise e


def get_fixed_problem_answer_txt_content(file_name: str):
    """
        获取固定问题答案内容
    """
    file_name = file_name+".txt"
    answer_txt_file = os.path.join(current_dir , '../', 'config', 'fixed_answer' , file_name)
    server_logger.info(f"固定回答文本内容: {answer_txt_file}")
    if not os.path.exists(answer_txt_file):
        raise ValueError("固定回答文本不存在")
    
    try:
        result_list = []
        with open(answer_txt_file, 'r', encoding='utf-8') as f:
           result_list=f.readlines()
        return "".join(result_list)
    except Exception as e:
        handler_err(server_logger, e,err_name="get_fixed_problem_answer_txt_content")
        server_logger.error(f"获取固定回答文本失败: {e}")
        raise e




def verify_user_role(user_role: str):
    """
        验证用户角色
            普通用户 common  ，不能检索查询知识库
            租户用户 tenant  ，只有租户才能检索查询知识库
    """
    if user_role in [UserRoleEnum.TENANT.code]:
        return True
    
    return False