M2LOrder WebUI定制开发：嵌入企业微信/钉钉机器人实现情感告警推送
M2LOrder WebUI定制开发：嵌入企业微信/钉钉机器人实现情感告警推送
1. 项目概述与需求背景
M2LOrder是一个基于.opt模型文件的情绪识别与情感分析服务，提供HTTP API和WebUI两种访问方式。在实际企业应用中，单纯的情感分析结果展示往往不够，我们需要将分析结果实时推送到日常办公平台，让团队第一时间获取重要情感反馈。
本文将介绍如何在M2LOrder WebUI基础上进行定制开发，集成企业微信和钉钉机器人，实现情感告警自动推送功能。当系统检测到强烈负面情绪或特定情感模式时，能够立即通过机器人通知相关人员，便于快速响应和处理。
核心价值：
- 实时情感监控：7×24小时不间断情感分析
- 智能告警推送：基于阈值触发自动通知
- 多平台支持：同时兼容企业微信和钉钉
- 无缝集成：在原有WebUI基础上增强功能
2. 环境准备与基础配置
2.1 系统环境确认
首先确认M2LOrder服务正常运行：
# 进入项目目录 cd /root/m2lorder # 检查服务状态 supervisorctl -c supervisor/supervisord.conf status # 预期输出 m2lorder-api RUNNING pid 1234, uptime 0:10:30 m2lorder-webui RUNNING pid 1235, uptime 0:10:30 2.2 安装必要的依赖包
在原有依赖基础上，添加机器人推送相关库：
# 激活conda环境 source /opt/miniconda3/etc/profile.d/conda.sh conda activate torch28 # 安装额外依赖 pip install requests python-dotenv loguru 2.3 配置文件增强
修改/root/m2lorder/config/settings.py，添加机器人配置项：
# 机器人配置 ROBOT_CONFIG = { "enabled": True, "alert_threshold": 0.8, # 置信度阈值 "target_emotions": ["angry", "anxious", "sad"], # 需要告警的情感类型 # 企业微信机器人配置 "wecom_webhook": os.getenv("WECOM_WEBHOOK", ""), "wecom_enabled": os.getenv("WECOM_ENABLED", "false").lower() == "true", # 钉钉机器人配置 "dingtalk_webhook": os.getenv("DINGTALK_WEBHOOK", ""), "dingtalk_secret": os.getenv("DINGTALK_SECRET", ""), "dingtalk_enabled": os.getenv("DINGTALK_ENABLED", "false").lower() == "true", # 通知模板 "message_template": { "wecom": "情感告警: {emotion}情绪检测\n置信度: {confidence:.2f}\n原文: {text}\n时间: {time}", "dingtalk": "情感告警: {emotion}情绪检测\\n置信度: {confidence:.2f}\\n原文: {text}\\n时间: {time}" } } 3. 机器人推送功能实现
3.1 创建机器人服务模块
在/root/m2lorder/app/core/目录下创建robot_notifier.py：
import requests import json import time import hmac import hashlib import base64 import urllib.parse from datetime import datetime from loguru import logger from typing import Dict, Optional from config.settings import ROBOT_CONFIG class RobotNotifier: def __init__(self): self.config = ROBOT_CONFIG self.enabled = self.config["enabled"] def should_alert(self, emotion: str, confidence: float) -> bool: """判断是否需要发送告警""" if not self.enabled: return False if emotion in self.config["target_emotions"] and confidence >= self.config["alert_threshold"]: return True return False def send_wecom_message(self, text: str, emotion: str, confidence: float) -> bool: """发送企业微信消息""" if not self.config["wecom_enabled"] or not self.config["wecom_webhook"]: return False try: message = self.config["message_template"]["wecom"].format( emotion=emotion, confidence=confidence, text=text[:100] + "..." if len(text) > 100 else text, time=datetime.now().strftime("%Y-%m-%d %H:%M:%S") ) payload = { "msgtype": "text", "text": { "content": message } } response = requests.post( self.config["wecom_webhook"], json=payload, timeout=5 ) if response.status_code == 200: logger.info(f"企业微信消息发送成功: {emotion} {confidence}") return True else: logger.error(f"企业微信消息发送失败: {response.status_code} {response.text}") return False except Exception as e: logger.error(f"企业微信消息发送异常: {str(e)}") return False def send_dingtalk_message(self, text: str, emotion: str, confidence: float) -> bool: """发送钉钉消息""" if not self.config["dingtalk_enabled"] or not self.config["dingtalk_webhook"]: return False try: # 生成签名（如果有secret） timestamp = str(round(time.time() * 1000)) secret = self.config["dingtalk_secret"] if secret: sign = self._generate_dingtalk_sign(timestamp, secret) webhook_url = f"{self.config['dingtalk_webhook']}×tamp={timestamp}&sign={sign}" else: webhook_url = self.config["dingtalk_webhook"] message = self.config["message_template"]["dingtalk"].format( emotion=emotion, confidence=confidence, text=text[:100] + "..." if len(text) > 100 else text, time=datetime.now().strftime("%Y-%m-%d %H:%M:%S") ) payload = { "msgtype": "text", "text": { "content": message } } response = requests.post( webhook_url, json=payload, timeout=5 ) if response.status_code == 200: result = response.json() if result.get("errcode") == 0: logger.info(f"钉钉消息发送成功: {emotion} {confidence}") return True else: logger.error(f"钉钉消息发送失败: {result}") return False else: logger.error(f"钉钉消息发送失败: {response.status_code}") return False except Exception as e: logger.error(f"钉钉消息发送异常: {str(e)}") return False def _generate_dingtalk_sign(self, timestamp: str, secret: str) -> str: """生成钉钉签名""" string_to_sign = f"{timestamp}\n{secret}" hmac_code = hmac.new( secret.encode('utf-8'), string_to_sign.encode('utf-8'), digestmod=hashlib.sha256 ).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) return sign def notify(self, text: str, emotion: str, confidence: float) -> bool: """发送通知""" if not self.should_alert(emotion, confidence): return False success = False # 并行发送到所有启用的平台 if self.config["wecom_enabled"]: success = self.send_wecom_message(text, emotion, confidence) or success if self.config["dingtalk_enabled"]: success = self.send_dingtalk_message(text, emotion, confidence) or success return success 3.2 集成到API服务
修改/root/m2lorder/app/api/main.py，集成机器人通知功能：
from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from app.core.model_manager import ModelManager from app.core.opt_parser import OptParser from app.core.robot_notifier import RobotNotifier # 新增导入 # 初始化机器人通知器 robot_notifier = RobotNotifier() app = FastAPI(title="M2LOrder Emotion API") # 在预测端点中添加通知逻辑 @app.post("/predict") async def predict_emotion(request: PredictionRequest): try: # 原有预测逻辑 result = model_manager.predict(model_id, input_data) # 新增：发送机器人通知 robot_notifier.notify( text=input_data, emotion=result["emotion"], confidence=result["confidence"] ) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # 在批量预测中也添加通知 @app.post("/predict/batch") async def batch_predict(request: BatchPredictionRequest): try: results = [] for input_text in request.inputs: result = model_manager.predict(request.model_id, input_text) # 对每个结果进行通知 robot_notifier.notify( text=input_text, emotion=result["emotion"], confidence=result["confidence"] ) results.append(result) return {"model_id": request.model_id, "predictions": results} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) 4. WebUI界面增强
4.1 添加机器人配置界面
修改/root/m2lorder/app/webui/main.py，在Gradio界面中添加配置选项：
import gradio as gr from app.core.robot_notifier import RobotNotifier from config.settings import ROBOT_CONFIG # 创建机器人配置界面 def create_robot_config_ui(): with gr.Accordion("🤖 机器人告警配置", open=False): with gr.Row(): alert_enabled = gr.Checkbox( label="启用情感告警", value=ROBOT_CONFIG["enabled"] ) alert_threshold = gr.Slider( minimum=0.5, maximum=1.0, value=ROBOT_CONFIG["alert_threshold"], label="告警置信度阈值" ) with gr.Row(): emotions = gr.CheckboxGroup( label="告警情感类型", choices=["angry", "anxious", "sad", "happy", "excited"], value=ROBOT_CONFIG["target_emotions"] ) with gr.Tab("企业微信"): wecom_enabled = gr.Checkbox( label="启用企业微信机器人", value=ROBOT_CONFIG["wecom_enabled"] ) wecom_webhook = gr.Textbox( label="Webhook地址", value=ROBOT_CONFIG["wecom_webhook"], type="password" ) with gr.Tab("钉钉"): dingtalk_enabled = gr.Checkbox( label="启用钉钉机器人", value=ROBOT_CONFIG["dingtalk_enabled"] ) dingtalk_webhook = gr.Textbox( label="Webhook地址", value=ROBOT_CONFIG["dingtalk_webhook"] ) dingtalk_secret = gr.Textbox( label="签名密钥(可选)", value=ROBOT_CONFIG["dingtalk_secret"], type="password" ) save_btn = gr.Button("💾 保存配置") return [ alert_enabled, alert_threshold, emotions, wecom_enabled, wecom_webhook, dingtalk_enabled, dingtalk_webhook, dingtalk_secret, save_btn ] # 保存配置的函数 def save_robot_config( alert_enabled, alert_threshold, emotions, wecom_enabled, wecom_webhook, dingtalk_enabled, dingtalk_webhook, dingtalk_secret ): try: # 更新配置 updated_config = { "enabled": alert_enabled, "alert_threshold": alert_threshold, "target_emotions": emotions, "wecom_enabled": wecom_enabled, "wecom_webhook": wecom_webhook, "dingtalk_enabled": dingtalk_enabled, "dingtalk_webhook": dingtalk_webhook, "dingtalk_secret": dingtalk_secret } # 这里可以添加配置保存逻辑 # 例如保存到配置文件或数据库 return "✅ 配置保存成功！" except Exception as e: return f"❌ 保存失败: {str(e)}" # 在主界面中集成配置界面 with gr.Blocks(title="M2LOrder情感分析") as demo: gr.Markdown("# M2LOrder 情感分析系统") # 原有界面组件... # 模型选择、文本输入、分析按钮等 # 添加机器人配置界面 config_components = create_robot_config_ui() save_btn = config_components[-1] # 绑定保存事件 save_btn.click( fn=save_robot_config, inputs=config_components[:-1], outputs=gr.Textbox(label="保存结果") ) 4.2 添加测试功能
在WebUI中添加机器人测试功能，方便验证配置是否正确：
# 添加测试按钮和结果展示 test_btn = gr.Button("🧪 测试机器人") test_result = gr.Textbox(label="测试结果", interactive=False) # 测试函数 def test_robot_notification(): notifier = RobotNotifier() test_cases = [ ("我非常生气！这个产品太难用了", "angry", 0.95), ("我好焦虑，明天就要交项目了", "anxious", 0.88), ("今天心情很低落", "sad", 0.92) ] results = [] for text, emotion, confidence in test_cases: if notifier.should_alert(emotion, confidence): success = notifier.notify(text, emotion, confidence) status = "成功" if success else "失败" results.append(f"{emotion}测试: {status}") else: results.append(f"{emotion}测试: 未触发告警") return "\n".join(results) # 绑定测试事件 test_btn.click( fn=test_robot_notification, outputs=test_result ) 5. 部署与配置指南
5.1 企业微信机器人配置
- 创建企业微信群聊，点击右上角群设置
- 添加群机器人，获取webhook地址
- 配置环境变量：
# 设置企业微信webhook export WECOM_WEBHOOK="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY" export WECOM_ENABLED="true" 5.2 钉钉机器人配置
- 创建钉钉群聊，点击智能群助手
- 添加自定义机器人，选择加签安全设置
- 获取webhook和secret：
# 设置钉钉机器人 export DINGTALK_WEBHOOK="https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN" export DINGTALK_SECRET="YOUR_SECRET" export DINGTALK_ENABLED="true" 5.3 重启服务应用配置
# 重启服务使配置生效 cd /root/m2lorder ./stop.sh ./start.sh # 或者使用supervisor supervisorctl -c supervisor/supervisord.conf restart all 6. 实际应用场景示例
6.1 客服情感监控
当客户表达强烈不满时自动通知客服主管：
# 示例：检测到愤怒情绪时通知相关团队 客户反馈："你们的服务太差了！我等了2小时都没人理我！" # 机器人推送内容： """ 情感告警: angry情绪检测 置信度: 0.96 原文: 你们的服务太差了！我等了2小时都没人理我！ 时间: 2024-01-20 14:30:25 """ 6.2 社交媒体情感分析
监控品牌社交媒体提及，及时发现负面情绪：
# 示例：社交媒体负面情绪监控 推文："@某某品牌 产品质量有问题，用了三天就坏了" # 触发告警并通知品牌团队 6.3 内部系统情感反馈
收集员工反馈，及时发现团队情绪问题：
# 示例：员工满意度调查负面反馈 反馈："工作压力太大，经常加班到很晚，感觉快要崩溃了" # 通知HR部门关注员工心理健康 7. 高级功能与优化建议
7.1 频率限制与去重
避免重复通知相同内容：
# 在robot_notifier.py中添加去重逻辑 from collections import deque from hashlib import md5 class RobotNotifier: def __init__(self): self.sent_messages = deque(maxlen=100) # 保存最近100条消息的hash def _get_message_hash(self, text: str, emotion: str) -> str: """生成消息哈希用于去重""" content = f"{text}_{emotion}" return md5(content.encode()).hexdigest() def should_send(self, text: str, emotion: str) -> bool: """检查是否应该发送（去重）""" message_hash = self._get_message_hash(text, emotion) if message_hash in self.sent_messages: return False self.sent_messages.append(message_hash) return True 7.2 分级告警机制
根据不同情感强度发送不同级别的通知：
# 分级告警配置 ALERT_LEVELS = { "high": {"emotions": ["angry"], "threshold": 0.9, "mention": "@所有人"}, "medium": {"emotions": ["anxious", "sad"], "threshold": 0.8, "mention": ""}, "low": {"emotions": ["angry", "anxious", "sad"], "threshold": 0.7, "mention": ""} } def get_alert_level(emotion: str, confidence: float) -> str: for level, config in ALERT_LEVELS.items(): if emotion in config["emotions"] and confidence >= config["threshold"]: return level return None 7.3 数据统计与报表
添加情感告警统计功能：
# 情感告警统计 class AlertStatistics: def __init__(self): self.stats = { "total_alerts": 0, "by_emotion": {}, "by_hour": {str(hour): 0 for hour in range(24)}, "last_alert": None } def record_alert(self, emotion: str): self.stats["total_alerts"] += 1 self.stats["by_emotion"][emotion] = self.stats["by_emotion"].get(emotion, 0) + 1 self.stats["by_hour"][str(datetime.now().hour)] += 1 self.stats["last_alert"] = datetime.now() 8. 总结与效果展示
通过本次定制开发，M2LOrder系统获得了实时的情感告警推送能力，主要实现效果：
核心功能实现：
- ✅ 企业微信机器人集成：支持文本消息推送
- ✅ 钉钉机器人集成：支持加签安全验证
- ✅ 智能告警触发：基于情感类型和置信度阈值
- ✅ WebUI配置界面：可视化配置管理
- ✅ 测试验证功能：一键测试机器人连通性
实际应用价值：
- 实时监控：7×24小时情感分析不间断
- 快速响应：重要情感事件立即通知
- 多平台支持：适应不同企业办公环境
- 灵活配置：可根据需求调整告警规则
优化改进：
- 增加消息去重机制，避免重复通知
- 添加分级告警，不同级别不同处理方式
- 完善统计功能，提供数据洞察
现在，当系统检测到强烈的负面情绪时，相关团队能够在第一时间收到通知，及时采取措施，大大提升了情感分析的实用价值和响应效率。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 ZEEKLOG星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
