ai_fwt_pro/main/xiaozhi-server/core/handle/intentHandler.py

import json
import uuid
import asyncio
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from core.connection import ConnectionHandler
from core.utils.dialogue import Message
from core.providers.tts.dto.dto import ContentType
from core.handle.helloHandle import checkWakeupWords
from plugins_func.register import Action, ActionResponse
from core.handle.sendAudioHandle import send_stt_message
from core.handle.reportHandle import enqueue_tool_report
from core.utils.util import remove_punctuation_and_length
from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType

TAG = __name__


async def handle_user_intent(conn: "ConnectionHandler", text):
    # 预处理输入文本，处理可能的JSON格式
    try:
        if text.strip().startswith("{") and text.strip().endswith("}"):
            parsed_data = json.loads(text)
            if isinstance(parsed_data, dict) and "content" in parsed_data:
                text = parsed_data["content"]  # 提取content用于意图分析
                conn.current_speaker = parsed_data.get("speaker")  # 保留说话人信息
    except (json.JSONDecodeError, TypeError):
        pass

    # 检查是否有明确的退出命令
    _, filtered_text = remove_punctuation_and_length(text)
    if await check_direct_exit(conn, filtered_text):
        return True

    # 检查是否是唤醒词
    if await checkWakeupWords(conn, filtered_text):
        return True

    direct_function_call = match_direct_nav_command(filtered_text)
    if direct_function_call is not None:
        function_name, function_args = direct_function_call
        conn.logger.bind(tag=TAG).info(
            f"识别到直达导航命令: {filtered_text} -> {function_name}"
        )
        return await dispatch_function_call(
            conn,
            text,
            function_name,
            function_args,
        )

    if conn.intent_type == "function_call":
        # 使用支持function calling的聊天方法,不再进行意图分析
        return False
    # 使用LLM进行意图分析
    intent_result = await analyze_intent_with_llm(conn, text)
    if not intent_result:
        return False
    # 会话开始时生成sentence_id
    conn.sentence_id = str(uuid.uuid4().hex)
    # 处理各种意图
    return await process_intent_result(conn, intent_result, text)


async def check_direct_exit(conn: "ConnectionHandler", text):
    """检查是否有明确的退出命令"""
    _, text = remove_punctuation_and_length(text)
    cmd_exit = conn.cmd_exit
    for cmd in cmd_exit:
        if text == cmd:
            conn.logger.bind(tag=TAG).info(f"识别到明确的退出命令: {text}")
            await send_stt_message(conn, text)
            await conn.close()
            return True
    return False


async def analyze_intent_with_llm(conn: "ConnectionHandler", text):
    """使用LLM分析用户意图"""
    if not hasattr(conn, "intent") or not conn.intent:
        conn.logger.bind(tag=TAG).warning("意图识别服务未初始化")
        return None

    # 对话历史记录
    dialogue = conn.dialogue
    try:
        intent_result = await conn.intent.detect_intent(conn, dialogue.dialogue, text)
        return intent_result
    except Exception as e:
        conn.logger.bind(tag=TAG).error(f"意图识别失败: {str(e)}")

    return None


def match_direct_nav_command(filtered_text: str):
    text = (filtered_text or "").strip()
    if not text:
        return None

    if text in {
        "开始录制路线",
        "开始记录路线",
        "录制路线",
        "开始记录",
        "开始记路线",
    }:
        return "nav_start_record", {}

    if text in {
        "结束录制",
        "停止录制",
        "结束记录",
        "停止记录",
        "录制结束",
    }:
        return "nav_stop_record", {}

    if text in {
        "开始预警",
        "开始播报",
        "开始领航",
        "开始导航",
    }:
        return "nav_start_run", {}

    if text in {
        "停止预警",
        "结束预警",
        "停止播报",
        "关闭预警",
    }:
        return "nav_stop_run", {}

    if text.startswith("标记"):
        if any(word in text for word in ("急转弯", "起转弯", "急弯")):
            return "nav_mark_hazard", {"hazard_type": "sharp_turn"}
        if "障碍" in text:
            return "nav_mark_hazard", {"hazard_type": "obstacle"}
        if any(word in text for word in ("大坑", "坑")):
            return "nav_mark_hazard", {"hazard_type": "bump"}
        if "跳台" in text:
            return "nav_mark_hazard", {"hazard_type": "jump"}
        if "陡坡" in text:
            return "nav_mark_hazard", {"hazard_type": "slope"}
        if "涉水" in text:
            return "nav_mark_hazard", {"hazard_type": "water"}
        return "nav_mark_hazard", {"hazard_type": "other"}

    return None


async def dispatch_function_call(
    conn: "ConnectionHandler",
    original_text: str,
    function_name: str,
    function_args: dict | None = None,
):
    conn.sentence_id = str(uuid.uuid4().hex)
    function_args = function_args or {}
    function_args_json = json.dumps(function_args, ensure_ascii=False)
    function_call_data = {
        "name": function_name,
        "id": str(uuid.uuid4().hex),
        "arguments": function_args_json,
    }

    await send_stt_message(conn, original_text)
    conn.client_abort = False
    enqueue_tool_report(conn, function_name, function_args)

    def process_function_call():
        conn.dialogue.put(Message(role="user", content=original_text))

        tool_call_timeout = int(conn.config.get("tool_call_timeout", 30))
        try:
            result = asyncio.run_coroutine_threadsafe(
                conn.func_handler.handle_llm_function_call(conn, function_call_data),
                conn.loop,
            ).result(timeout=tool_call_timeout)
        except Exception as e:
            conn.logger.bind(tag=TAG).error(f"工具调用失败: {e}")
            result = ActionResponse(
                action=Action.ERROR,
                result="工具调用超时，请一会再试下哈",
                response="工具调用超时，请一会再试下哈",
            )

        if result:
            enqueue_tool_report(
                conn,
                function_name,
                function_args,
                str(result.result) if result.result else None,
                report_tool_call=False,
            )

            if result.action == Action.RESPONSE:
                text = result.response
                if text is not None:
                    speak_txt(conn, text)
            elif result.action == Action.REQLLM:
                text = result.result
                conn.dialogue.put(Message(role="tool", content=text))
                llm_result = conn.intent.replyResult(text, original_text)
                if llm_result is None:
                    llm_result = text
                speak_txt(conn, llm_result)
            elif result.action == Action.NOTFOUND or result.action == Action.ERROR:
                text = result.response if result.response else result.result
                if text is not None:
                    speak_txt(conn, text)
            elif function_name != "play_music":
                text = result.response
                if text is None:
                    text = result.result
                if text is not None:
                    speak_txt(conn, text)

    conn.executor.submit(process_function_call)
    return True


async def process_intent_result(
    conn: "ConnectionHandler", intent_result, original_text
):
    """处理意图识别结果"""
    try:
        # 尝试将结果解析为JSON
        intent_data = json.loads(intent_result)

        # 检查是否有function_call
        if "function_call" in intent_data:
            # 直接从意图识别获取了function_call
            conn.logger.bind(tag=TAG).debug(
                f"检测到function_call格式的意图结果: {intent_data['function_call']['name']}"
            )
            function_name = intent_data["function_call"]["name"]
            if function_name == "continue_chat":
                return False

            if function_name == "result_for_context":
                await send_stt_message(conn, original_text)
                conn.client_abort = False

                def process_context_result():
                    conn.dialogue.put(Message(role="user", content=original_text))

                    from core.utils.current_time import get_current_time_info

                    current_time, today_date, today_weekday, lunar_date = (
                        get_current_time_info()
                    )

                    # 构建带上下文的基础提示
                    context_prompt = f"""当前时间：{current_time}
                                        今天日期：{today_date} ({today_weekday})
                                        今天农历：{lunar_date}

                                        请根据以上信息回答用户的问题：{original_text}"""

                    response = conn.intent.replyResult(context_prompt, original_text)
                    speak_txt(conn, response)

                conn.executor.submit(process_context_result)
                return True

            function_args = {}
            if "arguments" in intent_data["function_call"]:
                function_args = intent_data["function_call"]["arguments"]
                if function_args is None:
                    function_args = {}
            if isinstance(function_args, str):
                function_args = json.loads(function_args) if function_args else {}

            return await dispatch_function_call(
                conn,
                original_text,
                function_name,
                function_args if isinstance(function_args, dict) else {},
            )
        return False
    except json.JSONDecodeError as e:
        conn.logger.bind(tag=TAG).error(f"处理意图结果时出错: {e}")
        return False


def speak_txt(conn: "ConnectionHandler", text):
    # 记录文本到 sentence_id 映射
    conn.tts.store_tts_text(conn.sentence_id, text)

    conn.tts.tts_text_queue.put(
        TTSMessageDTO(
            sentence_id=conn.sentence_id,
            sentence_type=SentenceType.FIRST,
            content_type=ContentType.ACTION,
        )
    )
    conn.tts.tts_one_sentence(conn, ContentType.TEXT, content_detail=text)
    conn.tts.tts_text_queue.put(
        TTSMessageDTO(
            sentence_id=conn.sentence_id,
            sentence_type=SentenceType.LAST,
            content_type=ContentType.ACTION,
        )
    )
    conn.dialogue.put(Message(role="assistant", content=text))