455 lines
17 KiB
Python
455 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__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =============================================================================
|
|||
|
|
# 路由关键词预过滤列表
|
|||
|
|
# =============================================================================
|
|||
|
|
# 说明：覆盖 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",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async with httpx.AsyncClient(timeout=httpx.Timeout(timeout)) as client:
|
|||
|
|
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}")
|