搭建Node-RED + MQTT Broker实现AI大模型交互

本文档详细记录了使用Node-RED和MQTT Broker构建一个可与DeepSeek AI大模型交互的物联网平台的完整过程。

一、系统架构

[接收设备消息] --> [处理数据] --> [调用DeepSeek API] --> [处理API响应] --> [MQTT Out]
| ^
| |
v |
[错误消息] --> [错误消息接收] --> [MQTT err]

系统主要组件：

1. 设备/客户端：向MQTT Broker发送请求消息
2. MQTT Broker：消息代理，处理发布/订阅
3. Node-RED：流程编排引擎，处理逻辑和API交互
4. DeepSeek API：AI大模型服务

数据流向：

1. 设备发布消息到device/data主题
2. Node-RED订阅并处理消息
3. Node-RED调用DeepSeek API
4. Node-RED将响应发布到device/response主题
5. 设备接收响应

二、环境准备与安装

1. 安装Node.js

首先确保安装了Node.js v18或更高版本：

curl -fsSL https://deb.nodesource.com/setup_18.x | bash -
apt-get install -y nodejs

验证安装：

node -v

2. 安装Mosquitto MQTT Broker

apt-get update
apt-get install -y mosquitto mosquitto-clients

3. 配置Mosquitto

创建配置文件，允许匿名连接：

cat > /etc/mosquitto/conf.d/default.conf << EOF
listener 1883
allow_anonymous true
EOF

重启Mosquitto服务：

systemctl restart mosquitto

4. 安装Node-RED

npm install -g --unsafe-perm node-red

5. 配置Node-RED监听所有网络接口

修改配置文件：

cat > /root/.node-red/settings.js << EOF
module.exports = {
    uiPort: process.env.PORT || 1880,
    uiHost: "0.0.0.0",
}
EOF

6. 启动Node-RED

nohup node-red > node-red.log 2>&1 &

三、Node-RED流程配置

访问Node-RED界面：http://服务器IP:1880/

1. 创建新流程

1. 点击"+"按钮创建新的流程
2. 将流程命名为"DeepSeek AI交互"

2. 添加并配置MQTT In节点

1. 从节点面板中拖动"mqtt in"节点到工作区
2. 双击节点进行配置：
   • 服务器：点击编辑按钮添加新的MQTT Broker
     • 名称：本地MQTT Broker
     • 服务器：localhost
     • 端口：1883
   • 主题：device/data
   • QoS：2
   • 输出：自动检测(JSON对象、字符串或buffer)
   • 名称：接收设备消息

3. 添加并配置处理数据Function节点

1. 从节点面板中拖动"function"节点到工作区
2. 双击节点进行配置：
   • 名称：处理数据
   • 函数代码：

// 处理接收到的设备数据
const deviceData = msg.payload;

// 构建发送给DeepSeek API的请求
msg.payload = {
    "model": "deepseek-chat",
    "messages": [
        {"role": "system", "content": "你是一个助手。"},
        {"role": "user", "content": deviceData.message}
    ]
};

// 设置请求头
msg.headers = {
    "Content-Type": "application/json"
};

// 设置超时时间为60秒
msg.requestTimeout = 60000;

return msg;

4. 添加并配置HTTP Request节点

1. 从节点面板中拖动"http request"节点到工作区
2. 双击节点进行配置：
   • 名称：调用DeepSeek API
   • 方法：POST
   • URL：https://api.deepseek.com/chat/completions
   • 返回：解析为JSON对象
   • 在认证选项卡中：
     • 使用：Bearer Authentication
     • Token：您的DeepSeek API Token (例如: sk-b30b58c4056e4149872d87eb9228ed54)
   • 添加请求头：
     • Content-Type: application/json

5. 添加并配置处理响应Function节点

1. 从节点面板中拖动"function"节点到工作区
2. 双击节点进行配置：
   • 名称：处理API响应
   • 函数代码：

// 处理DeepSeek API的响应
const response = msg.payload;

// 提取AI回复内容
let aiResponse = "";
if (response && response.choices && response.choices.length > 0) {
    aiResponse = response.choices[0].message.content;
} else {
    aiResponse = "无法获取有效回复";
    node.warn("API响应格式不符合预期: " + JSON.stringify(response));
}

// 构建回复消息
msg.payload = {
    "status": "success",
    "response": aiResponse,
    "timestamp": new Date().toISOString()
};

return msg;

6. 添加并配置MQTT Out节点

1. 从节点面板中拖动"mqtt out"节点到工作区
2. 双击节点进行配置：
   • 服务器：选择之前创建的本地MQTT Broker
   • 主题：device/response
   • QoS：1
   • 保留：否
   • 名称：MQTT Out

7. 添加错误处理

1. 从节点面板中拖动"catch"节点到工作区

2. 双击节点进行配置：

   • 名称：错误消息

3. 添加处理错误的Function节点：

   • 名称：错误消息接收
   • 函数代码：

// 记录错误
node.error("处理错误: " + JSON.stringify(msg.error));

// 构建错误响应
msg.payload = {
    "status": "error",
    "message": msg.error ? (msg.error.message || "未知错误") : "处理请求时发生错误",
    "code": msg.statusCode || 500,
    "timestamp": new Date().toISOString()
};

// 设置主题（确保错误消息发送到正确的主题）
msg.topic = "device/error";

return msg;

4. 添加用于错误的MQTT Out节点：
   • 服务器：选择之前创建的本地MQTT Broker
   • 主题：device/error
   • QoS：1
   • 保留：否
   • 名称：MQTT err

8. 连接节点

按照以下顺序连接节点：

1. 接收设备消息 → 处理数据
2. 处理数据 → 调用DeepSeek API
3. 调用DeepSeek API → 处理API响应
4. 处理API响应 → MQTT Out
5. 错误消息 → 错误消息接收
6. 错误消息接收 → MQTT err

9. 部署流程

点击右上角的"部署"按钮使配置生效。

四、测试系统

1. 创建测试脚本

创建一个简单的Node.js脚本来测试系统：

cat > /root/test-mqtt.js << EOF
const mqtt = require('mqtt');
const client = mqtt.connect('mqtt://localhost:1883');

client.on('connect', function () {
  console.log('已连接到MQTT Broker');
  
  client.subscribe('device/response');
  client.subscribe('device/error');
  
  const testMessage = { 
    deviceId: 'test-001', 
    message: '介绍一下Node-RED的基本功能', 
    timestamp: new Date().toISOString() 
  };
  
  console.log('发送测试消息:', testMessage);
  client.publish('device/data', JSON.stringify(testMessage));
});

client.on('message', function (topic, message) {
  console.log('收到消息，主题:', topic);
  try { 
    console.log(JSON.parse(message.toString())); 
  } catch(e) { 
    console.log(message.toString()); 
  }
});

setTimeout(function() { 
  client.end(); 
  console.log('测试完成，已断开连接'); 
}, 120000);
EOF

2. 安装MQTT客户端库

npm install mqtt

3. 运行测试

node /root/test-mqtt.js

输出结果示例：

已连接到MQTT Broker
发送测试消息: { deviceId: 'test-001',
  message: '介绍一下Node-RED的基本功能',
  timestamp: '2025-05-15T06:51:24.069Z' }
收到消息，主题: device/response
{
  status: 'success',
  response: 'Node-RED 是一个基于 Node.js 开发的低代码/可视化编程工具，主要用于连接硬件设备、API 和在线服务，构建物联网（IoT）应用或自动化工作流。其核心特点是通过拖放节点（Nodes）和连线（Flows）快速实现数据流处理，无需深入编码。以下是它的基本功能：\n\n1. 可视化流程编排\n   - 节点（Nodes）：预置了大量功能模块\n   - 连线（Flows）：用连线将节点按逻辑顺序连接\n\n2. 丰富的节点类型\n   - 输入节点：如 HTTP 请求、MQTT 订阅等\n   - 处理节点：函数、延迟、切换等\n   - 输出节点：数据库、API 调用、邮件通知等\n\n3. 易于集成和扩展\n   - 支持各种协议和服务的集成\n   - 可通过npm安装扩展节点',
  timestamp: '2025-05-15T06:51:42.361Z'
}
测试完成，已断开连接

五、在其他Linux客户端使用MQTT与系统交互

1. 安装MQTT客户端工具

# Debian/Ubuntu系统
sudo apt-get install mosquitto-clients

# RHEL/CentOS系统
sudo yum install mosquitto-clients

# Arch系统
sudo pacman -S mosquitto

2. 订阅消息（接收响应）

# 订阅响应主题
mosquitto_sub -h 服务器IP -t "device/response" -v

# 订阅错误主题
mosquitto_sub -h 服务器IP -t "device/error" -v

3. 发送消息（请求AI处理）

# 发送消息
mosquitto_pub -h 服务器IP -t "device/data" -m '{"deviceId":"linux-001","message":"什么是物联网？","timestamp":"'$(date -Iseconds)'"}'

4. 使用Python进行交互（可选）

import paho.mqtt.client as mqtt
import json
import time
from datetime import datetime

# MQTT服务器信息
broker_address = "服务器IP"
port = 1883
pub_topic = "device/data"
sub_topics = ["device/response", "device/error"]

# 回调函数 - 连接成功
def on_connect(client, userdata, flags, rc):
    print("已连接到MQTT Broker")
    # 订阅主题
    for topic in sub_topics:
        client.subscribe(topic)
        print(f"已订阅主题: {topic}")

# 回调函数 - 接收消息
def on_message(client, userdata, msg):
    print(f"\n收到消息 主题: {msg.topic}")
    try:
        payload = json.loads(msg.payload.decode())
        print(json.dumps(payload, indent=2, ensure_ascii=False))
    except:
        print(msg.payload.decode())

# 创建客户端
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

# 连接到Broker
client.connect(broker_address, port, 60)

# 启动网络循环
client.loop_start()

# 发送测试消息
def send_message(message_text):
    msg = {
        "deviceId": "python-device-001",
        "message": message_text,
        "timestamp": datetime.now().isoformat()
    }
    print(f"发送消息: {json.dumps(msg, ensure_ascii=False)}")
    client.publish(pub_topic, json.dumps(msg))

# 等待连接建立
time.sleep(1)

# 示例查询
send_message("请解释什么是物联网？")

# 保持脚本运行以接收响应
try:
    while True:
        user_input = input("\n输入问题(输入'exit'退出): ")
        if user_input.lower() == 'exit':
            break
        send_message(user_input)
        time.sleep(1)
except KeyboardInterrupt:
    print("程序被用户中断")

# 断开连接
client.loop_stop()
client.disconnect()

六、系统安全与扩展

安全配置

1. MQTT安全加强

   • 添加用户名密码认证：修改/etc/mosquitto/conf.d/default.conf
   • 启用TLS加密：配置证书

2. Node-RED安全加强

   • 添加登录认证：修改settings.js
   • 部署HTTPS：配置证书

系统扩展

1. 添加Dashboard

   • 安装Node-RED Dashboard节点
   • 创建可视化界面监控系统

2. 添加数据存储

   • 连接数据库(MySQL/MongoDB)存储交互历史

3. 支持多种AI模型

   • 扩展Function节点支持其他AI模型API

4. 设备认证与管理

   • 开发设备注册和认证机制

七、故障排除

1. 无法连接MQTT Broker

   • 检查防火墙是否开放1883端口
   • 检查Mosquitto服务状态

2. 无法访问Node-RED

   • 检查防火墙是否开放1880端口
   • 检查Node-RED进程是否运行

3. API调用失败

   • 检查API Token是否正确
   • 检查网络连接和API地址
   • 检查HTTP Request节点配置
   • 增加请求超时时间

4. 消息格式错误

   • 确保发送JSON格式消息
   • 确保包含message字段

八、总结

通过本文档，我们就完成了一个基于Node-RED和MQTT Broker的AI交互系统搭建，实现了：

1. 接收来自设备的MQTT消息
2. 处理消息并调用DeepSeek AI API
3. 将AI响应通过MQTT返回给设备
4. 错误处理和日志记录

这个系统可以作为物联网设备与AI大模型交互的基础平台，可根据实际需求进行扩展和优化。