# ============================================================================= # 企微IT智能服务台 — 统一 AI 回复处理器 # ============================================================================= # 说明：统一封装 AI 调用逻辑，供 H5 端和企微回调端共用，确保两端行为一致。 # 1. 打招呼检测 → 引导用户描述问题（不计数） # 2. 呼叫人工拦截 → 引导先描述问题（不计数） # 3. AI 调用 → 命中则回复并计数，未命中则转人工 # 4. AI 异常降级 → 模板回复（不计数，不转人工） # # 为什么需要此模块： # - 原 h5.py 和 MessageRouter._try_ai_reply() 各自独立实现 AI 调用逻辑 # - 两端行为不一致：打招呼/呼叫人工拦截只在 H5 端有，计数规则不同 # - 统一后确保无论从哪个入口进来，用户获得的 AI 体验完全一致 # ============================================================================= import logging import random from dataclasses import dataclass from typing import Optional from app.services.ai_service import AIService logger = logging.getLogger(__name__) # -------------------------------------------------------------------------- # 打招呼关键词（匹配后 AI 引导用户描述问题，不计数） # -------------------------------------------------------------------------- _GREETING_KEYWORDS = [ "你好", "您好", "hi", "hello", "嗨", "在吗", "在不在", "哈喽", "早", "早上好", "下午好", "晚上好", ] # -------------------------------------------------------------------------- # 直接呼叫人工关键词（匹配后 AI 引导用户先描述问题，不计数） # -------------------------------------------------------------------------- _CALL_HUMAN_KEYWORDS = [ "人工", "人工坐席", "转人工", "客服", "我要人工", "找人工", "人工客服", "转接人工", "人工服务", "找客服", "联系人工", "我要找人", "不要机器人", "真人", "摇人", "找人", ] # AI 引导话术（打招呼） _GREETING_GUIDE = ( "你好！我是 IT 智能助手 🤖\n" "请直接描述你遇到的 IT 问题，比如：\n" "• 打印机连不上\n" "• 电脑蓝屏了\n" "• VPN 无法登录\n" "我先帮你分析，搞不定再帮你转人工坐席～" ) # AI 引导话术（呼叫人工） _CALL_HUMAN_GUIDE = ( "别急～先告诉我你遇到了什么问题？\n" "我先帮你排查一下，大部分问题我都能解决 💪\n" "如果确实需要人工坐席，我会帮你转接的！" ) # AI 未命中转人工话术 _AI_MISS_GUIDE = ( "🤖 AI 暂未学习到相关知识，正在为您转接 IT 坐席，请稍候..." ) # -------------------------------------------------------------------------- # 非文本消息自动回复模板（图片引导，其余类型暂不支持） # -------------------------------------------------------------------------- _IMAGE_REPLY = ( "收到您的截图 📷\n" "请补充文字描述您遇到的问题，以便更快为您处理。\n" "例如：\n" "• 这是什么软件的报错截图？\n" "• 您在操作什么时出现的？\n" "• 错误信息的具体内容是什么？" ) _NON_TEXT_REPLY_TEMPLATE = ( "暂不支持{type_name}消息 😅\n" "请用文字描述您的问题，我会尽快为您处理。" ) # AI 调用失败降级模板（使用 {topic} 占位符，运行时替换为用户消息摘要） _FALLBACK_TEMPLATES = [ "收到！关于「{topic}」的问题，让我来帮你分析一下…\n\n" "这类问题通常由以下原因导致：\n" "1. 网络连接异常\n" "2. 设备驱动问题\n" "3. 系统配置错误\n\n" "你可以先尝试重启设备，如果问题依旧，请告诉我具体的错误提示。", "明白，关于「{topic}」的问题…\n\n" "建议按以下步骤排查：\n" "1. 检查网络是否正常\n" "2. 确认相关服务是否启动\n" "3. 查看是否有报错提示\n\n" "如果以上步骤无法解决，请补充更多细节。", ] @dataclass class AIReplyResult: """AI 回复结果（统一返回结构）。 无论消息经过哪种处理路径（打招呼/呼叫人工/AI命中/AI未命中/降级）， 都返回此结构，由调用方决定如何持久化和发送。 Attributes: content: 回复内容 reply_type: 回复类型 - "greeting": 打招呼引导 - "call_human": 呼叫人工拦截引导 - "ai_hit": AI 命中知识库 - "ai_miss": AI 未命中，需转人工 - "ai_fallback": AI 调用异常，降级模板回复 is_guidance: 是否为引导类消息（打招呼或呼叫人工），前端据此决定 UI 展示 should_count: 是否应增加 ai_substantive_reply_count（仅 AI 命中时为 True） should_transfer: 是否应转人工（状态改为 queued） dify_conversation_id: Dify 会话ID（用于多轮对话上下文，AI 命中/未命中时更新） """ content: str reply_type: str is_guidance: bool = False should_count: bool = False should_transfer: bool = False dify_conversation_id: Optional[str] = None class AIHandler: """统一 AI 回复处理器。 封装打招呼检测、呼叫人工拦截、AI 调用、命中判断、计数规则、 转人工逻辑，供 H5 端和企微回调端（MessageRouter）共用， 确保两端行为完全一致。 处理流程（按优先级）： 1. 检测打招呼 → 返回引导话术（不计数，不转人工） 2. 检测呼叫人工 → 返回拦截引导（不计数，不转人工） 3. 调用 Dify API： - 命中 → 返回 AI 回复（计数+1，不转人工） - 未命中 → 返回转人工提示（不计数，转人工） 4. AI 调用异常 → 返回降级模板回复（不计数，不转人工） 计数规则（统一）： - 仅 AI 命中知识库时 ai_substantive_reply_count +1 - 打招呼/呼叫人工/未命中/降级 均不计数 """ def __init__(self, ai_service: AIService): """初始化 AI 处理器。 Args: ai_service: AI 服务实例（Dify API 封装），通常为应用级共享单例 """ self.ai_service = ai_service def is_greeting(self, content: str) -> bool: """检测是否为打招呼消息。 匹配规则：用户消息（小写化、去前后空格后）包含任意打招呼关键词。 Args: content: 用户消息内容 Returns: bool: 是否为打招呼 """ text = content.strip().lower() return any(kw in text for kw in _GREETING_KEYWORDS) def is_call_human(self, content: str) -> bool: """检测是否为直接呼叫人工。 匹配规则：用户消息（小写化、去前后空格后）包含任意呼叫人工关键词。 拦截后引导用户先描述问题，避免直接转人工浪费坐席资源。 Args: content: 用户消息内容 Returns: bool: 是否为呼叫人工 """ text = content.strip().lower() return any(kw in text for kw in _CALL_HUMAN_KEYWORDS) async def handle_message( self, content: str, dify_conversation_id: Optional[str] = None, user_id: Optional[str] = None, ) -> AIReplyResult: """处理用户消息，返回统一的 AI 回复结果。 按照优先级依次检测：打招呼 → 呼叫人工 → AI 调用。 每种路径返回不同的 reply_type，由调用方根据结果更新会话状态和计数。 Args: content: 用户消息内容 dify_conversation_id: Dify 会话ID（用于多轮对话上下文） user_id: 用户标识（用于 Dify 日志追溯） Returns: AIReplyResult: 统一的 AI 回复结果 """ # ================================================================== # 1. 检测打招呼 → 引导描述问题，不计数，不转人工 # ================================================================== if self.is_greeting(content): logger.info(f"打招呼引导: user_id={user_id}") return AIReplyResult( content=_GREETING_GUIDE, reply_type="greeting", is_guidance=True, should_count=False, should_transfer=False, dify_conversation_id=dify_conversation_id, ) # ================================================================== # 2. 检测呼叫人工 → 拦截引导，不计数，不转人工 # ================================================================== if self.is_call_human(content): logger.info(f"人工拦截引导: user_id={user_id}") return AIReplyResult( content=_CALL_HUMAN_GUIDE, reply_type="call_human", is_guidance=True, should_count=False, should_transfer=False, dify_conversation_id=dify_conversation_id, ) # ================================================================== # 3. 调用 Dify API 获取 AI 回复 # ================================================================== try: ai_result = await self.ai_service.get_reply( message=content, conversation_id=dify_conversation_id, user_id=user_id, ) # 提取 Dify 返回的 conversation_id（用于多轮对话上下文） new_conv_id = ai_result.get("conversation_id") or dify_conversation_id if ai_result["hit"]: # AI 命中：使用 Dify 回复，计数+1 logger.info( f"AI命中: user_id={user_id}, " f"content_length={len(ai_result['content'])}" ) return AIReplyResult( content=ai_result["content"], reply_type="ai_hit", is_guidance=False, should_count=True, should_transfer=False, dify_conversation_id=new_conv_id, ) else: # AI 未命中：转人工 logger.info(f"AI未命中转人工: user_id={user_id}") return AIReplyResult( content=_AI_MISS_GUIDE, reply_type="ai_miss", is_guidance=False, should_count=False, should_transfer=True, dify_conversation_id=new_conv_id, ) except Exception as e: # ============================================================== # 4. AI 调用异常：降级模板回复 # - 不计数（修复原 h5.py 降级误计数的 Bug） # - 不转人工（降级是临时故障，用户可继续尝试） # ============================================================== logger.error(f"AI调用失败（降级模板回复）: {e}") topic = content.strip()[:15] fallback_content = random.choice(_FALLBACK_TEMPLATES).format(topic=topic) return AIReplyResult( content=fallback_content, reply_type="ai_fallback", is_guidance=False, should_count=False, should_transfer=False, dify_conversation_id=dify_conversation_id, )