diff --git a/backend/app/config.py b/backend/app/config.py index 7db6854..6785fc6 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -224,7 +224,7 @@ class Settings(BaseSettings): # Dify 审批意图识别应用 API Key approval_dify_api_key: str = "" # Dify 审批意图识别请求超时(秒) - approval_dify_timeout: int = 15 + approval_dify_timeout: int = 8 # v4.0 批次3 D1 止血:15s→8s(路由检测与主Dify串行叠加 15+30=45s → 8+30=38s) # 审批意图置信度阈值(≥ 此值才触发审批卡片) approval_confidence_threshold: float = 0.7 diff --git a/backend/app/tasks/h5_ai_task.py b/backend/app/tasks/h5_ai_task.py index 0d74183..c39c213 100644 --- a/backend/app/tasks/h5_ai_task.py +++ b/backend/app/tasks/h5_ai_task.py @@ -1002,6 +1002,254 @@ async def _push_asset_recommends( # 资产推荐失败不影响主对话流程 +# ============================================================================= +# 管线步骤函数(v4.0 批次 3:process_h5_ai_reply 管线化重构) +# ============================================================================= +# 设计:主函数从 200+ 行/11 对 try/except 收敛为 ~40 行编排代码, +# 每步一个函数,步骤内部自管异常,主函数零嵌套。 +# 行为承诺:与原 v3.2 实现外部行为一致(仅结构调整 + v3.0 降级结果补 type 字段)。 +# ============================================================================= + + +async def _step_load_conversation(db, conversation_id: str): + """步骤1:加载会话(重试 3 次,处理事务未提交竞态)。""" + for attempt in range(3): + conversation = await db.get(Conversation, conversation_id) + if conversation: + return conversation + if attempt < 2: + await asyncio.sleep(0.5) + # 刷新 session 以看到已提交的数据 + await db.rollback() + logger.warning(f"后台 AI 任务:会话不存在(重试3次后) {conversation_id}") + return None + + +async def _step_byod_intercept(db, conversation, employee_id, content, msg_type) -> bool: + """步骤2:BYOD 关键词拦截。命中返回 True(终止管线)。""" + if msg_type == "text" and _byod_keyword_prefilter(content): + await _handle_byod_query(db, conversation, employee_id, content) + return True + return False + + +async def _step_routing_intercept(db, conversation, employee_id, content, msg_type) -> bool: + """步骤3:非IT业务路由拦截。命中并发送名片返回 True(终止管线)。""" + if msg_type != "text" or not routing_keyword_prefilter(content): + return False + return await _handle_routing(db, conversation, employee_id, content) + + +async def _step_local_quick_reply(db, conversation, employee_id, content, msg_type, dify_conversation_id) -> bool: + """步骤4:本地快判断(打招呼/呼叫人工)。命中返回 True(终止管线)。""" + if msg_type != "text": + return False + ai_handler = get_shared_ai_handler() + if not (ai_handler.is_greeting(content) or ai_handler.is_call_human(content)): + return False + result = await ai_handler.handle_message( + content=content, + dify_conversation_id=dify_conversation_id, + user_id=employee_id, + ) + await _persist_and_push( + db, conversation, employee_id, result.content, + result.is_guidance, result.should_count, + result.should_transfer, result.dify_conversation_id, + ) + return True + + +async def _step_enrich_image(db, content, msg_type, media_url, conversation_id, employee_id) -> str: + """步骤5:图片消息增强(VisionService 分析 -> 视觉描述融合)。失败降级为原文。""" + if msg_type != "image" or not media_url: + return content + logger.info(f"图片消息检测: conversation={conversation_id}, media_url={media_url}") + try: + enriched = await _enrich_image_content( + db=db, media_url=media_url, original_content=content, + conversation_id=conversation_id, employee_id=employee_id, + ) + logger.info( + f"图片内容增强完成: original_len={len(content)}, enriched_len={len(enriched)}" + ) + return enriched + except Exception as vision_err: + logger.error(f"VisionService 处理失败,降级为纯文本: {vision_err}") + # 降级:使用原始 content,AI 会收到 "[图片] 截图" 这样的占位符 + return content + + +async def _step_graph_shortcut(db, conversation, employee_id, enriched_content, msg_type) -> bool: + """步骤6:Neo4j 图谱短路。命中返回 True(终止管线)。异常降级继续。""" + if msg_type != "text" or not enriched_content: + return False + try: + from app.services.graph_query_service import get_graph_query_service + from app.services.neo4j_client import get_neo4j_client + + neo4j_client = await get_neo4j_client() + if not neo4j_client: + return False + graph_service = await get_graph_query_service(neo4j_client) + solution = ( + await graph_service.find_solution_by_question(enriched_content) + if graph_service else None + ) + if not solution: + return False + logger.info( + f"图谱命中: question={enriched_content[:30]}, solution={solution.action_name}" + ) + # 直接返回图谱解决方案,跳过 Dify 调用 + await _persist_and_push_solution(db, conversation, employee_id, solution) + return True + except Exception as graph_err: + # 图谱查询失败不阻断,继续原有 Dify 流程 + import traceback + logger.warning(f"图谱查询异常(降级继续): {graph_err}\n{traceback.format_exc()}") + return False + + +def _build_keyword_fallback_result(matched_card, conversation, result=None, source="keyword_fallback"): + """构建关键词降级结果(v3.0 超时降级 / v3.1 无action降级共用)。 + + v4.0 批次 3 合并:统一补 "type": "approval_card"(v3.0 原缺此字段, + 前端 approve-direct-card 渲染依赖它,属于缺陷修复)。 + """ + options = matched_card.get("options") or [] + return { + "text": f"我来帮您提交{matched_card.get('title', '审批')},请点击下方卡片。", + "action": { + "type": "approval_card", + "card_data": matched_card, + "url": options[0].get("url", "") if options else "", + }, + "options": None, + "hit": True, + "conversation_id": (result or {}).get("conversation_id") or conversation.dify_conversation_id, + "is_structured": True, + "diagnosis_stage": "recommending" if result else None, + "response_time_ms": (result or {}).get("response_time_ms", 0), + "source": source, + } + + +async def _step_notify_failure(conversation_id: str, employee_id: str, message: str): + """步骤:统一失败通知(ai_reply_failed WS 推送)。""" + try: + await ws_manager.broadcast_to_employees([employee_id], { + "type": "ai_reply_failed", + "data": {"conversation_id": conversation_id, "message": message}, + }) + except Exception: + # 推送失败也无所谓,员工端 3 秒轮询兜底 + pass + + +async def _step_call_dify(db, conversation, employee_id, conversation_id, content, enriched_content, dify_conversation_id): + """步骤7:Dify 主推理(thinking 指示器 + 30s 超时 + 关键词降级)。 + + 返回结构化结果 dict; + 超时且降级失败时内部推送 ai_reply_failed 并返回 None(终止管线); + 超时但降级成功时已推送降级卡片,返回 None(终止管线)。 + """ + # 1. 推送 "正在思考..." 指示器(员工气泡动画 + 坐席状态指示) + try: + await ws_manager.broadcast_to_employees([employee_id], { + "type": "ai_thinking", + "data": {"conversation_id": conversation_id}, + }) + except Exception as thinking_err: + logger.warning(f"推送 AI 思考指示器失败: {thinking_err}") + try: + await ws_manager.broadcast({ + "type": "ai_thinking", + "data": {"conversation_id": conversation_id, "employee_id": employee_id}, + }) + except Exception: + pass # 坐席端通知失败不影响主流程 + + # 2. 启动延迟 "仍在思考" 后台任务(15 秒后触发,Dify 返回则取消) + async def _push_still_thinking(): + await asyncio.sleep(15) + await ws_manager.broadcast_to_employees([employee_id], { + "type": "ai_thinking", + "data": {"conversation_id": conversation_id, "status": "still_thinking"}, + }) + + thinking_task = asyncio.create_task(_push_still_thinking()) + + # 3. 调用 Dify(blocking + 30s 硬超时) + ai_handler = get_shared_ai_handler() + try: + result = await asyncio.wait_for( + ai_handler.ai_service.get_structured_reply( + message=enriched_content, + conversation_id=dify_conversation_id or conversation.dify_conversation_id, + user_id=employee_id, + ), + timeout=30, + ) + except asyncio.TimeoutError: + # v3.0: Dify 超时 -> 关键词降级匹配 + thinking_task.cancel() + logger.warning(f"Dify 30 秒超时: conversation={conversation_id},尝试关键词降级") + + from app.services.approval_matcher import get_approval_matcher + matched_card = get_approval_matcher().match_by_keywords(content) + + if matched_card: + logger.info(f"[Fallback] 关键词降级成功: {matched_card.get('title')}") + fallback_result = _build_keyword_fallback_result(matched_card, conversation) + try: + await _persist_and_push_structured(db, conversation, employee_id, fallback_result) + return None # 已推送降级卡片,终止管线 + except Exception as fallback_err: + logger.error(f"[Fallback] 降级推送失败: {fallback_err}") + + # 降级也失败 -> 建议转人工 + await _step_notify_failure(conversation_id, employee_id, "AI 响应时间较长,建议转人工坐席处理。") + return None + + # 4. 取消 "仍在思考" 任务(Dify 已返回) + thinking_task.cancel() + try: + await thinking_task # 等待 task 真正取消,避免 warning + except asyncio.CancelledError: + pass + + # 5. v3.1: Dify 返回但无 action(如诊断 escalate)-> 关键词降级 + if not result.get("action"): + from app.services.approval_matcher import get_approval_matcher + matched_card = get_approval_matcher().match_by_keywords(content) + if matched_card: + logger.info(f"[Fallback-v3.1] Dify 无action但关键词命中: {matched_card.get('title')}") + result = _build_keyword_fallback_result(matched_card, conversation, result, "keyword_fallback_v3") + + return result + + +async def _step_persist(db, conversation, employee_id, result): + """步骤8:持久化 + 双 WS 推送(ai_reply + dynamic_recommend)。异常不中断管线。""" + try: + await _persist_and_push_structured(db, conversation, employee_id, result) + except Exception as persist_err: + logger.error(f"[Persist] AI回复持久化失败: {persist_err}", exc_info=True) + + +async def _step_assets(db, employee_id, content, result): + """步骤9:资产推荐推送(L1/L2/L3 分层,独立运维触达通道)。异常不中断管线。""" + try: + await _push_asset_recommends(db, employee_id, content, result) + except Exception as asset_err: + logger.error(f"[Asset] 资产推荐推送失败: {asset_err}", exc_info=True) + + +# ============================================================================= +# 管线编排主函数(v4.0 批次 3 重构版) +# ============================================================================= + async def process_h5_ai_reply( conversation_id: str, employee_id: str, @@ -1012,25 +1260,20 @@ async def process_h5_ai_reply( ): """H5 发送消息后的 AI 回复处理(asyncio.create_task 入口)。 - v2.0 改造(2026-07-13): - - AI 回复从流式 SSE 改为 blocking + JSON 结构化输出 - - Dify 返回 {text, action, options} JSON → 后端解析 → 双 WS 推送 - - 聊天气泡收到 ai_reply(text + options),侧边栏收到 dynamic_recommend(action) - - 新增 ai_thinking 指示器,用户发送后立即看到"正在思考..." - - v2.1 改造(2026-07-13 Phase 4): - - 新增图片消息处理分支(msg_type=image) - - 图片 → VisionService.analyze_screenshot() → 视觉描述 → 融合到用户文字 - - 消息融合:查询最近 5 秒内员工的文字消息,与图片描述合并后传给 Dify - - 降级:VisionService 失败/低置信度 → 使用原始文字或提示用户描述问题 + v4.0 批次 3 管线化重构:原 12 步/11 对 try/except 编排为 9 个步骤函数, + 主函数仅最外层 1 个 try/except,行为与 v3.2 外部表现一致。 流程: - 1. BYOD 关键词拦截 → byod_card 卡片 - 2. 路由关键词拦截 → 名片推荐 - 3. 本地快判断(打招呼/呼叫人工)→ 同步引导 - 4. ★ 图片消息处理(Phase 4A)→ VisionService 分析 → 内容增强 - 5. ★ 结构化 AI 回复(blocking + JSON 解析 + 双 WS 推送) - 6. 任意异常 → 推 ai_reply_failed + 1. _step_load_conversation 加载会话(重试3次) + 2. _step_byod_intercept BYOD 关键词拦截 + 3. _step_routing_intercept 非IT业务路由拦截(名片推荐) + 4. _step_local_quick_reply 本地快判断(打招呼/呼叫人工) + 5. _step_enrich_image 图片消息 VisionService 增强 + 5b. _enrich_with_last_ai_context 简短回复上下文拼接(v2.3) + 6. _step_graph_shortcut Neo4j 图谱短路 + 7. _step_call_dify Dify 主推理(含超时/无action关键词降级) + 8. _step_persist 持久化 + 双 WS 推送 + 9. _step_assets 资产推荐推送 Args: conversation_id: 会话 ID @@ -1040,277 +1283,46 @@ async def process_h5_ai_reply( msg_type: 消息类型(text/image/file),默认 text media_url: 媒体文件 URL(图片消息时使用) """ - ai_handler = get_shared_ai_handler() factory = _get_session_factory() async with factory() as db: try: - # 防御:会话刚创建时可能事务未提交,最多重试 3 次(每次 0.5s) - conversation = None - for attempt in range(3): - conversation = await db.get(Conversation, conversation_id) - if conversation: - break - if attempt < 2: - await asyncio.sleep(0.5) - # 刷新 session 以看到已提交的数据 - await db.rollback() - + conversation = await _step_load_conversation(db, conversation_id) if not conversation: - logger.warning(f"后台 AI 任务:会话不存在(重试3次后) {conversation_id}") return - # === BYOD 关键词拦截(仅文本消息)=== - # 图片消息的 content 是占位符(如 "[图片] 截图"),跳过关键词拦截 - if msg_type == "text" and _byod_keyword_prefilter(content): - await _handle_byod_query(db, conversation, employee_id, content) + # 前置拦截管线(任一命中即终止) + if await _step_byod_intercept(db, conversation, employee_id, content, msg_type): + return + if await _step_routing_intercept(db, conversation, employee_id, content, msg_type): + return + if await _step_local_quick_reply(db, conversation, employee_id, content, msg_type, dify_conversation_id): return - # === 业务路由检测(仅文本消息)=== - if msg_type == "text" and routing_keyword_prefilter(content): - routed = await _handle_routing(db, conversation, employee_id, content) - if routed: - return - - # === 本地快判断:打招呼 / 呼叫人工(仅文本消息)=== - if msg_type == "text" and (ai_handler.is_greeting(content) or ai_handler.is_call_human(content)): - result = await ai_handler.handle_message( - content=content, - dify_conversation_id=dify_conversation_id, - user_id=employee_id, - ) - await _persist_and_push( - db, conversation, employee_id, result.content, - result.is_guidance, result.should_count, - result.should_transfer, result.dify_conversation_id, - ) - return - - # === ★ v2.1 图片消息处理(Phase 4A/4B)=== - # 做什么:检测到图片消息 → 调用 VisionService 分析截图 → - # 将视觉描述与用户文字融合 → 传给 Dify 推理 - # 为什么:Dify 文本模型无法"看"图片,需要先将图片转为文字描述 - # 降级:VisionService 失败 → 使用原始 content 继续流程 - enriched_content = content # 默认使用原始内容 - if msg_type == "image" and media_url: - logger.info( - f"图片消息检测: conversation={conversation_id}, " - f"media_url={media_url}" - ) - try: - enriched_content = await _enrich_image_content( - db=db, - media_url=media_url, - original_content=content, - conversation_id=conversation_id, - employee_id=employee_id, - ) - logger.info( - f"图片内容增强完成: original_len={len(content)}, " - f"enriched_len={len(enriched_content)}" - ) - except Exception as vision_err: - logger.error( - f"VisionService 处理失败,降级为纯文本: {vision_err}" - ) - # 降级:使用原始 content,AI 会收到 "[图片] 截图" 这样的占位符 - # Dify 会回复"我收到了您的截图,请描述一下问题" - - # === ★ v2.3 临时修复:Dify 对话历史缺失,为简短回复拼接上下文 === - # 问题:Dify 工作流未配置「对话历史」节点,conversation_id 传递了但 LLM 看不到历史 - # 改进:查询最近 10 条消息(用户+AI)构建完整对话摘要,含用户原始问题 - # 触发:消息短(<=50字符)、无问号、无换行 → 拼接后传给 Dify - # 后续:Dify 工作流配置对话历史后可移除此修复 + # 内容增强管线 + enriched_content = await _step_enrich_image( + db, content, msg_type, media_url, conversation_id, employee_id, + ) enriched_content = await _enrich_with_last_ai_context( - db, conversation_id, enriched_content + db, conversation_id, enriched_content, ) - # === ★ Neo4j 知识图谱查询(新增 v3.0)=== - # 做什么:在调用 Dify 之前先查询知识图谱 - # 为什么:简单问题可以直接从图谱返回解决方案,响应更快 - # 效果:简单问题响应从 3-15秒 → 毫秒级 - logger.info(f"图谱检查: msg_type={msg_type}, content_len={len(enriched_content) if enriched_content else 0}") - if msg_type == "text" and enriched_content: - try: - from app.services.graph_query_service import get_graph_query_service - from app.services.neo4j_client import get_neo4j_client - - neo4j_client = await get_neo4j_client() - if neo4j_client: - graph_service = await get_graph_query_service(neo4j_client) - if graph_service: - solution = await graph_service.find_solution_by_question( - enriched_content - ) - else: - solution = None - if solution: - logger.info( - f"图谱命中: question={enriched_content[:30]}, " - f"solution={solution.action_name}" - ) - # 直接返回图谱解决方案,跳过 Dify 调用 - await _persist_and_push_solution( - db, conversation, employee_id, solution - ) - return - except Exception as graph_err: - # 图谱查询失败不阻断,继续原有 Dify 流程 - import traceback - logger.warning(f"图谱查询异常(降级继续): {graph_err}\n{traceback.format_exc()}") - - # === ★ v2.0 结构化 AI 回复(替代流式)=== - # 1. 立即推送 "正在思考..." 指示器 - # 同时推给员工(气泡动画)和坐席(状态指示) - try: - await ws_manager.broadcast_to_employees([employee_id], { - "type": "ai_thinking", - "data": { - "conversation_id": conversation_id, - }, - }) - except Exception as thinking_err: - logger.warning(f"推送 AI 思考指示器失败: {thinking_err}") - - # 坐席端也通知:AI 正在处理此会话的消息 - try: - await ws_manager.broadcast({ - "type": "ai_thinking", - "data": { - "conversation_id": conversation_id, - "employee_id": employee_id, - }, - }) - except Exception: - pass # 坐席端通知失败不影响主流程 - - # 2. 启动延迟 "仍在思考" 后台任务(15 秒后触发) - # 如果 Dify 在 15 秒内返回,此任务会被取消 - async def _push_still_thinking(): - """15 秒后推送 "仍在思考" 提示,缓解用户等待焦虑。""" - await asyncio.sleep(15) - await ws_manager.broadcast_to_employees([employee_id], { - "type": "ai_thinking", - "data": { - "conversation_id": conversation_id, - "status": "still_thinking", - }, - }) - - thinking_task = asyncio.create_task(_push_still_thinking()) - - # 3. 调用 Dify(blocking 模式 + JSON 解析 + 30 秒硬超时) - # get_structured_reply 内部处理 HTTP 错误和 JSON 解析失败 - # asyncio.wait_for 处理 30 秒硬超时 → 建议转人工 - # 注意:图片消息使用 enriched_content(视觉描述+用户文字融合) - try: - result = await asyncio.wait_for( - ai_handler.ai_service.get_structured_reply( - message=enriched_content, - conversation_id=dify_conversation_id or conversation.dify_conversation_id, - user_id=employee_id, - ), - timeout=30, - ) - except asyncio.TimeoutError: - # v3.0: Dify 超时 → 关键词降级匹配 - thinking_task.cancel() - logger.warning(f"Dify 30 秒超时: conversation={conversation_id},尝试关键词降级") - - from app.services.approval_matcher import get_approval_matcher - matcher = get_approval_matcher() - matched_card = matcher.match_by_keywords(content) - - if matched_card: - logger.info(f"[Fallback] 关键词降级成功: {matched_card.get('title')}") - fallback_result = { - "text": f"我来帮您提交{matched_card.get('title', '审批')},请点击下方卡片。", - "action": {"card_data": matched_card, "url": matched_card.get("options", [{}])[0].get("url", "")} if matched_card.get("options") else None, - "options": None, - "hit": True, - "conversation_id": conversation.dify_conversation_id, - "is_structured": True, - "diagnosis_stage": None, - "response_time_ms": 0, - "source": "keyword_fallback", - } - try: - await _persist_and_push_structured(db, conversation, employee_id, fallback_result) - return - except Exception as fallback_err: - logger.error(f"[Fallback] 降级推送失败: {fallback_err}") - - # 降级也失败 → 建议转人工 - await ws_manager.broadcast_to_employees([employee_id], { - "type": "ai_reply_failed", - "data": { - "conversation_id": conversation_id, - "message": "AI 响应时间较长,建议转人工坐席处理。", - }, - }) + # 图谱短路 + if await _step_graph_shortcut(db, conversation, employee_id, enriched_content, msg_type): return - # 4. 取消 "仍在思考" 任务(Dify 已返回) - thinking_task.cancel() - try: - await thinking_task # 等待 task 真正取消,避免 warning - except asyncio.CancelledError: - pass + # 主推理(含 thinking 指示器 + 超时/无action 降级) + result = await _step_call_dify( + db, conversation, employee_id, conversation_id, + content, enriched_content, dify_conversation_id, + ) + if result is None: + return # 降级路径已全部处理(超时转人工 或 已推送降级卡片) - # v3.1: Dify 返回但无 action(如诊断 escalate/回复"AI服务暂时不可用")→ 关键词降级 - # 这是对 v3.0 的补充:v3.0 只在 asyncio.TimeoutError 触发降级,但 Dify LLM 自身可能误判 - if not result.get("action"): - from app.services.approval_matcher import get_approval_matcher - matcher = get_approval_matcher() - matched_card = matcher.match_by_keywords(content) - if matched_card: - logger.info(f"[Fallback-v3.1] Dify 无action但关键词命中: {matched_card.get('title')}") - result = { - "text": f"我来帮您提交{matched_card.get('title', '审批')},请点击下方卡片。", - "action": { - "type": "approval_card", - "card_data": matched_card, - "url": matched_card.get("options", [{}])[0].get("url", "") if matched_card.get("options") else "", - }, - "options": None, - "hit": True, - "conversation_id": result.get("conversation_id") or conversation.dify_conversation_id, - "is_structured": True, - "diagnosis_stage": "recommending", - "response_time_ms": result.get("response_time_ms", 0), - "source": "keyword_fallback_v3", - } - - # 5. 持久化 + 双 WS 推送(ai_reply + dynamic_recommend) - # 添加单独异常处理,避免影响主流程 - try: - await _persist_and_push_structured( - db, conversation, employee_id, result, - ) - except Exception as persist_err: - logger.error(f"[Persist] AI回复持久化失败: {persist_err}", exc_info=True) - - # 6. v3.0 资产推荐推送(L1/L2/L3 分层) - # 独立于对话的运维触达通道 - try: - await _push_asset_recommends( - db, employee_id, content, result, - ) - except Exception as asset_err: - logger.error(f"[Asset] 资产推荐推送失败: {asset_err}", exc_info=True) + # 后置处理管线 + await _step_persist(db, conversation, employee_id, result) + await _step_assets(db, employee_id, content, result) except Exception as e: import traceback - # 记录完整的堆栈跟踪信息 - tb_str = traceback.format_exc() - logger.error(f"后台 AI 任务异常: {e}\n堆栈跟踪:\n{tb_str}", exc_info=True) - try: - await ws_manager.broadcast_to_employees([employee_id], { - "type": "ai_reply_failed", - "data": { - "conversation_id": conversation_id, - "message": "AI 服务异常,请转人工坐席或稍后重试。", - }, - }) - except Exception: - # 推送失败也无所谓,员工端 3 秒轮询兜底 - pass + logger.error(f"后台 AI 任务异常: {e}\n堆栈跟踪:\n{traceback.format_exc()}", exc_info=True) + await _step_notify_failure(conversation_id, employee_id, "AI 服务异常,请转人工坐席或稍后重试。")