【飞凌嵌入式FCU1501试用】异常检测与LLM智能诊断

引言

在上一篇文章中，我们介绍了基于RK3506的Modbus-TCP传感器采集系统的基础架构和数据采集功能。本文将深入讲解系统的核心功能——异常检测机制和LLM智能诊断，包括多种异常检测算法、事件序列化、LLM集成方案以及关键的性能优化。

异常检测算法

异常类型定义

系统支持6种异常类型，覆盖通信故障和物理数据异常：

异常类型	枚举值	严重级别	描述
COMM_TIMEOUT	通信超时	Critical	连续3次通信失败
COMM_CRC_ERROR	CRC校验错误	Critical	数据传输完整性校验失败
PHYSICAL_HIGH	物理上限越界	Critical	传感器值超过设定上限
PHYSICAL_LOW	物理下限越界	Critical	传感器值低于设定下限
SPIKE	突变/尖峰	Critical	数据变化速率超过阈值
STUCK	数据卡死	Warning	数据长时间保持不变

异常规则配置

每种传感器的每个字段都可以配置独立的检测规则：

typedef struct {
    double min;              // 物理下限阈值
    double max;              // 物理上限阈值
    double max_delta_per_sec; // 每秒最大变化量（突变检测）
    int stuck_seconds;       // 卡死判定时间（秒）
} anomaly_rule_t;

初始化规则配置：

void anomaly_rules_init(void) {
    // 环境传感器 (Slave 1)
    anomaly_rules[0][0] = (anomaly_rule_t){0.0, 50.0, 5.0, 600};   // 温度：0-50°C，每秒变化≤5°C，10分钟卡死
    anomaly_rules[0][1] = (anomaly_rule_t){20.0, 80.0, 10.0, 600}; // 湿度：20-80%RH
    
    // 振动传感器 (Slave 2) - RMS值
    anomaly_rules[1][3] = (anomaly_rule_t){0.0, 4.5, 2.0, 600};    // RMS：0-4.5mm/s
    
    // 电力传感器 (Slave 3)
    anomaly_rules[2][0] = (anomaly_rule_t){20.0, 28.0, 5.0, 600};  // 电压：20-28V
    anomaly_rules[2][1] = (anomaly_rule_t){0.0, 8.0, 2.0, 600};    // 电流：0-8A
    
    // 管道传感器 (Slave 4)
    anomaly_rules[3][0] = (anomaly_rule_t){0.2, 0.8, 0.1, 600};    // 压力：0.2-0.8MPa
    anomaly_rules[3][1] = (anomaly_rule_t){0.0, 100.0, 10.0, 600};  // 流量：0-100m³/h
}

物理越界检测

检测传感器值是否超出物理范围：

static int check_range(sensor_data_t *data, int field_idx, anomaly_event_t *event) {
    anomaly_rule_t *rule = &anomaly_rules[data->map - sensor_maps][field_idx];
    double val = data->values[field_idx];
    
    // 上限检测
    if (rule->max > 0 && val > rule->max) {
        event->type = ANOMALY_PHYSICAL_HIGH;
        event->severity = SEVERITY_CRITICAL;
        event->sensor = data->map;
        event->field_index = field_idx;
        event->current_value = val;
        event->threshold = rule->max;
        event->timestamp = time(NULL);
        return 1;
    }
    
    // 下限检测
    if (rule->min > 0 && val < rule->min) {
        event->type = ANOMALY_PHYSICAL_LOW;
        event->severity = SEVERITY_CRITICAL;
        event->sensor = data->map;
        event->field_index = field_idx;
        event->current_value = val;
        event->threshold = rule->min;
        event->timestamp = time(NULL);
        return 1;
    }
    
    return 0;
}

突变检测（SPIKE）

检测数据变化速率是否超过阈值：

static int check_spike(sensor_data_t *data, int field_idx, anomaly_event_t *event) {
    anomaly_rule_t *rule = &anomaly_rules[data->map - sensor_maps][field_idx];
    if (rule->max_delta_per_sec <= 0) return 0;
    
    time_t now = time(NULL);
    if (data->last_update == 0 || data->prev_values[field_idx] == 0) return 0;
    
    // 计算变化速率
    double delta = fabs(data->values[field_idx] - data->prev_values[field_idx]);
    double time_diff = now - data->last_update;
    if (time_diff <= 0) time_diff = 1;
    
    double rate = delta / time_diff;
    if (rate > rule->max_delta_per_sec) {
        event->type = ANOMALY_SPIKE;
        event->severity = SEVERITY_CRITICAL;
        event->sensor = data->map;
        event->field_index = field_idx;
        event->current_value = data->values[field_idx];
        event->prev_value = data->prev_values[field_idx];
        event->threshold = rule->max_delta_per_sec;
        event->timestamp = now;
        return 1;
    }
    
    return 0;
}

数据卡死检测（STUCK）

检测数据是否长时间保持不变：

static int check_stuck(sensor_data_t *data, int field_idx, anomaly_event_t *event) {
    anomaly_rule_t *rule = &anomaly_rules[data->map - sensor_maps][field_idx];
    if (rule->stuck_seconds <= 0) return 0;
    
    time_t now = time(NULL);
    if (data->last_update == 0) return 0;
    
    double delta = fabs(data->values[field_idx] - data->prev_values[field_idx]);
    // 变化小于0.001且超过卡死时间
    if (delta < 0.001 && (now - data->last_update) >= rule->stuck_seconds) {
        event->type = ANOMALY_STUCK;
        event->severity = SEVERITY_WARNING;
        event->sensor = data->map;
        event->field_index = field_idx;
        event->current_value = data->values[field_idx];
        event->prev_value = data->prev_values[field_idx];
        event->duration_seconds = now - data->last_update;
        event->timestamp = now;
        return 1;
    }
    return 0;
}

通信超时检测

检测连续通信失败次数：

int anomaly_detect(sensor_data_t *data, anomaly_event_t *event) {
    if (!data->valid) {
        // 连续3次通信失败触发异常
        if (data->comm_error_count >= 3) {
            event->type = ANOMALY_COMM_TIMEOUT;
            event->severity = SEVERITY_CRITICAL;
            event->sensor = data->map;
            event->timestamp = time(NULL);
            return 1;
        }
        return 0;
    }
    
    // 依次检测：越界 → 突变 → 卡死
    for (int i = 0; i < data->map->count; i++) {
        if (check_range(data, i, event)) return 1;
        if (check_spike(data, i, event)) return 1;
        if (check_stuck(data, i, event)) return 1;
    }
    
    return 0;
}

事件JSON序列化

结构化事件数据

将异常事件转换为JSON格式，便于传输和存储：

int event_to_json(anomaly_event_t *event, char *buf, int max_len) {
    struct tm *tm = localtime(&event->timestamp);
    char ts[32];
    strftime(ts, sizeof(ts), "%Y-%m-%dT%H:%M:%S+08:00", tm);
    
    int len = snprintf(buf, max_len,
        "{"
        "\"event_id\":\"rk3506-%s-%06d\","
        "\"timestamp\":\"%s\","
        "\"device\":\"rk3506-edge-01\","
        "\"sensor\":{"
        "\"slave_id\":%d,"
        "\"name\":\"%s\","
        "\"field\":\"%s\","
        "\"unit\":\"%s\""
        "},"
        "\"abnormal_type\":\"%s\","
        "\"severity\":\"%s\","
        "\"current_value\":%.2f",
        ts, (int)(event->timestamp % 1000000),
        ts,
        event->sensor->slave_id,
        event->sensor->name,
        event->sensor->field[event->field_index],
        event->sensor->unit[event->field_index],
        anomaly_type_name(event->type),
        severity_name(event->severity),
        event->current_value);
    
    // 根据异常类型添加额外字段
    if (event->type == ANOMALY_PHYSICAL_HIGH || event->type == ANOMALY_PHYSICAL_LOW) {
        len += snprintf(buf + len, max_len - len,
            ",\"threshold\":{\"%s\":%.2f}",
            event->type == ANOMALY_PHYSICAL_HIGH ? "max" : "min",
            event->threshold);
    }
    
    len += snprintf(buf + len, max_len - len,
        ",\"modbus\":{"
        "\"baudrate\":9600,"
        "\"parity\":\"N\""
        "}"
        "}");
    
    return len;
}

JSON输出示例

{
    "event_id": "rk3506-2026-07-06T13:15:13+08:00-343713",
    "timestamp": "2026-07-06T13:15:13+08:00",
    "device": "rk3506-edge-01",
    "sensor": {
        "slave_id": 1,
        "name": "env",
        "field": "temperature",
        "unit": "C"
    },
    "abnormal_type": "PHYSICAL_HIGH",
    "severity": "critical",
    "current_value": 85.00,
    "threshold": {
        "max": 50.00
    },
    "modbus": {
        "baudrate": 9600,
        "parity": "N"
    }
}

LLM集成方案

方案对比

方案	优势	劣势	适用场景
本地Ollama	低延迟、隐私保护	需要GPU资源、模型受限	高端边缘设备、数据敏感场景
在线API	无需本地资源、模型丰富	网络依赖、响应延迟	轻量边缘设备、快速部署

最终选择

由于RK3506资源有限（无GPU），选择在线OpenAI兼容接口方案：

static const char *llm_url = "https://your-api-endpoint.com/v1/chat/completions";
static const char *llm_api_key = "YOUR_API_KEY";
static const char *llm_model = "your-model-name";

C语言LLM客户端

利用curl命令封装HTTP请求：

int llm_send_event(const char *event_json, char *response, int response_len) {
    if (!event_json || !response) return -1;
    
    // JSON转义（处理特殊字符）
    char escaped[MAX_JSON_LEN];
    json_escape(event_json, escaped, sizeof(escaped));
    
    // 创建临时文件存放请求体
    char tmpfile[] = "/tmp/rk3506_event_XXXXXX";
    int fd = mkstemp(tmpfile);
    FILE *fp = fdopen(fd, "w");
    
    fprintf(fp, "{\"model\":\"%s\",\"messages\":[{\"role\":\"system\",\"content\":\"你是一个工业设备异常诊断专家，请用简洁专业的中文回答，不超过200字。\"},{\"role\":\"user\",\"content\":\"分析以下异常事件：%s\"}],\"temperature\":0.3}",
            llm_model, escaped);
    fclose(fp);
    
    // 构建curl命令
    char cmd[MAX_RESPONSE_LEN];
    snprintf(cmd, sizeof(cmd),
        "curl -s -k -X POST '%s' "
        "-H 'Content-Type: application/json' "
        "-H 'Authorization: Bearer %s' "
        "--max-time 30 "
        "--data-binary @%s 2>/dev/null",
        llm_url, llm_api_key, tmpfile);
    
    // 执行curl命令并读取响应
    fp = popen(cmd, "r");
    int len = fread(response, 1, response_len - 1, fp);
    response[len] = '\0';
    pclose(fp);
    
    unlink(tmpfile);
    return len;
}

JSON转义处理

处理JSON中的特殊字符，确保请求格式正确：

static void json_escape(const char *src, char *dst, int dst_len) {
    int i = 0, j = 0;
    while (src[i] && j < dst_len - 1) {
        switch (src[i]) {
            case '\\': dst[j++] = '\\'; dst[j++] = '\\'; break;
            case '"':  dst[j++] = '\\'; dst[j++] = '"';  break;
            case '\n': dst[j++] = '\\'; dst[j++] = 'n';  break;
            case '\r': dst[j++] = '\\'; dst[j++] = 'r';  break;
            case '\t': dst[j++] = '\\'; dst[j++] = 't';  break;
            default:   dst[j++] = src[i];                 break;
        }
        i++;
    }
    dst[j] = '\0';
}

性能优化：异步处理

问题分析

在初期版本中，LLM调用是同步的，导致严重的性能问题：

问题链路：
Slave 1读取 → 检测到异常 → 调用LLM（阻塞3-10秒）→ Slave 2读取（超时失败）
                                               ↑
                                       LLM响应延迟导致TCP连接超时

现象：Slave 1检测到异常后，LLM调用阻塞主循环，导致后续Slave 2、3、4的Modbus通信超时失败。

解决方案：fork异步处理

使用fork()创建后台子进程执行LLM诊断：

static void llm_diagnose_async(const char *json) {
    char response[MAX_RESPONSE_LEN];
    int ret = llm_send_event(json, response, sizeof(response));
    if (ret > 0 && strlen(response) > 0) {
        char diagnosis[MAX_RESPONSE_LEN];
        extract_diagnosis(response, diagnosis, sizeof(diagnosis));
        printf("[LLM DIAGNOSIS] %s\n", diagnosis);
    } else {
        printf("[LLM] No response (network may be unavailable)\n");
    }
}

static void handle_anomaly(anomaly_event_t *event) {
    char json[MAX_JSON_LEN];
    event_to_json(event, json, sizeof(json));
    printf("\n[ANOMALY] %s (%s): %s=%s\n",
           anomaly_type_name(event->type),
           severity_name(event->severity),
           event->sensor->field[event->field_index],
           json);
    
    printf("[LLM] Starting async diagnosis...\n");
    
    // 创建子进程执行LLM调用
    pid_t pid = fork();
    if (pid == 0) {
        // 子进程：执行LLM诊断
        llm_diagnose_async(json);
        exit(0);
    } else if (pid > 0) {
        // 父进程：设置SIGCHLD忽略，避免僵尸进程
        signal(SIGCHLD, SIG_IGN);
    } else {
        // fork失败：降级为同步处理
        printf("[LLM] Failed to fork async process\n");
        llm_diagnose_async(json);
    }
}

关键技术点

fork()创建子进程：主循环立即返回，继续处理下一个传感器
signal(SIGCHLD, SIG_IGN)：自动回收子进程，避免僵尸进程
降级机制：fork失败时降级为同步处理，保证功能可用

优化效果

优化前（同步）：
--- 13:15:13 ---
[env] Slave 1: temperature=85.00°C, humidity=51.00%RH
[ANOMALY] PHYSICAL_HIGH (critical)
[LLM DIAGNOSIS] ... (等待3-10秒)
[vibration] Slave 2: COMM ERROR (-3)  ← 超时失败！

优化后（异步）：
--- 13:19:30 ---
[env] Slave 1: temperature=85.00°C, humidity=51.00%RH
[ANOMALY] PHYSICAL_HIGH (critical)
[LLM] Starting async diagnosis...      ← 立即返回
[vibration] Slave 2: x=0.71mm/s, y=0.15mm/s, z=0.15mm/s, rms=0.74mm/s  ✅
[power] Slave 3: voltage=24.40V, current=1.40A, power=34.50W            ✅
[pipe] Slave 4: pressure=0.64MPa, flow=54.72m3/h                         ✅
[LLM DIAGNOSIS] ... (后台执行，几秒后输出)

完整测试演示

测试环境

组件	IP地址	端口	说明
传感器模拟器	<SIMULATOR_IP>		注入高温故障
RK3506采集器	<RK3506_IP>	-	运行采集程序

启动模拟器

python3 simulator/modbus_slave_sim.py \
    --mode tcp \
    --tcp-port 7502 \
    --slaves 1,2,3,4 \
    --fault high_temp \
    --fault-slave 1

运行采集器

./modbus_collector --mode tcp --host <SIMULATOR_IP> --port <PORT>

测试输出

Starting RK3506 Modbus-TCP Collector
TCP host: <SIMULATOR_IP>:<PORT>
Polling sensors...

--- 13:19:30 ---
[env] Slave 1: temperature=85.00°C, humidity=51.00%RH

[ANOMALY] PHYSICAL_HIGH (critical): temperature={"event_id":"rk3506-2026-07-06T13:19:30+08:00-343970","timestamp":"2026-07-06T13:19:30+08:00","device":"rk3506-edge-01","sensor":{"slave_id":1,"name":"env","field":"temperature","unit":"C"},"abnormal_type":"PHYSICAL_HIGH","severity":"critical","current_value":85.00,"threshold":{"max":50.00},"modbus":{"baudrate":9600,"parity":"N"}}
[LLM] Starting async diagnosis...
[vibration] Slave 2: x=0.71mm/s, y=0.15mm/s, z=0.15mm/s, rms=0.74mm/s
[power] Slave 3: voltage=24.40V, current=1.40A, power=34.50W
[pipe] Slave 4: pressure=0.64MPa, flow=54.72m3/h
[LLM DIAGNOSIS] 环境温度传感器(slave_id 1)检测到异常高温，当前值85℃已严重超出阈值50℃。该事件严重等级为critical，可能原因包括设备过载、散热系统故障或环境温度异常。建议立即检查设备散热状况、运行负载及周边通风条件，防止设备因过热损坏。

LLM诊断结果分析

LLM返回的诊断报告包含：

异常描述：环境温度传感器检测到异常高温
严重程度：critical（严重）
可能原因：设备过载、散热系统故障、环境温度异常
处理建议：检查散热、降低负载、改善通风

总结

本文介绍了基于RK3506的工业边缘采集系统的第二部分——异常检测与LLM智能诊断。主要内容包括：

异常检测算法：4种检测机制（物理越界、突变、卡死、通信超时）
事件序列化：JSON格式输出，包含完整上下文信息
LLM集成：在线OpenAI兼容接口，curl命令封装
异步优化：fork子进程解决LLM延迟问题
完整测试：高温故障模拟，LLM诊断响应正常

技术亮点

技术点	实现方式	价值
多维度异常检测	阈值+速率+时间三重判断	覆盖各类工业异常场景
结构化事件数据	JSON格式	便于后续分析和存储
在线LLM集成	curl命令封装	零依赖，快速部署
异步处理	fork子进程	避免阻塞主采集循环
故障注入测试	模拟器支持多种故障模式	便于测试验证

技术交流

欢迎在评论区留言讨论！如果您有任何问题或建议，我会及时回复。

📌 提示：本文是系列文章的第二篇，完整介绍了异常检测和LLM智能诊断功能。如果您觉得本文对你有帮助，欢迎点赞、收藏、关注！