LQAgentPlatform/agent/intent.py

# ！/usr/bin/ python
# -*- coding: utf-8 -*-
'''
@Project    : lq-agent-api
@File       :intent.py
@IDE        :PyCharm
@Author     : 
@Date       :2025/7/14 12:04
'''


import os
import sys
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from logger.loggering import server_logger
from utils.utils import get_models
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.prompts import HumanMessagePromptTemplate
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.prompts import FewShotChatMessagePromptTemplate
from utils import yaml_utils
from agent.session.session_manager import SessionContextMemoryManager
from base.config import config_handler


class IntentIdentifyClient:

    def __init__(self):
        """
            创建意图识别类
        """
          # 获取部署的模型列表
        llm, chat, embed = get_models()
        self.llm_recognition = chat
        # 加载 意图识别系统配置信息
        self.intent_prompt = yaml_utils.get_intent_prompt()

    async def recognize_intent(self , trace_id: str , config: dict , input: str):
        """
        意图识别
        输入：用户输入的问题
        输出：识别出的意图，可选项：
        - cattle_farm_common
        - cattle_farm_query
        - cattle_farm_warning_plan
        """
        session_id = config.sessionId
        use_history_recognize_intent = config_handler.get("lru", "USE_RECOGNIZE_INTENT_HISTORY_MESSAGES" , "False")
        server_logger.info(f"[使用用户最新历史记录作为意图识别]use_history_recognize_intent: {use_history_recognize_intent}")
        history = "无"
        if use_history_recognize_intent == "True":
            # 上下文管理器
            session_context_memory_manager = SessionContextMemoryManager(trace_id , session_id)
            # 获取内存最新的多少条历史记录(将消息列表序列化为字符串)
            history = await session_context_memory_manager.get_memory_last_history_str()
        # 根据历史记录和用户问题进行识别意图
        return self.recognize_intent_history(input=input , history=history)


    def recognize_intent_history(self , input: str , history="无"):
        """
        意图识别
        输入：用户输入的问题
        输出：识别出的意图，可选项：
        - cattle_farm_common
        - cattle_farm_query
        - cattle_farm_warning_plan
        """
        # 准备few-shot样例
        examples = self.intent_prompt["intent_examples"]
        #server_logger.info(f"加载prompt配置.examples: {examples}")
        system_prompt = self.intent_prompt["system_prompt"]
        system_prompt = system_prompt.format(history=history)
        server_logger.info(f"增加用户历史记录，用于意图识别，prompt配置.system_prompt: {system_prompt}")

        # 定义样本模板
        examples_prompt = ChatPromptTemplate.from_messages(
            [
                ("human", "{inn}"),
                ("ai", "{out}"),
            ]
        )
        few_shot_prompt = FewShotChatMessagePromptTemplate(example_prompt=examples_prompt,
                                                           examples=examples)
        final_prompt = ChatPromptTemplate.from_messages(
            [
                ('system', system_prompt),
                few_shot_prompt,
                ('human', '{input}'),
            ]
        )

        chain = final_prompt | self.llm_recognition
        server_logger.info(f"意图识别输入input: {input}")
        result = chain.invoke(input={"input": input})
        # 容错处理
        if hasattr(result, 'content'):
            # 如果 result 有 content 属性，使用它
            return result.content
        else:
            # 否则，直接返回 result
            return result





intent_identify_client = IntentIdentifyClient()


if __name__ == '__main__':
    #input = "你好"
    #input = "我要查询信息"  #result=cattle_farm_query
    input = "查询最近步数情况" #result=cattle_farm_query
   
    
    input = "01、10"
    input = "继续"
    input = "当前没有"
    #input = "查询12步数"
    #input = "分析当前环境数据"
    result = intent_identify_client.recognize_intent_history(history=history , input=input)
    server_logger.info(f"result={result}")