Simon
473 lines
17 KiB
Python
473 lines
17 KiB
Python
# =============================================================================
|
||
# 企微IT智能服务台 — 业务路由推荐服务
|
||
# =============================================================================
|
||
# 说明：核心路由逻辑，在 H5 后台 AI 任务中拦截非IT业务消息，
|
||
# 调用 Dify 统一意图识别，判定业务类别后发送对应联系人名片卡片。
|
||
#
|
||
# 主要职责：
|
||
# 1. 关键词预过滤（ROUTING_PREFILTER_KEYWORDS）— 快速过滤非路由消息
|
||
# 2. Dify 统一意图识别 — 调用 /v1/chat-messages，解析 intent_type/business_category/routing_confidence
|
||
# 3. 联系人查询 — 按 business_category 查 business_contacts 表
|
||
# 4. 名片三段式发送 — 路由文本 → contact_card → 系统提示（WS双通道推送）
|
||
# 5. 路由事件记录（P1）— 记录路由命中统计
|
||
#
|
||
# 设计决策：
|
||
# - 路由检测放在后台任务而非前端调用，与 BYOD 卡片处理模式一致
|
||
# - 关键词预过滤与审批预过滤可能重叠，Dify Prompt 内部判断优先级确保正确分流
|
||
# - routing_confidence < 0.7 不触发名片推荐，走正常 AI 回复流程
|
||
# =============================================================================
|
||
|
||
import json
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Any, Optional
|
||
|
||
import httpx
|
||
from sqlalchemy import select
|
||
|
||
from app.config import settings
|
||
from app.models.business_contact import BusinessContact
|
||
from app.models.conversation import Conversation
|
||
from app.models.message import Message
|
||
from app.models.routing_event import RoutingEvent
|
||
from app.services.ws_manager import manager as ws_manager
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# =============================================================================
|
||
# 共享 httpx 客户端（v4.0 P1-2：修复每次调用新建连接的泄漏问题）
|
||
# =============================================================================
|
||
_routing_client: Optional[httpx.AsyncClient] = None
|
||
|
||
|
||
async def _get_routing_client(timeout: float) -> httpx.AsyncClient:
|
||
"""获取共享的 httpx.AsyncClient（懒加载单例）。
|
||
|
||
为什么：之前每次 detect_routing_intent 调用都 `async with httpx.AsyncClient()`
|
||
新建连接池，高并发下产生大量 TIME_WAIT 连接（与 AIService 修复前同源问题）。
|
||
"""
|
||
global _routing_client
|
||
if _routing_client is None or _routing_client.is_closed:
|
||
_routing_client = httpx.AsyncClient(timeout=httpx.Timeout(timeout))
|
||
return _routing_client
|
||
|
||
|
||
# =============================================================================
|
||
# 路由关键词预过滤列表
|
||
# =============================================================================
|
||
# 说明：覆盖 5 个业务类别的关键词，用于快速过滤非路由消息。
|
||
# 只要命中任意一个关键词才值得调用 Dify 做精确判断。
|
||
# 关键词可能与审批预过滤重叠（如"办公用品"），Dify Prompt 内部判断
|
||
# 优先级（先审批→再IT咨询→再非IT路由）确保正确分流。
|
||
|
||
ROUTING_PREFILTER_KEYWORDS: list[str] = [
|
||
# 行政
|
||
"打印机", "复印机", "扫描仪", "保洁", "名片印刷",
|
||
# 人力资源
|
||
"工牌", "考勤", "入职", "离职", "社保", "公积金",
|
||
# 财务
|
||
"报销", "发票", "借款", "工资条",
|
||
# 法务
|
||
"合同", "法务", "知识产权",
|
||
# 行政-物业
|
||
"空调", "电梯", "门禁", "停车",
|
||
]
|
||
|
||
# 关键词到业务类别的映射（Dify 不可用时降级兜底用）
|
||
ROUTING_KEYWORD_TO_CATEGORY: dict[str, str] = {
|
||
# 行政
|
||
"打印机": "行政", "复印机": "行政", "扫描仪": "行政",
|
||
"保洁": "行政", "名片印刷": "行政",
|
||
# 人力资源
|
||
"工牌": "人力资源", "考勤": "人力资源", "入职": "人力资源",
|
||
"离职": "人力资源", "社保": "人力资源", "公积金": "人力资源",
|
||
# 财务
|
||
"报销": "财务", "发票": "财务", "借款": "财务", "工资条": "财务",
|
||
# 法务
|
||
"合同": "法务", "法务": "法务", "知识产权": "法务",
|
||
# 行政-物业
|
||
"空调": "行政-物业", "电梯": "行政-物业", "门禁": "行政-物业", "停车": "行政-物业",
|
||
}
|
||
|
||
|
||
# =============================================================================
|
||
# 预过滤 & 降级兜底
|
||
# =============================================================================
|
||
|
||
def routing_keyword_prefilter(text: str) -> bool:
|
||
"""路由关键词预过滤：检查文本是否包含非IT业务关键词。
|
||
|
||
只要命中任意一个路由关键词即返回 True，未命中返回 False。
|
||
用于在调用 Dify 前快速过滤，减少不必要的 API 调用。
|
||
|
||
Args:
|
||
text: 用户消息文本
|
||
|
||
Returns:
|
||
bool: 是否包含路由关键词
|
||
"""
|
||
if not text:
|
||
return False
|
||
return any(kw in text for kw in ROUTING_PREFILTER_KEYWORDS)
|
||
|
||
|
||
def _keyword_fallback_category(text: str) -> Optional[str]:
|
||
"""关键词降级兜底：Dify 不可用时通过关键词匹配业务类别。
|
||
|
||
遍历 ROUTING_KEYWORD_TO_CATEGORY 映射，命中第一个关键词即返回对应业务类别。
|
||
|
||
Args:
|
||
text: 用户消息文本
|
||
|
||
Returns:
|
||
Optional[str]: 业务类别（行政/人力资源/财务/法务/行政-物业），未命中返回 None
|
||
"""
|
||
if not text:
|
||
return None
|
||
for kw, category in ROUTING_KEYWORD_TO_CATEGORY.items():
|
||
if kw in text:
|
||
return category
|
||
return None
|
||
|
||
|
||
# =============================================================================
|
||
# Dify 统一意图识别调用
|
||
# =============================================================================
|
||
|
||
async def detect_routing_intent(text: str, employee_id: str = "") -> dict:
|
||
"""调用 Dify 统一意图识别，解析路由相关字段。
|
||
|
||
复用 approval.py 的 _call_dify_approval_intent 调用模式（Dify 原生 API），
|
||
但本函数独立维护，解析路由关心的字段：
|
||
- intent_type: approval/it_consult/non_it_routing/chitchat
|
||
- business_category: 行政/人力资源/财务/法务/行政-物业（仅 non_it_routing 时有值）
|
||
- routing_confidence: 0.0~1.0，≥0.7 触发名片推荐
|
||
|
||
使用与审批意图识别相同的 Dify 应用（同一 API Key），只是解析各自关心的字段。
|
||
|
||
Args:
|
||
text: 用户消息文本
|
||
employee_id: 员工 ID（可选，传给 Dify 的 user 字段）
|
||
|
||
Returns:
|
||
dict: {
|
||
"intent_type": str,
|
||
"business_category": str | None,
|
||
"routing_confidence": float,
|
||
"is_approval_request": bool,
|
||
"confidence": float,
|
||
"approval_type": str | None,
|
||
}
|
||
|
||
Raises:
|
||
Exception: Dify 调用失败或响应解析失败
|
||
"""
|
||
base_url = settings.approval_dify_base_url
|
||
api_key = settings.approval_dify_api_key
|
||
timeout = settings.approval_dify_timeout
|
||
|
||
if not base_url or not api_key:
|
||
raise ValueError(
|
||
"Dify 统一意图识别应用未配置"
|
||
"（APPROVAL_DIFY_BASE_URL / APPROVAL_DIFY_API_KEY）"
|
||
)
|
||
|
||
# 构建请求 URL：base_url + /v1/chat-messages（Dify 原生 API）
|
||
url = f"{base_url.rstrip('/')}/v1/chat-messages"
|
||
|
||
body = {
|
||
"inputs": {},
|
||
"query": text,
|
||
"response_mode": "blocking",
|
||
"user": employee_id or "routing_detection",
|
||
}
|
||
headers = {
|
||
"Authorization": f"Bearer {api_key}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
client = await _get_routing_client(timeout)
|
||
response = await client.post(url, json=body, headers=headers)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
# 解析 Dify 原生响应：answer 字段包含 AI 返回的 JSON 字符串
|
||
answer = data.get("answer", "")
|
||
parsed = json.loads(answer)
|
||
|
||
# 解析统一意图识别的 6 个字段
|
||
return {
|
||
"is_approval_request": bool(parsed.get("is_approval_request", False)),
|
||
"confidence": float(parsed.get("confidence", 0.0)),
|
||
"approval_type": parsed.get("approval_type"),
|
||
"intent_type": str(parsed.get("intent_type", "chitchat")),
|
||
"business_category": parsed.get("business_category"),
|
||
"routing_confidence": float(parsed.get("routing_confidence", 0.0)),
|
||
}
|
||
|
||
|
||
# =============================================================================
|
||
# 联系人查询
|
||
# =============================================================================
|
||
|
||
async def get_contact_by_category(db, category: str) -> Optional[BusinessContact]:
|
||
"""按业务类别查询联系人。
|
||
|
||
按 business_category + is_active=True 查询，取第一条有效联系人。
|
||
P0 阶段单联系人推荐，P2 支持按服务区域匹配。
|
||
|
||
Args:
|
||
db: 异步 DB session
|
||
category: 业务类别（行政/人力资源/财务/法务/行政-物业）
|
||
|
||
Returns:
|
||
Optional[BusinessContact]: 联系人对象，未找到返回 None
|
||
"""
|
||
result = await db.execute(
|
||
select(BusinessContact)
|
||
.where(
|
||
BusinessContact.business_category == category,
|
||
BusinessContact.is_active == True, # noqa: E712
|
||
)
|
||
.order_by(BusinessContact.id)
|
||
.limit(1)
|
||
)
|
||
return result.scalar_one_or_none()
|
||
|
||
|
||
# =============================================================================
|
||
# 名片三段式发送
|
||
# =============================================================================
|
||
|
||
async def send_contact_card(
|
||
db,
|
||
conversation: Conversation,
|
||
employee_id: str,
|
||
contact: BusinessContact,
|
||
reason: str,
|
||
business_category: str,
|
||
routing_confidence: float,
|
||
) -> None:
|
||
"""发送名片三段式消息（路由文本 → contact_card → 系统提示）。
|
||
|
||
完全参考 _handle_byod_query 模式：
|
||
1. 创建路由说明文本消息（AI, text）→ 落库 + WS双通道推送
|
||
2. 创建 contact_card 名片消息（AI, contact_card）→ 落库 + WS双通道推送
|
||
3. 创建系统提示消息（system, system）→ 落库 + WS双通道推送
|
||
|
||
每条消息分别落库 + WS推送，与 PRD 4.3 交互流程一致。
|
||
|
||
Args:
|
||
db: 异步 DB session
|
||
conversation: 当前会话对象
|
||
employee_id: 员工企微 UserID
|
||
contact: 联系人对象
|
||
reason: 路由说明文本（如"打印机问题属于行政设备范畴..."）
|
||
business_category: 业务类别
|
||
routing_confidence: 路由置信度
|
||
"""
|
||
contact_data = contact.to_dict()
|
||
extra_data: dict[str, Any] = {
|
||
"contact": contact_data,
|
||
"routing_reason": reason,
|
||
"business_category": business_category,
|
||
"routing_confidence": routing_confidence,
|
||
}
|
||
|
||
# === 1. 路由说明文本消息 ===
|
||
routing_text_msg = Message(
|
||
conversation_id=conversation.id,
|
||
sender_type="ai",
|
||
sender_id="ai_bot",
|
||
sender_name="Duckula（达寇拉）",
|
||
content=reason,
|
||
msg_type="text",
|
||
is_read=True,
|
||
)
|
||
db.add(routing_text_msg)
|
||
await db.flush()
|
||
|
||
await ws_manager.broadcast_to_employees([employee_id], {
|
||
"type": "ai_reply",
|
||
"data": {
|
||
"message_id": str(routing_text_msg.id),
|
||
"conversation_id": str(conversation.id),
|
||
"sender_type": "ai",
|
||
"sender_id": "ai_bot",
|
||
"sender_name": "Duckula（达寇拉）",
|
||
"content": reason,
|
||
"msg_type": "text",
|
||
"is_guidance": False,
|
||
"ai_reply_count": conversation.ai_substantive_reply_count,
|
||
"can_call_agent": conversation.ai_substantive_reply_count >= 3,
|
||
"conversation_status": conversation.status,
|
||
},
|
||
})
|
||
try:
|
||
await ws_manager.broadcast({
|
||
"type": "new_message",
|
||
"data": {
|
||
"conversation_id": str(conversation.id),
|
||
"message_id": str(routing_text_msg.id),
|
||
"sender_type": "ai",
|
||
"sender_id": "ai_bot",
|
||
"sender_name": "Duckula（达寇拉）",
|
||
"content": reason,
|
||
"msg_type": "text",
|
||
},
|
||
})
|
||
except Exception as ws_err:
|
||
logger.warning(f"路由文本 WS 广播给坐席失败: {ws_err}")
|
||
|
||
# === 2. contact_card 名片消息 ===
|
||
contact_card_msg = Message(
|
||
conversation_id=conversation.id,
|
||
sender_type="ai",
|
||
sender_id="ai_bot",
|
||
sender_name="Duckula（达寇拉）",
|
||
content=f"为您推荐{business_category}服务联系人：{contact.name}",
|
||
msg_type="contact_card",
|
||
extra_data=extra_data,
|
||
is_read=True,
|
||
)
|
||
db.add(contact_card_msg)
|
||
await db.flush()
|
||
|
||
await ws_manager.broadcast_to_employees([employee_id], {
|
||
"type": "ai_reply",
|
||
"data": {
|
||
"message_id": str(contact_card_msg.id),
|
||
"conversation_id": str(conversation.id),
|
||
"sender_type": "ai",
|
||
"sender_id": "ai_bot",
|
||
"sender_name": "Duckula（达寇拉）",
|
||
"content": f"为您推荐{business_category}服务联系人：{contact.name}",
|
||
"msg_type": "contact_card",
|
||
"extra_data": extra_data,
|
||
"is_guidance": False,
|
||
"ai_reply_count": conversation.ai_substantive_reply_count,
|
||
"can_call_agent": conversation.ai_substantive_reply_count >= 3,
|
||
"conversation_status": conversation.status,
|
||
},
|
||
})
|
||
try:
|
||
await ws_manager.broadcast({
|
||
"type": "new_message",
|
||
"data": {
|
||
"conversation_id": str(conversation.id),
|
||
"message_id": str(contact_card_msg.id),
|
||
"sender_type": "ai",
|
||
"sender_id": "ai_bot",
|
||
"sender_name": "Duckula（达寇拉）",
|
||
"content": f"为您推荐{business_category}服务联系人：{contact.name}",
|
||
"msg_type": "contact_card",
|
||
"extra_data": extra_data,
|
||
},
|
||
})
|
||
except Exception as ws_err:
|
||
logger.warning(f"名片卡片 WS 广播给坐席失败: {ws_err}")
|
||
|
||
# === 3. 系统提示消息 ===
|
||
system_text = "以上为AI自动推荐，点击名片可直接发起企微聊天"
|
||
system_msg = Message(
|
||
conversation_id=conversation.id,
|
||
sender_type="system",
|
||
sender_id="system",
|
||
sender_name="系统",
|
||
content=system_text,
|
||
msg_type="system",
|
||
is_read=True,
|
||
)
|
||
db.add(system_msg)
|
||
await db.flush()
|
||
|
||
await ws_manager.broadcast_to_employees([employee_id], {
|
||
"type": "ai_reply",
|
||
"data": {
|
||
"message_id": str(system_msg.id),
|
||
"conversation_id": str(conversation.id),
|
||
"sender_type": "system",
|
||
"sender_id": "system",
|
||
"sender_name": "系统",
|
||
"content": system_text,
|
||
"msg_type": "system",
|
||
"is_guidance": False,
|
||
"ai_reply_count": conversation.ai_substantive_reply_count,
|
||
"can_call_agent": conversation.ai_substantive_reply_count >= 3,
|
||
"conversation_status": conversation.status,
|
||
},
|
||
})
|
||
try:
|
||
await ws_manager.broadcast({
|
||
"type": "new_message",
|
||
"data": {
|
||
"conversation_id": str(conversation.id),
|
||
"message_id": str(system_msg.id),
|
||
"sender_type": "system",
|
||
"sender_id": "system",
|
||
"sender_name": "系统",
|
||
"content": system_text,
|
||
"msg_type": "system",
|
||
},
|
||
})
|
||
except Exception as ws_err:
|
||
logger.warning(f"系统提示 WS 广播给坐席失败: {ws_err}")
|
||
|
||
# 更新会话状态（路由推荐视为一次实质性 AI 回复）
|
||
conversation.ai_substantive_reply_count += 1
|
||
conversation.updated_at = datetime.now()
|
||
db.add(conversation)
|
||
await db.flush()
|
||
await db.commit()
|
||
|
||
logger.info(
|
||
f"路由名片发送完成: employee_id={employee_id}, category={business_category}, "
|
||
f"contact={contact.name}, confidence={routing_confidence}"
|
||
)
|
||
|
||
|
||
# =============================================================================
|
||
# 路由事件记录（P1）
|
||
# =============================================================================
|
||
|
||
async def record_routing_event(
|
||
db,
|
||
conversation_id: str,
|
||
employee_id: str,
|
||
message_content: str,
|
||
business_category: str,
|
||
routing_confidence: float,
|
||
contact: Optional[BusinessContact],
|
||
) -> None:
|
||
"""记录路由命中事件（P1）。
|
||
|
||
为后续优化 Prompt 准确率、分析高频非IT业务提供数据支撑。
|
||
|
||
Args:
|
||
db: 异步 DB session
|
||
conversation_id: 会话ID
|
||
employee_id: 员工ID
|
||
message_content: 触发路由的员工消息（截断至500字）
|
||
business_category: 业务类别
|
||
routing_confidence: 路由置信度
|
||
contact: 推荐的联系人对象（可能为 None）
|
||
"""
|
||
try:
|
||
event = RoutingEvent(
|
||
conversation_id=conversation_id,
|
||
employee_id=employee_id,
|
||
message_content=message_content[:500],
|
||
business_category=business_category,
|
||
routing_confidence=routing_confidence,
|
||
contact_id=contact.id if contact else None,
|
||
contact_name=contact.name if contact else "",
|
||
is_clicked=False,
|
||
)
|
||
db.add(event)
|
||
await db.flush()
|
||
await db.commit()
|
||
except Exception as e:
|
||
# 路由事件记录失败不影响主流程，仅记录 warning
|
||
logger.warning(f"路由事件记录失败: {e}")
|