当前位置：首页 > news >正文

Phi-3 Mini部署教程：集成企业微信/钉钉机器人实现消息推送通知

news 2026/3/26 21:51:57

Phi-3 Mini部署教程：集成企业微信/钉钉机器人实现消息推送通知

1. 项目概述

Phi-3 Forest Laboratory是一个基于微软Phi-3 Mini 128K Instruct模型构建的AI对话终端，以其极简主义和治愈系设计风格著称。本教程将指导您完成模型部署，并实现与企业微信/钉钉机器人的集成，打造一个能够自动推送通知的智能助手系统。

2. 环境准备

2.1 硬件要求

GPU：推荐NVIDIA RTX 3090/4090或更高性能显卡
内存：至少16GB RAM
存储：20GB可用空间

2.2 软件依赖

# 创建Python虚拟环境 python -m venv phi3_env source phi3_env/bin/activate # Linux/macOS # phi3_env\Scripts\activate # Windows # 安装核心依赖 pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 pip install transformers>=4.40.0 streamlit requests

3. 基础部署

3.1 下载模型

from transformers import AutoModelForCausalLM, AutoTokenizer model_name = "microsoft/Phi-3-mini-128k-instruct" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForCausalLM.from_pretrained(model_name, device_map="auto")

3.2 创建基础对话界面

新建app.py文件：

import streamlit as st from transformers import AutoModelForCausalLM, AutoTokenizer # 初始化模型 @st.cache_resource def load_model(): model_name = "microsoft/Phi-3-mini-128k-instruct" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForCausalLM.from_pretrained(model_name, device_map="auto") return model, tokenizer model, tokenizer = load_model() # 创建界面 st.title("🌿 Phi-3 Forest Laboratory") user_input = st.text_input("向森林深处发出的讯息") if user_input: inputs = tokenizer(user_input, return_tensors="pt").to("cuda") outputs = model.generate(**inputs, max_new_tokens=200) response = tokenizer.decode(outputs[0], skip_special_tokens=True) st.write(f"森林回应: {response}")

启动应用：

streamlit run app.py

4. 集成消息推送功能

4.1 企业微信机器人配置

在企业微信群中添加自定义机器人
获取Webhook地址（格式：https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=XXX）

4.2 钉钉机器人配置

在钉钉群设置中添加自定义机器人
选择"自定义"类型，获取Webhook地址和安全设置

4.3 实现推送功能

修改app.py添加以下代码：

import requests import json def send_wechat_work(message, webhook_url): headers = {"Content-Type": "application/json"} data = { "msgtype": "text", "text": {"content": message} } response = requests.post(webhook_url, headers=headers, data=json.dumps(data)) return response.status_code == 200 def send_dingtalk(message, webhook_url, secret=None): headers = {"Content-Type": "application/json"} timestamp = str(round(time.time() * 1000)) sign = "" if secret: string_to_sign = f"{timestamp}\n{secret}" sign = base64.b64encode(hmac.new(secret.encode('utf-8'), string_to_sign.encode('utf-8'), digestmod=hashlib.sha256).digest()).decode('utf-8') data = { "msgtype": "text", "text": {"content": message}, "at": {"isAtAll": False} } if secret: webhook_url += f"&timestamp={timestamp}&sign={sign}" response = requests.post(webhook_url, headers=headers, data=json.dumps(data)) return response.status_code == 200

5. 完整集成方案

5.1 修改主程序

更新app.py实现完整功能：

# 在Streamlit侧边栏添加配置选项 with st.sidebar: st.header("消息推送设置") notification_type = st.selectbox("选择通知方式", ["无", "企业微信", "钉钉"]) if notification_type == "企业微信": webhook_url = st.text_input("企业微信Webhook地址") elif notification_type == "钉钉": webhook_url = st.text_input("钉钉Webhook地址") secret = st.text_input("钉钉加签密钥(可选)") # 在生成响应后添加推送逻辑 if user_input and response: notification_msg = f"新消息:\n用户: {user_input}\nAI回复: {response}" if notification_type == "企业微信" and webhook_url: if send_wechat_work(notification_msg, webhook_url): st.success("已推送至企业微信") else: st.error("推送失败") elif notification_type == "钉钉" and webhook_url: if send_dingtalk(notification_msg, webhook_url, secret if secret else None): st.success("已推送至钉钉") else: st.error("推送失败")

5.2 环境变量配置（可选）

创建.env文件：

WECHAT_WORK_WEBHOOK=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key DINGTALK_WEBHOOK=https://oapi.dingtalk.com/robot/send?access_token=your_token DINGTALK_SECRET=your_secret

修改代码读取环境变量：

from dotenv import load_dotenv import os load_dotenv() # 在Streamlit界面中使用默认值 webhook_url_wechat = os.getenv("WECHAT_WORK_WEBHOOK", "") webhook_url_dingtalk = os.getenv("DINGTALK_WEBHOOK", "") dingtalk_secret = os.getenv("DINGTALK_SECRET", "")

6. 高级功能扩展

6.1 定时推送功能

添加定时任务，每天推送精选内容：

import schedule import time from threading import Thread def daily_report(webhook_url): prompt = "生成一份包含励志语录、今日建议和科技新闻摘要的日报" inputs = tokenizer(prompt, return_tensors="pt").to("cuda") outputs = model.generate(**inputs, max_new_tokens=500) report = tokenizer.decode(outputs[0], skip_special_tokens=True) send_wechat_work(report, webhook_url) def run_scheduler(): schedule.every().day.at("09:00").do(daily_report, webhook_url=os.getenv("WECHAT_WORK_WEBHOOK")) while True: schedule.run_pending() time.sleep(60) # 在单独线程中运行定时任务 Thread(target=run_scheduler, daemon=True).start()

6.2 消息过滤与格式化

def format_notification(user_input, response): # 简化AI回复，去除重复内容 clean_response = response.replace(user_input, "").strip() # 限制长度 if len(clean_response) > 300: clean_response = clean_response[:300] + "..." return f"💬 用户咨询: {user_input}\n🌿 AI回复: {clean_response}" # 在推送前调用 notification_msg = format_notification(user_input, response)