MQTT Agent

MQTT Agent 将 EMQX 从 MQTT 基础设施升级为 MQTT 原生的 AI 编排平台。

MQTT Agent 使 EMQX 能够运行事件驱动的 AI 自动化流程，利用 EMQX 的连接能力对客户端事件做出响应。

与常见的面向人类的 Agent 不同，MQTT Agent 并非以聊天界面为核心。它是为无人值守的 AI 自动化而构建的：面向大量设备、大量并发工作流、受限的外部系统访问，以及可审计的工具调用。

MQTT Agent 无需将消息代理、API 网关、无服务器运行时、AI 服务、工作流引擎和集成平台分别搭建、再逐一对接，而是将这些能力整合进同一个 MQTT 原生运行时中。

该插件围绕三个可组合的基础能力构建，均通过 MQTT 主题提供：

工具（Tools）：可复用、经过 Schema 校验的能力，用于 MQTT 发布、MQTT 请求/响应、HTTP 调用、数据库查询等。
会话（Sessions）：通过 MQTT 主题路由的可寻址 LLM 对话。会话是上下文的保存者：它保存对话历史、待处理事件、排队中的请求、工具调用状态以及用量计数。
流水线（Pipelines）：事件触发的工作流实例，编排工具与会话的调用以处理 MQTT 事件。

这使得 EMQX 成为一个场所：连接设备产生的事件可以直接触发安全的 AI 工作流——LLM 只能看到被授权的工具，工具可以限定主题与资源边界，会话可以跟踪用量，流水线可以借助 OTP 的故障隔离能力以 EMQX 的规模运行。

能力概览

将 AI 引入 MQTT 运维场景：连接设备产生的事件可以直接触发模型辅助的决策、信息补全、检测、分类以及后续动作。
让自动化贴近 Broker：工作流运行在 MQTT 连接、路由、授权边界与运维遥测已经具备的位置。
限定 AI 的可执行范围：LLM 只能获得被授权的工具，且每个工具都可以被限定在特定的主题、端点、数据库或流上。
应对机器规模的事件流：该能力面向大量设备与大量并发工作流设计，而非面向单一的人类聊天会话。

MQTT Agent 接口

MQTT Agent 使用 MQTT 主题来提供功能。Agent 相关主题使用 $ 前缀，因此它们属于 MQTT 系统主题，不会被普通的 # 订阅匹配到。

工具

工具是受限的动作，可以直接被流水线的某一步骤使用，也可以提供给流水线中 LLM 步骤使用。

工具通过类型和 ID 进行寻址：type@id。类型标识工具的具体实现，例如发起一次 HTTP 请求或执行一次数据库查询；id 标识该工具的一组已配置的选项与限制。

工具主题

工具调用是一次 MQTT 请求/响应交换。调用方将 JSON 请求发布到工具实例的请求主题（$cap/<type>/<tool_id>/request/<req_id>），并在对应的响应主题（$cap/<type>/<tool_id>/response/<req_id>）上等待 JSON 响应：

工具会解码请求负载，依据工具的输入 Schema 校验 args 字段，执行相应动作，并将结果发布到带有相同 req_id 的响应主题上。

例如，以请求 ID req-42 调用 message__publish@alerts 时使用：

请求主题为 $cap/message__publish/alerts/request/req-42
响应主题为 $cap/message__publish/alerts/response/req-42

调用方将以下请求负载发布到请求主题：

json

// PUBLISH $cap/message__publish/alerts/request/req-42
{
  "args": {
    "topic": "factory/line-1/alerts",
    "payload": {"severity": "warning", "reason": "temperature_high"}
  },
  "iid": "pipeline-instance-id",
  "trace_id": "trace-id"
}

发布该 MQTT 消息后，工具会将以下响应负载发布到响应主题：

json

// PUBLISH $cap/message__publish/alerts/response/req-42
{
  "status": "ok",
  "result": {"published": true}
}

类型、实例与上下文

工具类型是通用的实现。工具 ID 是该实现的一个已配置实例。该实例携带一个上下文：调用方在调用时无法更改的固定配置。

例如，postgresql__query 是一个通用的 PostgreSQL 查询执行器。就其本身而言，它只知道如何渲染 SQL 参数、通过一个 EMQX PostgreSQL 连接运行预置查询并返回行数据。而一个已配置的实例会将其收窄为一个更具体的能力：

json

{
  "type": "postgresql__query",
  "id": "orders_by_device",
  "desc": "Read recent orders for one device",
  "resource": "pg-main",
  "query": "select id, status, created_at from orders where device_id = ${device_id} order by created_at desc limit 10"
}

这会创建工具引用 postgresql__query@orders_by_device。流水线或 LLM 步骤可以携带 {"device_id": "dev-001"} 调用该引用，但不能选择其他数据库连接、执行任意 SQL、移除 where 子句或修改 limit。这些固定的部分保存在与 orders_by_device 关联的实例上下文中。

其他工具类型也遵循同样的模式：message__publish 实例固定了发布边界，http 实例固定了端点形态，stream 或 KV 实例固定了存储目标。

内置工具类型

工具类型	用途
`message__publish`	在配置的主题前缀下发布 MQTT 消息。
`message__request`	发送 MQTT 5 请求/响应消息并等待响应。
`http`	使用 Schema 定义的输入调用外部 HTTP 端点。
`postgresql__query`	通过已配置的连接执行带参数的 PostgreSQL 查询。
`stream__write`	向 EMQX 流写入带键数据。
`stream__read`	从 EMQX 流读取带键数据。
`stream__del`	删除带键数据或清空一个 EMQX 流。
`kv__write`	向 EMQX 的 last-value 流写入一条键值数据。
`kv__read`	从 EMQX 的 last-value 流读取一条键值数据。
`kv__read_all`	从 EMQX 的 last-value 流读取全部键值数据。
`kv__del`	从 EMQX 的 last-value 流删除一条键值数据。
`kv__clear`	清空 EMQX 的 last-value 流中的全部键值数据。

图像处理机制

http 和 message__request 工具可以从工具响应中提取图像，以便安全地将多模态数据传递给 LLM。兼容 OpenAI 的 API 不接受直接嵌入在工具响应消息中的图像，因此 Agent 会将负载中提取出的图像替换为 Image <id> 占位符，并将图像数据作为独立的附件返回。

图像提取支持两种模式：

autodiscover_images：扫描响应负载，查找 data:image/...;base64,... 形式的值。
images：使用如 .image_url 或 .（表示根值）这样的路径显式指定图像位置。

当响应的内容类型为图像媒体类型（例如 image/png）时，也可以提取二进制图像响应。

自动发现示例

假设一个 HTTP 工具返回如下包含内联 data URI 的 JSON：

json

{
  "inspection_status": "accepted",
  "image_url": "data:image/png;base64,iVBORw0KGgoAAA...",
  "comment": "front camera frame"
}

启用 autodiscover_images 后，工具响应中会包含脱敏后的结果以及提取出的附件：

json

{
  "status": "ok",
  "result": {
    "inspection_status": "accepted",
    "image_url": "Image .image_url",
    "comment": "front camera frame"
  },
  "attachments": [
    {
      "id": ".image_url",
      "type": "image",
      "mime_type": "image/png",
      "data": "iVBORw0KGgoAAA..."
    }
  ]
}

result 字段会进一步作为工具响应传递给 LLM，attachments 则作为额外的多模态数据传递。

显式路径示例

如果响应中有多个类似图像的字段，可以只配置需要模型检查的那一个：

json

{
  "autodiscover_images": false,
  "images": [".inspection.photo"]
}

对于以下响应：

json

{
  "inspection": {
    "photo": "data:image/jpeg;base64,/9j/4AAQSk...",
    "thumbnail": "data:image/jpeg;base64,/9j/2wBD..."
  }
}

只有 .inspection.photo 会被提取；thumbnail 保持为普通的负载数据。完整的工具响应如下：

json

{
  "status": "ok",
  "result": {
    "inspection": {
      "photo": "Image .inspection.photo",
      "thumbnail": "data:image/jpeg;base64,/9j/2wBD..."
    }
  },
  "attachments": [
    {
      "id": ".inspection.photo",
      "type": "image",
      "mime_type": "image/jpeg",
      "data": "/9j/4AAQSk..."
    }
  ]
}

二进制响应示例

如果一个 HTTP 端点以 Content-Type: image/png 返回原始 PNG 字节，该二进制数据会被视为根 “value”：

text

Content-Type: image/png

<raw PNG bytes>

根负载会被表示为 Image .，PNG 字节会作为独立附件返回：

json

{
  "status": "ok",
  "result": "Image .",
  "attachments": [
    {
      "id": ".",
      "type": "image",
      "mime_type": "image/png",
      "data": "iVBORw0KGgoAAA..."
    }
  ]
}

元工具（Meta-Tools）

元工具允许构建用于修改 Agent 配置本身的流水线。它们本质上也是普通工具，但通常只暴露给受信任的构建类工作流。

agent__create_tool
agent__update_tool
agent__delete_tool
agent__query_tools
agent__create_pipeline
agent__update_pipeline
agent__delete_pipeline
agent__query_pipelines
agent__insert_pipeline_step
agent__update_pipeline_step
agent__delete_pipeline_step
agent__query_providers
agent__query_connections

会话

会话是通过 MQTT 主题路由的可寻址 LLM 状态机。一个会话保存对话历史、待处理事件、排队中的请求、工具调用状态以及用量计数。

会话流量使用两种主题格式：

$sess/in/<sid> —— 发往会话的入站帧。
$sess/out/<sid> —— 来自会话的出站帧。

每个会话由一个集群内唯一的 sid（会话 ID）标识。

$sess/in/<sid> 上的入站帧：

帧类型	用途
`request`	携带 provider、model、instructions、input、tools 及持久化设置，启动一次 LLM 处理。
`tool_result`	返回会话所请求的工具调用结果。
`event`	为下一轮 LLM 对话添加新的事件上下文。
`stop`	显式终止该会话。

$sess/out/<sid> 上的出站帧：

帧类型	用途
`intermediate`	在本轮结束前流式返回一个中间模型输出块。携带 `chunk_type`（例如 `content`）以及 `chunk` 中的数据块。
`tool_request`	请求等待中的流水线通过 `$cap/...` 调用一个工具。
`final`	结束当前这一轮 LLM 对话，返回结果及用量计数。
`error`	报告会话侧的失败，例如 provider 不可用或历史压缩出错。

每个出站帧都包含 sid、iid、trace_id 以及累计的 usage。目前模型的推理/思考数据块保留在会话内部；只有已发布的流式数据块会以 intermediate 帧的形式出现。

启用持久化后，会话不会在 final 发布后终止，而是继续存在，可以接收后续请求，形成多轮对话。

流水线

一个流水线定义包含一个 ID、一个 MQTT 触发条件以及若干有序步骤。当一条到来的 MQTT 消息匹配触发主题过滤器时，MQTT Agent 会启动一个流水线实例，该消息会作为流水线上下文中的 $.event 提供给该实例。

流水线的触发主题是普通的 MQTT 主题过滤器，匹配 $evt/... 事件主题，例如：

text

$evt/device/+/done

流水线生命周期事件会以 JSON 形式发布到：

text

$pipe/<pipeline_id>/inst/<iid>/events

支持的步骤类型：

call_tool：调用一个工具（例如 MQTT 发布、HTTP、PostgreSQL、KV 或流存储），并将其结果写入上下文。
llm_loop：将工作发送给一个会话，将选定的工具作为 LLM 工具暴露出去，并在会话返回结果时保存最终结果或结构化结果。
break：根据上下文中的某个值提前终止流水线。

流水线上下文

流水线上下文是一个在同一流水线实例的所有步骤间共享的、以二进制为键的映射。它以 {event: event_payload} 起始。步骤输入可以使用类似 JSONPath 的字符串（例如 $.event.device_id 或 $.inspection.status）引用先前的值；步骤输出则通过该步骤设置的 result_path 写入，例如 $.inspection。

流水线逻辑

流水线是单轮处理器：一个触发事件创建一个流水线实例，该实例针对这一事件协调有序的工作，发布完成或失败结果，然后终止。它不是一个长时间运行、面向人类的 Agent 循环。

这是有意为之的设计。面向人类的 Agent 通常依赖人类不断向主会话中补充带反馈的提示，因此以轮次为单位的对话是其天然的外层模型。而 MQTT Agent 面向的是无人值守场景：事件来自设备、Broker 钩子、订阅、规则、外部系统及其他自动化来源，没有单一的人类对话需要维系。取而代之，流水线负责处理单个事件，会话在需要时提供 LLM 层面的连续性，而 kv_* 或 stream_* 工具则在多个事件之间提供显式的工作流记忆。

流水线可以处于激活或草稿状态；草稿流水线会被保存，但在被激活前不会运行。

LLM 步骤的键表达式

为了模拟多轮对话，仍然可以在 llm_loop 步骤中使用持久化的 LLM 会话。此时，该步骤会为每个流水线实例使用由该步骤的 键表达式（key expression） 所决定的会话标识符。通过使用不同的键表达式，可以按 clientid、主题或其他条件各自维护独立的会话。

管理入口

主管理界面通过插件 API 网关提供：

text

/api/v5/plugin_api/emqx_agent/ui

其他页面：

text

/api/v5/plugin_api/emqx_agent/builder/ui
/api/v5/plugin_api/emqx_agent/apple-box/ui

同一管理入口也可以通过 /api/v5/plugin_api/emqx_agent 下的插件 API 路径访问：

路径	用途
`/tools`	列出并创建工具。
`/tools/:type/:id`	获取、更新或删除一个工具。
`/tools/statuses`	查看工具运行时协调状态。
`/connections`	列出并创建工具连接。
`/connections/:id`	获取、更新或删除一个连接。
`/connections/:id/start`	启用并协调一个连接。
`/connections/:id/stop`	停用并协调一个连接。
`/connections/statuses`	查看连接运行时状态。
`/providers`	列出已配置的 AI provider。
`/pipelines`	列出并创建流水线定义。
`/pipelines/:id`	获取、更新或删除一个流水线。

演示页面

该插件包含两个浏览器端演示：

Pipeline Builder（位于 /builder/ui）：一个聊天式界面，用于构建事件驱动的 AI 工作流。
Apple Box Conveyor（位于 /apple-box/ui）：一个模拟苹果箱质检流程的 MQTT/AI 工作流演示。

在 EMQX 启动并启用该插件后，从代码仓库根目录预置演示资源。两个演示都需要一个兼容 OpenAI 的 API Key：

bash

export OPENAI_API_KEY='sk-...'

可选的环境变量：

变量	默认值	用途
`EMQX_BASE_URL`	`http://localhost:18083/api/v5/plugin_api/emqx_agent`	MQTT Agent 插件 API 的基础 URL。
`EMQX_CORE_BASE_URL`	`http://localhost:18083/api/v5`	用于 AI provider 管理的 EMQX 核心 API 基础 URL。
`EMQX_API_CREDS`	`key:secret`	Basic Auth API 凭证。
`OPENAI_BASE_URL`	`https://api.openai.com/v1`	兼容 OpenAI 的 API 基础 URL。
`OPENAI_MODEL`	脚本各自的默认值	演示流水线使用的模型。
`PGHOST`、`PGPORT`、`PGDATABASE`、`PGUSER`、`PGPASSWORD`	`pgsql`、`5432`、`mqtt`、`root`、`public`	演示工具使用的 PostgreSQL 连接信息。

预置 Apple Box Conveyor 演示：

bash

python3 plugins/emqx_agent/demo_apple_box_init.py

该脚本会创建 apple-inspector AI provider、PostgreSQL 连接、apple-box 相关工具、数据库表，以及处于激活状态的 apple-box-inspection 流水线。在以下地址打开界面：

text

/api/v5/plugin_api/emqx_agent/apple-box/ui

预置 Pipeline Builder 演示：

bash

python3 plugins/emqx_agent/demo_builder_init.py

该脚本会创建 builder AI provider、PostgreSQL 连接、builder 相关元工具、回复工具、数据库表，以及处于激活状态的 pipeline-builder 流水线。在以下地址打开界面：

text

/api/v5/plugin_api/emqx_agent/builder/ui

两个脚本在预置资源前，都可能重新创建各自的演示资源，并删除已存在的 Agent 演示资源。如需显式移除演示资源，运行：

bash

python3 plugins/emqx_agent/demo_teardown.py

构建与测试

从代码仓库根目录构建该插件：

bash

make plugin-emqx_agent

运行该插件的 Common Test 测试套件：

bash

make plugins/emqx_agent-ct

依赖 LLM 的演示测试套件需要一个具备相应能力的 LLM，因此只有在设置了 OPENAI_API_KEY 时才会运行，否则会被跳过。

开发

在节点中构建、安装、启用并启动该插件：

bash

plugins/emqx_agent/script/start_dev.sh

管理界面可通过插件 API 网关访问：

text

/api/v5/plugin_api/emqx_agent/ui

下载

各 EMQX 版本对应的插件安装包：

EMQX 版本	插件版本	安装包
6.3.0	1.0.0	emqx_agent-1.0.0.tar.gz (sha256)

Kubernetes

EMQX Operator

管理 EMQX 集群

API Reference

密码认证

监控

访问控制

集成

管理

集成 Prometheus

集成 OpenTelemetry

端到端追踪

插件目录

JT/T 808 协议网关

GB/T 32960 协议网关

火山引擎 RTC

场景样例

GPT-Realtime

MQTT Agent ​

能力概览 ​

MQTT Agent 接口 ​

工具 ​

工具主题 ​

类型、实例与上下文 ​

内置工具类型 ​

图像处理机制 ​

自动发现示例 ​

显式路径示例 ​

二进制响应示例 ​

元工具（Meta-Tools） ​

会话 ​

流水线 ​

流水线上下文 ​

流水线逻辑 ​

LLM 步骤的键表达式 ​

管理入口 ​

演示页面 ​

构建与测试 ​

开发 ​

下载 ​

MQTT Agent

能力概览

MQTT Agent 接口

工具

工具主题

类型、实例与上下文

内置工具类型

图像处理机制

自动发现示例

显式路径示例

二进制响应示例

元工具（Meta-Tools）

会话

流水线

流水线上下文

流水线逻辑

LLM 步骤的键表达式

管理入口

演示页面

构建与测试

开发

下载