文章目录
- 前言
- 一、准备工作
- (一)钉钉机器人
- (二)智谱 GLM-4-Flash
- (三)内网穿透工具 cpolar
- (四)需要准备的工具和环境
- 二、钉钉机器人的创建与配置
- 步骤1:创建钉钉机器人
- 步骤2:将机器人加入群聊
- 步骤3:配置服务器相关信息
- 三、开发环境准备
- (一) 本地开发环境搭建
- 步骤1:创建 Python环境
- 步骤2:安装必要的依赖库
- (二) 获取 API 密钥
- 四、消息接收与处理
- (一)接收钉钉消息
- (二)解析消息内容
- (三)验证消息合法性
- 五、调用大模型生成答案
- (一)处理用户指令
- (二)发送回复消息
- 六、运行与测试
- (一)本地运行
- 艾特机器人并发送“地震播报”
- 艾特机器人并发送“北京天气”
- 艾特机器人并询问问题
- 总结
- 附录
前言
在当今数字化办公的时代,钉钉作为一款广泛使用的办公协作工具,其机器人功能为企业和团队带来了诸多便利。通过创建自定义钉钉机器人,我们可以实现自动化消息推送、智能问答等功能,极大地提高工作效率。而智谱 GLM-4-Flash 作为一款强大的语言模型,能够生成高质量的文本内容,为我们的机器人赋予更智能的回复能力。本文将详细介绍如何从零开始打造一个智能钉钉机器人,实现调用智谱 API 来生成智能回复的全过程。
一、准备工作
在开始打造智能钉钉机器人之前,我们需要对一些基础知识有所了解,并准备好必要的工具和环境。
(一)钉钉机器人
钉钉机器人是一种可以自动接收和处理钉钉群消息的工具,它通过 webhook 接口与钉钉进行通信。机器人可以被添加到钉钉群聊中,当群成员艾特机器人时,机器人会接收到消息并根据预设的逻辑进行处理和回复。钉钉机器人广泛应用于自动化办公、信息推送、智能问答等场景,能够帮助企业提高工作效率和协作体验。
(二)智谱 GLM-4-Flash
GLM-4-Flash,智谱AI发布的矩阵模型。它能够生成高质量的文本内容,适用于多种自然语言处理任务,如文本生成、问答系统、文本摘要等。通过调用智谱 GLM-4-Flash 的 API 接口,我们可以将用户的输入传递给模型,并获取模型生成的回答内容,从而实现智能回复功能。
(三)内网穿透工具 cpolar
内网穿透是一种技术,它允许外部网络访问内网中的服务。在本地开发环境中,我们通常处于内网,无法直接被钉钉服务器访问。因此,我们需要使用内网穿透工具 cpolar 来创建一个隧道,将本地端口映射到公网,使得钉钉服务器可以通过公网地址访问到本地运行的服务。
(四)需要准备的工具和环境
在开始开发之前,我们需要准备以下工具和环境:
- 钉钉开发者账号:注册钉钉开发者账号,用于创建机器人和获取 webhook token。
- 智谱 GLM-4-Flash 账号:注册智谱平台账号,创建项目并获取 API 密钥。
- 内网穿透工具 cpolar:下载并安装 cpolar 客户端,用于本地开发环境的内网穿透。
- 天气 API 账号(可选):如果需要实现天气查询功能,可以注册天气 API 服务,获取用户 ID 和密钥。
- Python 开发环境:安装 Python 环境(推荐 Python 3.8+),并安装必要的 Python 库。
二、钉钉机器人的创建与配置
注意:需要在有开发者权限的组织才可以进行下面步骤,若发现自己没有开发者权限请联系管理员或自行创建组织测试,本博客选择后者
步骤1:创建钉钉机器人
- 登录钉钉开发者后台:https://developers.dingtalk.com/。
- 在开发者后台的导航栏中找到“机器人”选项,点击进入机器人管理页面。
- 点击“创建应用”按钮,选择“自定义机器人”。
- 填写机器人名称(如“智能助手”)、描述(如“基于智谱 GLM-4-Flash 的智能回复机器人”)等基本信息。
步骤2:将机器人加入群聊
- 打开钉钉应用,进入你希望添加机器人的群聊。
- 点击群设置,找到“机器人”选项。
- 点击“添加机器人”,选择刚才创建的机器人。
- 确认添加后,机器人将自动加入群聊。
完成创建后,会生成一个 webhook access_token,这是后续开发中与钉钉进行通信的关键凭证,务必妥善保存。
步骤3:配置服务器相关信息
- 访问 cpolar 官方网站(https://cpolar.com/),根据自己的操作系统下载并安装 cpolar 客户端。
- 打开 cpolar 客户端,登录账号。创建一个隧道,将本地端口(如 8080)映射到公网。
- 记录下 cpolar 提供的公网地址(如 http://your-public-address:8080)。
- 配置钉钉服务器出口 IP:
- 在钉钉开发者后台的机器人开发管理设置页面,找到“消息接收地址”配置项。
- 将 cpolar 提供的公网 IP 地址填写到配置框中。
- 提示:在这里有公网IP或者云服务器的也可以直接填入公网IP地址,这里做本地测试所以使用内网穿透展示
三、开发环境准备
在开始开发智能钉钉机器人之前,我们需要确保开发环境已经搭建好,并安装了必要的依赖库。以下是详细的环境创建和依赖安装步骤:
(一) 本地开发环境搭建
步骤1:创建 Python环境
本博客为隔离开发环境,使用Anaconda 来管理 Python 环境,也推荐大家这样做。
Anaconda安装教程:https://blog.csdn.net/pdsu_Zhe/article/details/129432873
- 创建新的 Anaconda 环境
conda create -n dingtalk_bot python=3.10
- 激活环境:
conda activate dingtalk_bot
步骤2:安装必要的依赖库
在激活的环境中,使用 pip 安装以下依赖库:
- requests:用于发送 HTTP 请求。
- hmac、hashlib、base64:用于签名验证。
- socket:用于创建服务器端口,接收钉钉发送的消息。
- zhipuai:智谱 GLM-4-Flash 的 Python SDK,用于调用大模型。
pip install requests hmac hashlib base64 socket zhipuai
(二) 获取 API 密钥
1、智谱 GLM-4-Flash API 密钥
- 访问智谱AI开放平台(https://bigmodel.cn/),注册账号并登录。
- 在平台中创建一个项目,获取 API 密钥。
- 将 API 密钥保存到本地,后续代码中会用到。
2、其他功能API密钥
由于博主在开发地震相关的,所以接入了中国地震台网的全球最新地震信息API,大家有需要的API可以自行搜索方法,这里不在讲述。
四、消息接收与处理
(一)接收钉钉消息
使用 Python 的 socket 模块创建一个服务器端口,监听钉钉发送的消息。
def handle_client(client_socket):
request_data = client_socket.recv(20000)
print(f"接收到的消息:{request_data}")
client_socket.close()
if __name__ == "__main__":
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(("", 8080))
server_socket.listen(120)
print("服务器启动,等待连接...")
while True:
client_socket, client_address = server_socket.accept()
print(f"[{client_address}] 用户连接上了")
handle_client(client_socket)
(二)解析消息内容
解析钉钉发送的消息,提取用户 ID、签名、时间戳和消息内容。
def getPost(request_data):
request_data = str(request_data, encoding="utf8").split('\r\n')
items = []
for item in request_data[1:-2]:
items.append(item.split(':'))
post_useful = dict(items)
post_mes = json.loads(request_data[-1])
post_sign = post_useful.get('Sign').strip()
post_timestamp = post_useful.get('Timestamp').strip()
post_userid = post_mes.get('senderId').strip()
post_mes = post_mes.get('text').get('content').strip()
return post_userid, post_sign, post_timestamp, post_mes
(三)验证消息合法性
def initKey(post_userid, post_sign, post_timestamp, post_mes):
app_secret = '你的app_secret' # 替换为你的 app_secret
string_to_sign = f'{post_timestamp}\n{app_secret}'
hmac_code = hmac.new(app_secret.encode('utf-8'), string_to_sign.encode('utf-8'), digestmod=hashlib.sha256).digest()
sign = base64.b64encode(hmac_code).decode('utf-8')
if post_sign == sign:
print("消息验证通过")
return True
else:
print("消息验证失败")
return False
五、调用大模型生成答案
(一)处理用户指令
根据用户发送的消息内容,调用相应的处理逻辑。
from zhipuai import ZhipuAI
client = ZhipuAI(api_key="你的GLM_API_KEY") # 替换为你的智谱 API 密钥
def selectMes(post_userid, post_mes):
if post_mes == "地震播报":
earthquake_data = "地震等级: 5.0, 地震时间: 2025-03-09 10:00, 位置: 四川省" # 替换为实际的地震数据获取逻辑
response = client.chat.completions.create(
model="glm-4-flash",
messages=[
{"role": "system", "content": "生成地震播报文本"},
{"role": "user", "content": earthquake_data}
]
)
answer = response.choices[0].message.content
elif "天气" in post_mes:
province, place = "四川", "成都" # 替换为实际的省份和地点提取逻辑
weather_data = f"{province} {place} 的天气数据" # 替换为实际的天气数据获取逻辑
response = client.chat.completions.create(
model="glm-4-flash",
messages=[
{"role": "system", "content": "生成天气预报文本"},
{"role": "user", "content": weather_data}
]
)
answer = response.choices[0].message.content
else:
response = client.chat.completions.create(
model="glm-4-flash",
messages=[
{"role": "system", "content": "你是一个智能助手,能够回答各种问题。"},
{"role": "user", "content": post_mes}
]
)
answer = response.choices[0].message.content
return sendText(post_userid, answer)
(二)发送回复消息
import requests
import json
def sendText(post_userid, send_mes):
webhook = "https://oapi.dingtalk.com/robot/send?access_token=你的token" # 替换为你的钉钉 webhook token
header = {"Content-Type": "application/json", "Charset": "UTF-8"}
message_json = json.dumps({
"msgtype": "text",
"text": {
"content": send_mes
},
"at": {
"atDingtalkIds": [post_userid],
"isAtAll": False
}
})
response = requests.post(url=webhook, data=message_json, headers=header)
if response.status_code == 200:
print("消息发送成功")
else:
print("消息发送失败")
六、运行与测试
(一)本地运行
启动本地服务器,确保机器人能够正常接收和处理消息。运行以下代码:
if __name__ == "__main__":
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(("", 8080))
server_socket.listen(120)
print("服务器启动,等待连接...")
while True:
client_socket, client_address = server_socket.accept()
print(f"[{client_address}] 用户连接上了")
handle_client(client_socket)
该代码启动一个监听 8080 端口的服务器,等待钉钉发送的消息。
测试机器人功能:
在钉钉群聊中艾特机器人,测试不同指令的响应情况。例如:
艾特机器人并发送“地震播报”
艾特机器人并发送“北京天气”
艾特机器人并询问问题
总结
通过本文的介绍,我们从零开始打造了一个智能钉钉机器人,实现了调用智谱 GLM-4-Flash API 来生成智能回复的全过程。这个机器人可以根据用户的指令,自动获取地震数据、天气信息,并生成相应的播报内容,极大地提高了办公效率和协作体验。
附录
这里只提供大模型智能回复代码,若需其他功能的相关逻辑代码,可以联系我交流学习。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
@author: Richard Zhang
@time: 2025/3/7 19:55
@file: DingTalk.py
@describe: TODO
"""
# -*- coding: GBK -*-
import requests
import json
import time
import hmac
import hashlib
import base64
import socket
from multiprocessing import Process
from zhipuai import ZhipuAI
# 配置
GLM_API_KEY = "your_api_key" # 替换为你的智谱GLM-4-Flash API密钥
# 初始化智谱GLM-4-Flash客户端
client = ZhipuAI(api_key=GLM_API_KEY)
model_name = "glm-4-flash" # 模型名称
def handle_client(client_socket):
# 获取socket
request_data = client_socket.recv(20000)
post_userid, post_sign, post_timestamp, post_mes = getPost(request_data)
# 回应socket
initKey(post_userid, post_sign, post_timestamp, post_mes)
# 关闭socket
client_socket.close()
def getPost(request_data):
request_data = str(request_data, encoding="utf8").split('\r\n')
items = []
for item in request_data[1:-2]:
items.append(item.split(':'))
post_useful = dict(items)
post_mes = json.loads(request_data[-1])
post_sign = post_useful.get('Sign').strip()
post_timestamp = post_useful.get('Timestamp').strip()
post_userid = post_mes.get('senderId').strip()
post_mes = post_mes.get('text').get('content').strip()
return post_userid, post_sign, post_timestamp, post_mes
def initKey(post_userid, post_sign, post_timestamp, post_mes):
whtoken = "your_webhook_token" # 替换为你的webhook_token
timestamp = str(round(time.time() * 1000))
app_secret = 'your_app_secret' # 替换为你的app_secret
app_secret_enc = app_secret.encode('utf-8')
string_to_sign = '{}\n{}'.format(post_timestamp, app_secret)
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = base64.b64encode(hmac_code).decode('utf-8')
if (abs(int(post_timestamp) - int(timestamp)) < 3600000 and post_sign == sign):
webhook = "https://oapi.dingtalk.com/robot/send?access_token=" + whtoken + "×tamp=" + timestamp + "&sign=" + sign
header = {"Content-Type": "application/json", "Charset": "UTF-8"}
message_json = json.dumps(selectMes(post_userid, post_mes))
info = requests.post(url=webhook, data=message_json, headers=header)
print(info)
else:
print("Warning:Not DingDing's post")
def selectMes(post_userid, post_mes):
# 将用户输入传递给大模型生成回答
response = client.chat.completions.create(
model=model_name,
messages=[
{"role": "system", "content": "你是一个智能助手,能够回答各种问题。"},
{"role": "user", "content": post_mes}
]
)
answer = response.choices[0].message.content
# 返回文本消息
return sendText(post_userid, answer)
def sendText(post_userid, send_mes):
message = {
"msgtype": "text",
"text": {
"content": send_mes
},
"at": {
"atDingtalkIds": [post_userid],
"isAtAll": False
}
}
return message
if __name__ == "__main__":
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(("", 8080))
server_socket.listen(120)
while True:
client_socket, client_address = server_socket.accept()
print("[%s, %s]用户连接上了" % client_address)
handle_client_process = Process(target=handle_client, args=(client_socket,))
handle_client_process.start()
client_socket.close()