unsetunset搭建微信AI助理unsetunset
wechatbot-webhook部署
wechatbot-webhook 是一个轻量、可部署的微信机器人webhook服务，使用http接口收发微信消息, 用它作为个人通知、AIGC 应用或者 coze、n8n、Dify等自动化工作流的消息节点。
基于 http 请求，与hooks微信不同，因为基于web api，所以优势在于可以部署到arm架构等设备上。
1、开源地址：https://github.com/danni-cool/wechatbot-webhook
2、部署
Docker部署
# 拉取最新镜像
docker pull dannicool/docker-wechatbot-webhook
# # 启动容器并映射日志目录，日志按天维度生成，e.g: app.2024-08-029.log
docker run -d --name wxBotWebhook --restart unless-stopped -p 3001:3001 -v ~/wxBot_logs:/app/log -e ACCEPT_RECVD_MSG_MYSELF=true -e RECVD_MSG_API=http://119.91.148.69:3000/receive/ -e LOGIN_API_TOKEN=197653 dannicool/docker-wechatbot-webhook
ACCEPT_RECVD_MSG_MYSELF=true：机器人能接收自己发的消息。
RECVD_MSG_API=http://http://119.91.148.69:3000/receive：接收消息的 API 地址，实现处理接收消息的逻辑，如果你不想实现这个功能，可以不填，默认为空。
LOGIN_API_TOKEN=197653：自定义登录 API token，可以不填，默认自动生成一个。
使用 compose 部署 (可选)
wget -O docker-compose.yml https://cdn.jsdelivr.net/gh/danni-cool/wechatbot-webhook@main/docker-compose.yml && docker-compose down && docker-compose -p wx_bot_webhook up
3、登录
# -f 表示 "follow"，即实时跟踪日志输出。
docker logs -f wxBotWebhook
# 启动容器时，使用 -d 选项将其置于后台运行。
docker run -d --name wxBotWebhook c7feffe5b169(image_name)
# 将日志输出重定向到文件
docker logs wxBotWebhook > /path/to/logfile.log 2>&1 &
# 查看日志
tail -f /path/to/logfile.log
/path/to/logfile.log：替换为你希望保存日志的文件路径。 2>&1：将标准错误输出（stderr）也重定向到日志文件。 &：将命令放到后台执行。
这样，容器在后台运行，同时日志被写入指定文件，而不会实时打印到终端。
服务器端命令行扫码登录
http://119.91.148.69:3001/login?token=197653,扫码登录wx。
项目部署成功后，拿到发送消息的 API 地址：http://119.91.148.69:3001/webhook/msg/v2?token=197653。
推消息 API
1、发送私聊
def send_to_user(nick_name, data):
"""
私聊
:param nick_name: 昵称
:param data: 消息内容
:return:
"""
# 请求头
headers = {
"Content-Type": "application/json",
}
data = create_request_data(nick_name, data)
response = post_request_with_retries(url=BOT_URL, headers=headers, data=json.dumps(data))
print(response.json())
# 单条消息
data = {
"content": "爬虫与AI前沿"
}
# 同一个人发送多条消息
datas = [
{
"type": "text",
"content": "你好?"
},
{
"type": "fileUrl",
"content": "https://static.cninfo.com.cn/finalpage/2024-08-29/1221034055.PDF"
}
]
send_to_user(nick_name='摩诘', data=data)
send_to_user(nick_name='摩诘', data=datas)
2、发送群聊
def send_to_room(nick_name, data):
"""
群聊
:param nick_name: 群昵称
:param data: 消息内容
:return:
"""
# 请求头
headers = {
"Content-Type": "application/json",
}
data = create_request_data(nick_name, data, is_room=True)
response = post_request_with_retries(url=BOT_URL, headers=headers, data=data)
print(response.json())
# 单条消息
data = {
"content": "爬虫与AI前沿"
}
# 多条消息
datas = [
{
"type": "text",
"content": "你好?"
},
{
"type": "fileUrl",
"content": "https://static.cninfo.com.cn/finalpage/2024-08-29/1221034055.PDF"
}
]
send_to_room(nick_name='爬虫与AI前沿', data=data)
send_to_room(nick_name='爬虫与AI前沿', data=datas)
3、发送 Url 文件
def send_to_url(nick_name, content):
# 给 url 拼接 query 参数 $alias 可用于指定发送给目标的文件名
# 请求头
headers = {
"Content-Type": "application/json",
}
data = {
"to": nick_name,
"data": {
"type": "fileUrl",
"content": content
}
}
response = post_request_with_retries(url=BOT_URL, headers=headers, data=json.dumps(data))
print(response.json())
# 发文件 url 同时支持修改成目标文件名
content = "https://download.samplelib.com/jpeg/sample-clouds-400x300.jpg?$alias=1234.jpg"
send_to_url(nick_name='摩诘', content=content)
$alias：自定义文件名
4、发送本地文件
def send_to_local_file(nick_name, file_name, is_room=False):
"""
发送本地文件
:param nick_name: 昵称
:param file_name: 文件路径
:param is_room: 是否是群聊
:return:
"""
data = create_request_data(to=nick_name, is_room=is_room)
files = {'content': open(file_name, 'rb')}
response = post_request_with_retries(url=BOT_URL_FILE, headers=None, data=data, files=files)
print(response.json())
# 发送本地文件-私聊
file_path = '/Users/oscar/Downloads/dify.png'
send_to_local_file(nick_name='摩诘', file_name=file_path)
request 库中默认 ContentType: multipart/form-data类型，故此处headers设置为None.
5、给多人群发
def send_to_rooms():
"""
群发消息
:return:
"""
# 请求头
headers = {
"Content-Type": "application/json",
}
data = [
{
"to": "摩诘",
"data": {
"content": "你好?"
}
},
{
"to": "杨",
"data": [
{
"content": "你好?"
},
{
"content": "近况如何？测试群发机器人！"
}
]
}
]
response = post_request_with_retries(url=BOT_URL, headers=headers, data=json.dumps(data))
print(response.json())
返回值 response 结构:
{
"success": true,
"message": "Message sent successfully",
"task": {
"successCount": 3,
"totalCount": 3,
"failedCount": 0,
"reject": [],
"sentFailed": [],
"notFound": []
}
}
success: 消息发送成功与否，群发消息即使部份发送成功也会返回 true message: 出错时提示的消息 消息发送成功: Message sent successfully 参数校验不通过: Some params is not valid, sending task is suspend. 消息都发送失败: All Messages [number] sent failed... 部份发送成功: Part of the message sent successfully... task: 发送任务详细信息 task.successCount: 发送成功条数 task.totalCount: 总消息条数 task.failedCount: 发送失败条数 task.reject: 因为参数校验不通过的参数和 error 提示 task.sentFailed: 因为发送失败和 error 提示 task.notFound: 因为未找到用户或者群和 error 提示
收消息 API
payload 结构
Methods: POST ContentType: multipart/form-data Form格式如下：
RECVD_MSG_API即：http://119.91.148.69:3000/receive/ ,这个参数就是用来接收消息的。
1. 使用 pip 安装 FastAPI
在终端或命令提示符中运行以下命令：
pip install fastapi
2. 安装 Uvicorn（推荐的 ASGI 服务器）
FastAPI 是一个基于 ASGI 的框架，Uvicorn 是一个高性能的 ASGI 服务器，用于运行 FastAPI 应用。安装 Uvicorn 可以通过以下命令完成：
pip install uvicorn
3、安装 python-multipart
python-multipart 是一个用于处理多部分表单数据（multipart/form-data）的库，通常在处理文件上传时使用。要安装 python-multipart，你可以使用以下 pip 命令：
pip install python-multipart
4. 创建一个简单的 FastAPI 应用
你可以使用以下代码创建一个简单的 FastAPI 应用，并将其保存为 main.py：
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
4. 运行 FastAPI 应用
使用 Uvicorn 运行 FastAPI 应用，可以在终端中运行以下命令：
uvicorn server:app --reload
or
nohup python3 server.py > log.txt 2>&1 &
main是你保存 FastAPI 应用的文件名（不包括.py后缀）。app是 FastAPI 实例的名称。--reload选项启用自动重载，以便在代码更改时自动重新启动服务器。
5. 访问应用
打开浏览器，访问 http://127.0.0.1:3000。你应该能看到返回的 {"Hello": "World"}。
还可以访问自动生成的 API 文档，分别为：
Swagger UI： http://127.0.0.1:3000/docsReDoc： http://127.0.0.1:3000/redoc
创建 wechatbot服务端
1、创建一个名为server.py的文件
from fastapi import FastAPI, Form
from fastapi.responses import JSONResponse
import uvicorn
from loguru import logger
app = FastAPI()
# 配置 Loguru 日志
logger.add("logs/app.log", rotation="1 week", retention="1 month", level="INFO", format="{time} {level} {message}")
@app.post("/receive")
async def receive_message(
type: str = Form(...),
content: str = Form(...),
source: str = Form(...),
isMentioned: str = Form(...),
isMsgFromSelf: str = Form(...),
):
# 处理请求数据
response_data = {
"type": type,
"content": content,
"source": source,
"isMentioned": isMentioned,
"isMsgFromSelf": isMsgFromSelf,
}
try:
# 填写处理逻辑-开始
logger.info("Received data: {}", response_data)
# 填写处理逻辑-结束
return JSONResponse(content={"status": "success", "data": response_data})
except Exception as e:
logger.error("Error processing request: {}", e)
return JSONResponse(content={"status": "error", "data": "处理失败"})
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=3000)
上述处理逻辑部分，完全可以定制化： 比如接入 ChatGPT/文心一言 处理消息后进行回复。
2 、启动服务器
# 创建虚拟环境
conda create -n wechatbot python=3.10
# 激活虚拟环境
conda activate wechatbot
# 克隆代码
git clone https://github.com/huanglaoxie0503/wechatbot.git
# 安装所需的库
pip install -r requirements.txt
# 启动
nohup python server.py > log.txt 2>&1 &
wechatbot服务已经部署好，接下来咱们考虑接入Dify自动化工作流的消息节点，打造智能的 AI 助理。
测试工作流，并发布为工具。
测试Agent,此处使用文心一言大模型。
完整代码：https://github.com/huanglaoxie0503/wechatbo
