# !/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 foundation.observability.logger.loggering import server_logger import os from dotenv import load_dotenv from functools import wraps from foundation.observability.logger.loggering import server_logger from foundation.utils.common import handler_err from foundation.infrastructure.config import config_handler # 获取当前文件的目录 current_dir = os.path.dirname(__file__) # 获取项目根目录 project_root = os.path.dirname(os.path.dirname(current_dir)) # 构建到 .env 的相对路径 conf_file_path = os.path.join(project_root, '.env') #server_logger.info(f"当前目录: {conf_file_path}") 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_yaml_file_path(file_name: str) -> str: """ 获取yaml文件路径 :param file_name: :return: """ yaml_file = os.path.join(project_root, 'config', 'prompt' , file_name) if not os.path.exists(yaml_file): raise FileNotFoundError(f"Prompt文件不存在: {file_name}") return yaml_file 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 #获取系统提示语 - 延迟加载 system_prompt_config = None def get_system_prompt_config(): """获取系统提示语配置(延迟加载)""" global system_prompt_config if system_prompt_config is None: system_prompt_config = get_system_prompt() return system_prompt_config