## 一、业务背景：一条产线的“停机焦虑”

去年我所在的团队接手了某大型汽车零部件制造企业的智能化改造项目。这家企业在国内拥有 6 个生产基地，超过 200 条自动化产线，其核心痛点非常集中：

**设备故障导致的非计划停机是最大的成本黑洞。** 据统计，单条压铸产线每次意外停机造成的直接损失超过 12 万元/小时，而工厂平均每月发生 3-5 次非计划停机，年均损失接近千万。

深入产线后发现三个核心矛盾：
- **知识断层**：资深维修工程师平均工龄 15 年，故障排查全凭经验，但这套“手艺”没有系统化沉淀，老师傅退休就意味着知识流失。
- **响应延迟**：凌晨设备报警时，夜班操作工无法独立判断严重程度，往往要电话吵醒工程师远程指导，平均响应时间超过 45 分钟。
- **信息孤岛**：设备 PLC 数据在 SCADA 系统里，维修手册是 PDF 堆在文件服务器，历史工单在 ERP 里，故障排查时需要来回切换多个系统，效率极低。

管理层希望用 AI 技术打破这个困局，核心目标就一个：**把非计划停机降低 50%**。我们评估后决定使用火山引擎 HiAgent 平台，从零构建一个“设备智能运维助手”。

## 二、平台选型：为什么是 HiAgent

当时也在 Dify 和 Coze 之间摇摆过，但三个因素让我们最终选择了 HiAgent：

**1. 工业场景的私有化部署是刚需**

设备参数、工艺参数、产线布局都属于核心商业机密，不允许出企业内网。HiAgent 支持云原生轻量级架构私有化部署，企业数据不出域，这是硬指标。相比之下，Coze 当时主要以云服务为主，私有化选项有限。

**2. Agent DevOps 理念契合“边用边学”的需求**

工业场景下不可能一次性定义所有故障模式。HiAgent 基于 Agent DevOps 理念，提供从策略开发、能力搭建、效果评测、应用发布到线上观测的全生命周期管理，支持智能体“越用越聪明”。运维人员可以根据产线反馈持续优化智能体的诊断逻辑，形成闭环。

**3. 高低代码混合开发兼顾敏捷与深度**

设备对接涉及复杂的工业协议解析和定制算法，纯低代码无法满足。HiAgent 支持高代码与无/低代码混合开发，业务人员可以低代码配置常规问答，开发人员则通过代码节点实现深度定制。平台内置 100+ 行业场景模板和丰富插件，支持无缝集成 MCP 协议扩展工具生态。

## 三、总体架构设计

基于 HiAgent 的“1+N+X”架构理念，我们设计了三层体系：

```
┌─────────────────────────────────────────────┐
│             统一交互入口 (1)                    │
│    企业微信 / Web H5 / 产线终端平板            │
├─────────────────────────────────────────────┤
│             通用能力智能体 (N)                  │
│  ┌─────────┐ ┌──────────┐ ┌───────────┐     │
│  │故障诊断 │ │维修指导  │ │备件查询   │       │
│  └─────────┘ └──────────┘ └───────────┘     │
├─────────────────────────────────────────────┤
│           行业垂直智能体 (X)                    │
│  ┌─────────┐ ┌──────────┐ ┌───────────┐     │
│  │压铸工艺 │ │焊接质量  │ │注塑参数   │       │
│  └─────────┘ └──────────┘ └───────────┘     │
│            HiAgent 智能体工作站                 │
│  ┌──────────────────────────────────────┐    │
│  │ 工作流引擎 │ 知识库 │ 插件中心 │ 观测系统 │    │
│  └──────────────────────────────────────┘    │
│  ┌──────────────────────────────────────┐    │
│  │ 豆包大模型 │ 混元 │ DeepSeek │ 多模型管理  │    │
│  └──────────────────────────────────────┘    │
└─────────────────────────────────────────────┘
```

核心设计原则是**“分层解耦、知识先行”** ——先把企业已有的设备知识体系化，再逐步接入实时数据和复杂推理。

## 四、核心智能体实现（含代码）

我们围绕最关键的“故障诊断智能体”展开详细实现。这个智能体负责接收设备报警信息，综合知识库和实时数据，给出诊断结论与处理建议。

### 4.1 知识库构建：让设备手册“活”起来

知识库是整个智能体的基石。我们梳理了三类知识源：

| 类别 | 来源 | 处理方式 |
|------|------|----------|
| 维修手册 | 设备供应商提供的 PDF 文档（中英德混合） | TextIn 解析引擎 + 语义分段 |
| 历史工单 | ERP 系统中的故障维修记录（结构化数据） | 数据清洗后导入 HiAgent 知识库 |
| 专家经验 | 资深工程师口述的故障排查“土方法” | 人工整理为标准问答对 |

**踩坑记录——文档解析**

最初的维修手册是扫描版 PDF，默认的文档解析节点对表格和多栏排版处理效果不佳。后来我们接入了合合信息 TextIn 解析引擎，对段落、表格、标题和版面坐标做向量化处理，让 RAG 召回从“纯文本”升级为“多维度结构”。这一步将检索准确率从 67% 提升到 92%。

知识库分段策略上，我们强制要求每个故障条目以独立分段存储，标题包含设备型号和故障码，例如：

```markdown
## DMG-Mori-NHX5000 | 报警码 A-037 | 主轴过载
### 故障现象
主轴负载率超过额定值 120%，伴随异常振动和异响。
### 可能原因
1. 刀具磨损严重，切削阻力增大
2. 主轴轴承预紧力异常
3. 冷却液流量不足导致热膨胀
### 排查步骤
1. 检查当前刀具切削时长，超过 200 小时建议更换
2. 测量主轴轴承温度，正常范围 35-55°C
3. 检查冷却液管路压力，标准值 0.4-0.6 MPa
```

### 4.2 工作流设计：三层诊断流水线

故障诊断智能体采用 HiAgent 可视化工作流编排，通过拖拽方式将知识库、插件、大模型和代码块串联为完整的诊断流水线。

核心工作流分为三个阶段：

**第一层——意图识别与信息补全**

用户输入设备报警信息后（支持文本和语音），智能体首先完成意图分类和信息提取：

```python
# 代码节点：报警信息结构化提取
def extract_alarm_info(user_input: str, context: dict) -> dict:
    """从用户输入中提取设备报警关键信息"""
    # 通过 LLM 节点输出的意图分类结果获取结构化数据
    alarm_info = context.get("alarm_info", {})

    # 补充设备上下文（从会话变量中获取历史设备信息）
    device_id = context.get("device_id", "")
    if not device_id and alarm_info.get("device_code"):
        device_id = alarm_info["device_code"]

    # 补充时间上下文
    from datetime import datetime
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    return {
        "device_id": device_id,
        "alarm_code": alarm_info.get("alarm_code", ""),
        "alarm_description": alarm_info.get("description", ""),
        "severity": alarm_info.get("severity", "unknown"),
        "timestamp": timestamp,
        "extraction_complete": True
    }
```

**第二层——RAG 增强知识检索**

提取到报警信息后，进入 HiAgent 知识库检索节点。我们采用混合检索策略——向量检索捕获语义相似度，关键词检索锁定精确的报警码和设备型号，两者互补，大幅提升召回精度。

```
# HiAgent 知识库检索节点配置（概念示意）

检索策略：混合检索（向量 + 关键词）
向量模型：豆包 Embedding V3
Top K：5（开启 ReRank 重排序）
关键词权重：0.4
检索过滤：metadata.device_model == "{extracted_device_model}"
```

关键经验：我们为每个知识片段添加了元数据标签（设备型号、报警码、故障类别），检索时通过动态过滤器精准命中，避免了跨设备类型知识污染。

**第三层——LLM 综合诊断与代码校验**

检索结果进入 LLM 节点进行综合推理。但我们没有直接输出诊断结论，而是增加了一个代码节点做“安全校验”：

```python
# 代码节点：诊断结论安全校验与结构化输出
import json
import re

def validate_diagnosis(llm_output: str, alarm_info: dict, context: dict) -> dict:
    """校验 LLM 诊断结论的安全性并结构化输出"""

    diagnosis = json.loads(llm_output) if isinstance(llm_output, str) else llm_output

    # 1. 危险操作拦截：禁止直接建议断电、重启关键设备等操作
    dangerous_patterns = [
        r"直接.*断电",
        r"强制.*重启",
        r"绕过.*安全.*装置",
        r"手动.*短接"
    ]
    for pattern in dangerous_patterns:
        if re.search(pattern, diagnosis.get("suggestion", "")):
            return {
                "status": "blocked",
                "message": "诊断建议包含潜在危险操作，已转人工处理",
                "alert_level": "high"
            }

    # 2. 置信度阈值检查
    confidence = diagnosis.get("confidence", 0.0)
    if confidence < 0.6:
        diagnosis["final_action"] = "转人工复核"
        diagnosis["low_confidence_reason"] = f"诊断置信度 {confidence} 低于阈值 0.6"
    elif confidence < 0.85:
        diagnosis["final_action"] = "建议参考 + 人工确认"
    else:
        diagnosis["final_action"] = "可自动执行"

    # 3. 结构化输出，适配下游系统
    return {
        "status": "success",
        "device_id": alarm_info.get("device_id"),
        "alarm_code": alarm_info.get("alarm_code"),
        "root_cause": diagnosis.get("root_cause", ""),
        "possible_causes": diagnosis.get("possible_causes", []),
        "suggestion": diagnosis.get("suggestion", ""),
        "estimated_repair_time_min": diagnosis.get("estimated_repair_time_min", 0),
        "spare_parts_required": diagnosis.get("spare_parts_required", []),
        "confidence": confidence,
        "final_action": diagnosis.get("final_action", "转人工"),
        "timestamp": alarm_info.get("timestamp")
    }
```

这个校验节点是整个智能体的“安全阀”，上线后成功拦截了 4 次可能造成设备二次损坏的错误建议。

### 4.3 插件集成：打通 SCADA 实时数据

静态知识库只能解决“这类故障通常怎么修”，但生产环境的故障诊断必须结合实时设备数据。我们基于 HiAgent 的自定义插件能力，封装了 SCADA 系统的数据查询接口：

```python
# 自定义插件：SCADA 实时数据查询
# 在 HiAgent 插件中心注册为自定义插件 "SCADAQuery"

import requests
import hashlib
import time
from typing import Optional

class SCADAQueryPlugin:
    """SCADA 系统实时数据查询插件"""

    def __init__(self, config: dict):
        self.base_url = config["scada_base_url"]
        self.api_key = config["scada_api_key"]
        self.api_secret = config["scada_api_secret"]
        self.timeout = config.get("timeout", 10)

    def _sign_request(self, params: dict) -> dict:
        """生成请求签名（HMAC-SHA256）"""
        timestamp = str(int(time.time()))
        sign_str = f"{self.api_key}{timestamp}{json.dumps(params, sort_keys=True)}"
        signature = hashlib.sha256(
            (sign_str + self.api_secret).encode()
        ).hexdigest()
        return {
            "X-API-Key": self.api_key,
            "X-Timestamp": timestamp,
            "X-Signature": signature
        }

    def query_device_realtime(self, device_id: str, 
                               tags: Optional[list] = None) -> dict:
        """查询设备实时数据点"""
        if tags is None:
            # 默认查询关键监控指标
            tags = [
                "spindle_load_percent",
                "spindle_speed_rpm",
                "servo_current_A",
                "temperature_motor_celsius",
                "temperature_bearing_celsius",
                "coolant_flow_L_min",
                "coolant_pressure_MPa",
                "vibration_mm_s",
                "power_consumption_kW",
                "running_hours_since_maintenance"
            ]

        endpoint = f"{self.base_url}/api/v2/device/{device_id}/realtime"
        params = {"tags": ",".join(tags)}

        try:
            headers = self._sign_request(params)
            response = requests.get(
                endpoint,
                params=params,
                headers=headers,
                timeout=self.timeout
            )
            response.raise_for_status()
            data = response.json()

            # 附加阈值判断，辅助智能体理解数据含义
            thresholds = {
                "spindle_load_percent": {"normal_max": 85, "warning": 95, "critical": 110},
                "temperature_bearing_celsius": {"normal_max": 55, "warning": 65, "critical": 75},
                "vibration_mm_s": {"normal_max": 4.5, "warning": 7.0, "critical": 11.0},
                "coolant_pressure_MPa": {"normal_min": 0.4, "normal_max": 0.6}
            }

            for tag_name, value in data.get("tags", {}).items():
                if tag_name in thresholds:
                    status = self._evaluate_threshold(value, thresholds[tag_name])
                    data["tags"][f"{tag_name}_status"] = status

            return {"success": True, "data": data}

        except requests.exceptions.Timeout:
            return {"success": False, "error": "SCADA 查询超时", "device_id": device_id}
        except requests.exceptions.RequestException as e:
            return {"success": False, "error": str(e), "device_id": device_id}

    def _evaluate_threshold(self, value: float, threshold: dict) -> str:
        """根据阈值判断数据状态"""
        if "normal_max" in threshold:
            if value > threshold.get("critical", float("inf")):
                return "critical"
            elif value > threshold.get("warning", float("inf")):
                return "warning"
            elif value > threshold["normal_max"]:
                return "above_normal"
            else:
                return "normal"
        elif "normal_min" in threshold:
            if value < threshold.get("critical", -float("inf")):
                return "critical"
            elif value < threshold.get("warning", -float("inf")):
                return "warning"
            elif value < threshold["normal_min"]:
                return "below_normal"
            else:
                return "normal"
        return "unknown"

    def query_device_history(self, device_id: str, 
                             tag: str, 
                             hours_back: int = 24) -> dict:
        """查询设备历史趋势数据，用于趋势分析"""
        endpoint = f"{self.base_url}/api/v2/device/{device_id}/history"
        params = {
            "tag": tag,
            "hours_back": hours_back,
            "interval_minutes": 5
        }
        try:
            headers = self._sign_request(params)
            response = requests.get(
                endpoint,
                params=params,
                headers=headers,
                timeout=self.timeout
            )
            response.raise_for_status()
            return {"success": True, "data": response.json()}
        except requests.exceptions.RequestException as e:
            return {"success": False, "error": str(e), "device_id": device_id}
```

在工作流中，当报警码涉及温度、负载等可量化指标时，诊断节点先调 SCADA 插件获取实时数据，再将数据与知识库检索结果一起送入 LLM 进行综合推理。例如“主轴过载报警”场景下，插件返回实时负载率 127%，LLM 直接判定为严重过载，跳过知识库中的“检查刀具”步骤，给出紧急停机建议——这种“知识 + 数据”的双引擎模式，将诊断准确率从纯知识库的 78% 提升到了 93%。

### 4.4 多智能体协作：诊断与备件的协同

HiAgent 2.0 支持 Multi-Agent 模式，多个智能体可以像团队一样分工协作完成复杂任务。我们构建了一个“调度 Agent”来协调故障诊断 Agent 和备件查询 Agent：

```
# 多智能体协作流程（HiAgent Canvas 编排）

用户: "DMG 加工中心报警 A-037，主轴过载"
    │
    ▼
┌──────────────────┐
│   调度 Agent      │  ← 意图识别、任务分解
└──────┬───────────┘
       │
       ├──► ┌──────────────────┐
       │    │  故障诊断 Agent   │  ← 知识库 + SCADA 数据
       │    │  根因：刀具磨损    │
       │    └──────────────────┘
       │
       └──► ┌──────────────────┐
            │  备件查询 Agent   │  ← ERP 库存查询
            │  刀具型号：BT40   │
            │  库存：A仓-3号位  │
            └──────────────────┘
       │
       ▼
┌──────────────────┐
│   回复组装        │  "主轴过载原因为刀具磨损（负载率127%），
│   (调度Agent)     │   建议更换BT40刀具，备件在A仓3号位，
│                   │   当前库存5把，请立即处理。"
└──────────────────┘
```

这种多智能体协作模式，让操作工一次提问即可获得“故障原因 + 处理方案 + 备件位置”的完整答案，无需在不同系统间切换。

## 五、部署与观测：从上线到“越用越聪明”

### 5.1 多渠道发布

HiAgent 支持将智能体一键发布至飞书、钉钉、企业微信等 IM 平台，也支持通过 API 和 Web SDK 集成到企业现有系统。我们在产线终端平板上部署了 Web H5 版本，同时在维修工程师的企业微信中集成了 Bot。

发布策略采用灰度模式，先在 2 条试点产线运行 2 周，验证稳定后全量推广。

### 5.2 全链路观测

HiAgent 提供实时指标监控、链路追踪和会话管理能力，支撑排障和运营分析。我们重点监控三个维度的指标：

```python
# 观测指标配置示例（概念代码）
monitoring_config = {
    "business_metrics": {
        "diagnosis_accuracy": "> 90%",           # 诊断准确率
        "first_response_time_ms": "< 3000",       # 首次响应时间
        "resolution_rate": "> 75%",               # 自助解决率
        "avg_repair_time_reduction_percent": "30" # 维修时间缩短比例
    },
    "technical_metrics": {
        "knowledge_recall_precision": "> 85%",    # 知识召回精度
        "scada_query_success_rate": "> 99%",      # SCADA 查询成功率
        "llm_hallucination_rate": "< 3%",         # LLM 幻觉率
        "workflow_timeout_rate": "< 1%"           # 工作流超时率
    },
    "safety_metrics": {
        "dangerous_suggestion_blocked_count": "per_day",  # 危险建议拦截数
        "human_escalation_rate": "monitor"                # 转人工率趋势
    }
}
```

线上观测数据会自动沉淀为评测集，驱动智能体的持续调优，形成“使用→反馈→优化→再使用”的闭环。

### 5.3 持续迭代机制

- **每周**：运营团队根据观测面板中的低分回答和用户反馈，补充知识库或修正问答对。
- **每月**：基于累积的线上数据重新评测诊断准确率，必要时微调提示词或检索策略。
- **每季度**：引入新的故障模式训练数据，迭代知识库分段策略。

## 六、落地成效

上线 6 个月后的数据：

| 指标 | 上线前 | 上线后 | 变化 |
|------|--------|--------|------|
| 平均故障响应时间 | 45 分钟 | 3 分钟 | **↓ 93%** |
| 非计划停机次数/月 | 4.2 次 | 1.1 次 | **↓ 74%** |
| 单次停机损失 | 12 万元/时 | — | 年均减少损失约 450 万 |
| 夜班需人工介入比例 | 100% | 23% | **↓ 77%** |
| 新员工独立排查能力 | 需 6 个月培养 | 2 周即可 | **培训周期缩短 80%** |

维修工程师的反馈是：“以前半夜被叫起来是常态，现在智能体先过滤 70% 以上的简单故障，剩下的复杂问题推到我这之前已经有了完整的上下文和数据，决策效率完全不一样了。”

## 七、几点核心心得

**1. 工业场景的安全性是第一优先级**

智能体绝对不能给出“直接断电”“手动短接安全回路”等危险建议。我们通过代码节点的规则校验 + 低置信度自动转人工两层机制来兜底。建议从项目第一天就建立危险操作知识库，并持续更新拦截规则。

**2. 知识工程是智能体的地基，不能省**

花在知识库梳理上的时间比写代码多 3 倍，但这恰恰是决定成败的关键。我们的做法是“先质量后数量”——先精心打磨 50 个高频故障的知识条目，验证召回效果达标后再批量扩充。

**3. 实时数据是工业智能体的“灵魂”**

纯知识库只能回答“常见故障怎么修”，只有接入 SCADA 实时数据才能回答“当前这台设备具体出了什么问题”。HiAgent 的插件机制让这一步成本很低，但工程上需要做好超时降级、缓存和异常处理。

**4. HiAgent 的 Agent DevOps 理念很适合工业场景**

工业场景的故障模式永远在演化，不可能一次定义完整。HiAgent 的全生命周期管理体系让智能体能够随产线一起“成长”——观测面板发现问题，评测系统定位根因，优化后再上线，形成持续进化的闭环。

**5. 让一线人员参与智能体建设**

教会产线班长使用 HiAgent 的工作流编辑器和知识库管理功能后，他们直接贡献了 40% 的有效知识条目。平台的“低门槛”设计让业务专家真正成为了智能体的共建者而非旁观者。

目前我们正在基于当前基础，探索将设备振动频谱分析算法封装为 HiAgent 插件，让智能体直接从原始传感器波形中提取特征进行故障预判——从“故障后诊断”向“故障前预测”演进。如果你也在做工业场景的智能体落地，建议从最痛的一条产线、最标准的一类故障切入，先把“知识库 + 实时数据 + 安全校验”这条链路跑通，后续的扩展自然会水到渠成。