AI_customer_service/module/intent_judge.py

# 标准库导入
import time
from functools import lru_cache
from typing import Dict, List, Any, Optional

# 第三方库导入
from langchain.chains.base import Chain
from pydantic import Field

# 本地导入
from utils.logger_config import setup_logger
from utils.prompt_config import prompt_router
from utils.rag_config import BaseMethod, BotMethod, ConfigManager

# 配置
logger = setup_logger(__name__)

class Judge(Chain):
    """
    意图识别判官，用于意图识别模块的评估
    继承自Chain，并组合BotMethod的功能
    """
    
    # 声明Pydantic字段
    config_manager: ConfigManager = Field(default_factory=ConfigManager)
    bot_method: BotMethod = Field(default_factory=BotMethod)
    prompt_template: Any = None

    def __init__(self, sys_prompt, config_path=None):
        """
        初始化意图判官
        
        Args:
            config_path: 配置文件路径，如果为None则使用默认路径
        """
        super().__init__()
        # 使用指定的配置文件初始化
        if config_path:
            self.config_manager = ConfigManager(config_path)
            self.bot_method = BotMethod(config_path)
        self._init_prompt_template(sys_prompt)
    
    def _init_prompt_template(self, sys_prompt) -> None:
        """初始化提示词模板"""
        try:
            self.prompt_template = prompt_router(sys_prompt)
        except Exception as e:
            logger.error(f"初始化提示词模板失败: {e}")
            raise

    @property
    def input_keys(self) -> List[str]:
        """定义输入键"""
        return ["question"]

    @property
    def output_keys(self) -> List[str]:
        """定义输出键"""
        return ["answer"]

    def _generate_answer(self, content: str, question: str, llm_answer: str) -> str:
        """
        生成答案
        
        Args:
            content: 上下文信息
            
        Returns:
            生成的答案
            
        Raises:
            Exception: 生成答案失败时抛出异常
        """
        try:
            prompt = self.prompt_template.format(
                content=content,
                question=question,
                llm_answer=llm_answer
            )
            return self.bot_method.model_config.llm(prompt).content
        except Exception as e:
            logger.error(f"生成答案失败: {e}")
            raise

    def _call(self, inputs: Dict[str, Any]) -> Dict[str, str]:
        """
        处理用户输入并返回答案
        
        Returns:
            包含答案的字典
            
        Raises:
            KeyError: 输入缺少必要字段
            Exception: 处理过程中的其他异常
        """
        try:
            
            # 参数解析
            content = inputs["content"]
            question = inputs["question"]
            llm_answer = inputs["llm_answer"]

            # 生成答案
            answer = self._generate_answer(content=content, question=question, llm_answer=llm_answer)
            logger.info(f"评估结果: {answer}")

            # 返回结果
            return {"answer": answer}

        except Exception as e:
            logger.error(f"处理问题失败: {e}")
            raise

if __name__ == "__main__":
    # 使用示例
    chain = Judge(sys_prompt="intent_judge")
    try:
        content = "本商店支持全国顺丰包邮，提供运费险"
        question = "请问有运费险吗？"
        llm_answer = "亲亲，咱们是不提供运费险的呢~"
        
        result = chain.invoke({"content": content, "question": question, "llm_answer": llm_answer})
        print(f"回答: {result['answer']}")
    except Exception as e:
        logger.error(f"处理失败: {e}")