LQAgentPlatform/utils/yaml_utils.py

# ！/usr/bin/ python
# -*- coding: utf-8 -*-
'''
@Project    : lq-agent-api
@File       :yaml_utils.py
@IDE        :PyCharm
@Author     :
@Date       :2025/7/10 17:32
'''

import os
import yaml
from logger.loggering import server_logger

import os
from dotenv import load_dotenv
from enums.common_enums import BusinessSceneEnum, ErrorCodeEnum
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}")



def get_fixed_problem_answer() -> dict:
    """
        固定问题回答
    """
     # 构建文件路径 判断文件是否存在
    yaml_file = get_yaml_file_path("fixed_problem_answer.yaml")
    
    try:
        with open(yaml_file, 'r', encoding='utf-8') as f:
            prompt_config = yaml.safe_load(f)
        # 验证必需字段
        #validate_prompt_config(prompt_config, prompt_name)
        server_logger.info(f"成功加载[固定问题]回答系统配置.fixed_problem_answer: {prompt_config["fixed_problem_answer"]}")
        return prompt_config
        
    except Exception as e:
        server_logger.error(f"加载[固定问题]回答fixed_problem_answer文件失败: {yaml_file}, 错误: {str(e)}")
        raise



def get_intent_prompt() -> dict:
    """
        获取意图识别 系统提示语
    """
     # 构建文件路径 判断文件是否存在
    yaml_file = get_yaml_file_path("intent_prompt.yaml")
    
    try:
        with open(yaml_file, 'r', encoding='utf-8') as f:
            prompt_config = yaml.safe_load(f)
        # 验证必需字段
        #validate_prompt_config(prompt_config, prompt_name)
        server_logger.info(f"成功加载[意图识别]系统.system_prompt配置: {prompt_config["system_prompt"]}")
        server_logger.info(f"成功加载[意图识别]系统配置.examples: {prompt_config["intent_examples"]}")
        return prompt_config
        
    except Exception as e:
        server_logger.error(f"加载意图识别intent_prompt文件失败: {yaml_file}, 错误: {str(e)}")
        raise



def get_fixed_question_intent_prompt() -> dict:
    """
        获取 固定问题意图识别 系统提示语
    """
     # 构建文件路径 判断文件是否存在
    yaml_file = get_yaml_file_path("fixed_intent_prompt.yaml")
    
    try:
        with open(yaml_file, 'r', encoding='utf-8') as f:
            prompt_config = yaml.safe_load(f)
        # 验证必需字段
        #validate_prompt_config(prompt_config, prompt_name)
        server_logger.info(f"成功加载[固定问题意图识别]系统.system_prompt配置: {prompt_config["system_prompt"]}")
        server_logger.info(f"成功加载[固定问题意图识别]系统配置.examples: {prompt_config["fixed_problem_answer"]}")
        return prompt_config
        
    except Exception as e:
        server_logger.error(f"加载意图识别fixed_intent_prompt文件失败: {yaml_file}, 错误: {str(e)}")
        raise


def get_system_prompt() -> dict:
    """
        获取系统提示语
    """
     # 构建文件路径 判断文件是否存在
    yaml_file = get_yaml_file_path("system_prompt.yaml")
    
    try:
        with open(yaml_file, 'r', encoding='utf-8') as f:
            prompt_config = yaml.safe_load(f)
        # 验证必需字段
        #validate_prompt_config(prompt_config, prompt_name)
        server_logger.info(f"成功加载系统system_prompt配置: {prompt_config["system_prompt"]}")
        return prompt_config
        
    except Exception as e:
        server_logger.error(f"加载system_prompt文件失败: {yaml_file}, 错误: {str(e)}")
        raise





def get_business_scene_prompt(trace_id, business_scene) -> tuple[BusinessSceneEnum , dict]:
    """
        获取业务场景的提示语
    """
    # 默认公共查询提示语
    business_scene_enum = BusinessSceneEnum.COMMON_MODEL_QUERY
    prompt_file = business_scene_enum.prompt_file
    if not business_scene is None:
        # 2025-07-25 修改 如果未找到 返回默认值通用场景
        business_scene_enum = BusinessSceneEnum.get_item_by_code_def_val(business_scene , BusinessSceneEnum.CATTLE_FARM_COMMMON)
        if not business_scene_enum:
            raise ValueError("未找到枚举值")
        if business_scene_enum.prompt_file is None:
            raise ValueError("业务场景不存在")
        prompt_file = business_scene_enum.prompt_file
    
    # 构建文件路径 判断文件是否存在
    yaml_file = get_yaml_file_path(prompt_file)
    
    try:
        with open(yaml_file, 'r', encoding='utf-8') as f:
            prompt_config = yaml.safe_load(f)
            server_logger.info(trace_id=trace_id , msg=f"business_scene_enum:{business_scene_enum.code} Get prompt successfully.")
            return business_scene_enum , prompt_config
    except Exception as e:
        handler_err(server_logger, e, trace_id=trace_id, err_name="get_business_scene_prompt")
        server_logger.error(trace_id=trace_id , msg=f"获取业务场景任务提示语失败: {e}")
        raise e




def get_yaml_file_path(file_name: str) -> str:
    """
        获取yaml文件路径
        :param file_name:
        :return:
    """
    yaml_file = os.path.join(current_dir , '../', 'config', 'prompt' , file_name)
    if not os.path.exists(yaml_file):
        raise FileNotFoundError(f"Prompt文件不存在: {file_name}")
    return yaml_file



#获取系统提示语
system_prompt_config = get_system_prompt()