边缘计算与边缘AI：将智能推向数据源

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

// 边缘计算分层架构
interface EdgeArchitecture {
  // 云层：全局协调与训练
  cloud: {
    modelTraining: ModelTrainingService;
    globalOrchestration: OrchestrationService;
    dataLake: DataLakeStorage;
    analytics: GlobalAnalytics;
  };

  // 边缘层：区域协调与推理
  edge: {
    inference: EdgeInferenceEngine;
    aggregation: DataAggregationService;
    localTraining: FederatedLearning;
    cache: EdgeCacheLayer;
  };

  // 设备层：数据采集与执行
  device: {
    sensors: SensorArray;
    actuators: ActuatorController;
    preprocessing: DataPreprocessing;
    communication: DeviceCommunication;
  };
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 轻量级边缘节点容器
FROM alpine:3.19

# 安装运行时
RUN apk add --no-cache \
    nodejs \
    npm \
    python3 \
    py3-pip \
    tensorflow-lite

# 安装边缘服务
COPY edge-service/package.json /app/
RUN npm install --production

COPY edge-service/ /app/

# 配置资源限制
# 内存限制：512MB
# CPU限制：1核心
WORKDIR /app

CMD ["node", "index.js"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# Kubernetes 部署配置
apiVersion: v1
kind: Pod
metadata:
  name: edge-inference-node
  labels:
    app: edge-ai
    tier: edge
spec:
  containers:
  - name: inference-engine
    image: edge-ai:latest
    resources:
      limits:
        memory: "512Mi"
        cpu: "1000m"
      requests:
        memory: "256Mi"
        cpu: "500m"
    volumeMounts:
    - name: model-cache
      mountPath: /models
    env:
    - name: EDGE_NODE_ID
      valueFrom:
        fieldRef:
          fieldPath: metadata.uid
  volumes:
  - name: model-cache
    emptyDir: {}
  nodeSelector:
    edge-capable: "true"
  tolerations:
  - key: "edge-node"
    operator: "Equal"
    value: "true"
    effect: "NoSchedule"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

// TensorFlow Lite 边缘推理
import * as tflite from '@tensorflow/tfjs-tflite';

class EdgeInferenceEngine {
  private model: tflite.TFLiteModel;
  private interpreter: tflite.TFLiteInterpreter;

  async initialize(modelPath: string): Promise<void> {
    // 加载优化的 TFLite 模型
    this.model = await tflite.loadTFLiteModel(modelPath);
    this.interpreter = new tflite.TFLiteInterpreter(this.model);
  }

  async predict(input: Float32Array): Promise<number[]> {
    // 设置输入
    this.interpreter.setInput(0, input);

    // 运行推理
    await this.interpreter.invoke();

    // 获取输出
    const output = this.interpreter.getOutput(0);
    return Array.from(output.data);
  }

  // 批量推理
  async predictBatch(inputs: Float32Array[]): Promise<number[][]> {
    const results: number[][] = [];

    for (const input of inputs) {
      const result = await this.predict(input);
      results.push(result);
    }

    return results;
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

// 实时目标检测
class EdgeVideoAnalytics {
  private detector: ObjectDetector;
  private tracker: ObjectTracker;
  private alertManager: AlertManager;

  async processFrame(frame: ImageData): Promise<DetectionResult[]> {
    // 1. 目标检测
    const detections = await this.detector.detect(frame);

    // 2. 目标跟踪
    const tracks = await this.tracker.track(detections);

    // 3. 行为分析
    const anomalies = await this.analyzeBehavior(tracks);

    // 4. 触发告警
    if (anomalies.length > 0) {
      await this.alertManager.sendAlert(anomalies);
    }

    return detections;
  }

  private async analyzeBehavior(
    tracks: Track[]
  ): Promise<Anomaly[]> {
    const anomalies: Anomaly[] = [];

    for (const track of tracks) {
      // 检测异常行为
      if (this.isSuspiciousMovement(track)) {
        anomalies.push({
          type: 'suspicious_movement',
          trackId: track.id,
          confidence: track.confidence,
          timestamp: Date.now()
        });
      }

      // 检测入侵
      if (this.isIntrusion(track)) {
        anomalies.push({
          type: 'intrusion',
          trackId: track.id,
          confidence: track.confidence,
          timestamp: Date.now()
        });
      }
    }

    return anomalies;
  }
}

// 轻量级目标检测器
class LightweightDetector {
  private model: tflite.TFLiteModel;

  async detect(frame: ImageData): Promise<Detection[]> {
    // 预处理
    const processed = this.preprocess(frame);

    // 推理
    const outputs = await this.model.predict(processed);

    // 后处理
    const detections = this.postprocess(outputs);

    // 过滤低置信度检测
    return detections.filter(d => d.confidence > 0.5);
  }

  private preprocess(frame: ImageData): Float32Array {
    // 调整大小
    const resized = this.resize(frame, 320, 320);

    // 归一化
    const normalized = this.normalize(resized);

    return normalized;
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

# 边缘设备异常检测
import numpy as np
from scipy import signal
from sklearn.ensemble import IsolationForest

class EdgeAnomalyDetector:
    def __init__(self, model_path: str):
        self.model = self.load_model(model_path)
        self.threshold = 0.8

    def load_model(self, path: str):
        """加载异常检测模型"""
        import joblib
        return joblib.load(path)

    def process_sensor_data(self, data: np.ndarray) -> dict:
        """处理传感器数据"""
        # 特征提取
        features = self.extract_features(data)

        # 异常检测
        anomaly_score = self.model.score_samples(features.reshape(1, -1))[0]
        is_anomaly = anomaly_score < self.threshold

        # 频域分析
        freq_analysis = self.frequency_analysis(data)

        return {
            'anomaly_score': float(anomaly_score),
            'is_anomaly': bool(is_anomaly),
            'frequency_features': freq_analysis,
            'timestamp': time.time()
        }

    def extract_features(self, data: np.ndarray) -> np.ndarray:
        """提取统计特征"""
        features = []

        # 时域特征
        features.extend([
            np.mean(data),
            np.std(data),
            np.max(data),
            np.min(data),
            np.percentile(data, 25),
            np.percentile(data, 75)
        ])

        # 频域特征
        fft = np.fft.fft(data)
        features.extend([
            np.max(np.abs(fft)),
            np.mean(np.abs(fft)),
            np.sum(np.abs(fft) > np.mean(np.abs(fft)))
        ])

        return np.array(features)

    def frequency_analysis(self, data: np.ndarray) -> dict:
        """频域分析"""
        fft = np.fft.fft(data)
        freqs = np.fft.fftfreq(len(data))

        # 找到主频率
        dominant_freq = freqs[np.argmax(np.abs(fft))]

        return {
            'dominant_frequency': float(dominant_freq),
            'fft_magnitude': np.abs(fft).tolist()
        }

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100

// C++ 边缘感知系统
#include <opencv2/opencv.hpp>
#include <tensorflow/lite/interpreter.h>

class EdgePerceptionSystem {
private:
    std::unique_ptr<tflite::Interpreter> detector_;
    std::unique_ptr<tflite::Interpreter> segmenter_;

public:
    bool Initialize(const std::string& detector_path,
                   const std::string& segmenter_path) {
        // 加载检测模型
        detector_ = LoadModel(detector_path);
        if (!detector_) return false;

        // 加载分割模型
        segmenter_ = LoadModel(segmenter_path);
        if (!segmenter_) return false;

        return true;
    }

    PerceptionResult ProcessFrame(const cv::Mat& frame) {
        PerceptionResult result;

        // 1. 目标检测
        auto detections = DetectObjects(frame);
        result.objects = detections;

        // 2. 语义分割
        auto segmentation = SegmentFrame(frame);
        result.segmentation = segmentation;

        // 3. 深度估计
        auto depth = EstimateDepth(frame);
        result.depth_map = depth;

        // 4. 车道检测
        auto lanes = DetectLanes(frame);
        result.lanes = lanes;

        // 5. 融合结果
        result.fused = FuseResults(detections, segmentation, depth, lanes);

        return result;
    }

private:
    std::vector<Detection> DetectObjects(const cv::Mat& frame) {
        // 预处理
        cv::Mat processed = Preprocess(frame);

        // 设置输入
        FillInputTensor(detector_.get(), processed);

        // 推理
        detector_->Invoke();

        // 获取输出
        return ParseDetectionOutput(detector_.get());
    }

    cv::Mat SegmentFrame(const cv::Mat& frame) {
        // 类似检测流程
        cv::Mat processed = Preprocess(frame);
        FillInputTensor(segmenter_.get(), processed);
        segmenter_->Invoke();

        return ParseSegmentationOutput(segmenter_.get());
    }

    FusedResult FuseResults(
        const std::vector<Detection>& detections,
        const cv::Mat& segmentation,
        const cv::Mat& depth,
        const std::vector<Lane>& lanes
    ) {
        FusedResult fused;

        for (const auto& detection : detections) {
            // 结合深度信息
            float distance = GetDepthAtPoint(depth, detection.center);

            // 结合分割信息
            float road_confidence = GetSegmentationConfidence(
                segmentation,
                detection.bbox
            );

            fused.objects.push_back({
                .detection = detection,
                .depth = distance,
                .road_confidence = road_confidence
            });
        }

        return fused;
    }
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

# 模型量化工具
import tensorflow as tf
from tensorflow.lite.python import converter

class ModelQuantizer:
    def __init__(self, model_path: str):
        self.model = tf.keras.models.load_model(model_path)

    def quantize_dynamic(self) -> bytes:
        """动态量化：权重和激活都量化"""
        converter = tf.lite.TFLiteConverter.from_keras_model(self.model)

        # 优化选项
        converter.optimizations = [tf.lite.Optimize.DEFAULT]

        # 转换
        tflite_model = converter.convert()

        return tflite_model

    def quantize_full_integer(
        self,
        representative_data: np.ndarray
    ) -> bytes:
        """全整数量化"""
        converter = tf.lite.TFLiteConverter.from_keras_model(self.model)

        # 设置代表数据集
        def representative_dataset():
            for data in representative_data:
                yield [data.astype(np.float32)]

        converter.representative_dataset = representative_dataset
        converter.optimizations = [tf.lite.Optimize.DEFAULT]
        converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
        converter.inference_input_type = tf.int8
        converter.inference_output_type = tf.int8

        tflite_model = converter.convert()

        return tflite_model

    def quantize_float16(self) -> bytes:
        """Float16量化"""
        converter = tf.lite.TFLiteConverter.from_keras_model(self.model)

        converter.optimizations = [tf.lite.Optimize.DEFAULT]
        converter.target_spec.supported_types = [tf.float16]

        tflite_model = converter.convert()

        return tflite_model

    def save_model(self, tflite_model: bytes, output_path: str):
        """保存量化模型"""
        with open(output_path, 'wb') as f:
            f.write(tflite_model)

        # 打印模型大小
        print(f"Model size: {len(tflite_model) / 1024:.2f} KB")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

# 知识蒸馏
import torch
import torch.nn as nn

class DistillationTrainer:
    def __init__(self, teacher_model, student_model, temperature=3.0):
        self.teacher = teacher_model
        self.student = student_model
        self.temperature = temperature
        self.teacher.eval()

    def distillation_loss(
        self,
        student_output,
        teacher_output,
        labels,
        alpha=0.5
    ):
        """蒸馏损失函数"""
        # 软标签损失（来自教师）
        soft_loss = nn.KLDivLoss(reduction='batchmean')(
            nn.functional.log_softmax(student_output / self.temperature, dim=1),
            nn.functional.softmax(teacher_output / self.temperature, dim=1)
        ) * (self.temperature ** 2)

        # 硬标签损失（来自真实标签）
        hard_loss = nn.CrossEntropyLoss()(student_output, labels)

        # 组合损失
        return alpha * soft_loss + (1 - alpha) * hard_loss

    def train_step(self, inputs, labels, optimizer):
        """训练步骤"""
        # 教师模型推理（不计算梯度）
        with torch.no_grad():
            teacher_output = self.teacher(inputs)

        # 学生模型前向传播
        student_output = self.student(inputs)

        # 计算损失
        loss = self.distillation_loss(
            student_output,
            teacher_output,
            labels
        )

        # 反向传播
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        return loss.item()

    def train(self, train_loader, epochs, learning_rate):
        """训练学生模型"""
        optimizer = torch.optim.Adam(
            self.student.parameters(),
            lr=learning_rate
        )

        for epoch in range(epochs):
            total_loss = 0

            for inputs, labels in train_loader:
                loss = self.train_step(inputs, labels, optimizer)
                total_loss += loss

            avg_loss = total_loss / len(train_loader)
            print(f"Epoch {epoch + 1}/{epochs}, Loss: {avg_loss:.4f}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

# 联邦学习
import numpy as np
from typing import List, Dict

class FederatedLearningClient:
    def __init__(self, model, client_id: str):
        self.model = model
        self.client_id = client_id

    def local_train(
        self,
        local_data: np.ndarray,
        local_labels: np.ndarray,
        epochs: int = 5
    ) -> Dict:
        """本地训练"""
        initial_weights = self.model.get_weights()

        # 本地训练
        history = self.model.fit(
            local_data,
            local_labels,
            epochs=epochs,
            verbose=0
        )

        # 计算权重更新
        updated_weights = self.model.get_weights()
        weight_updates = [
            updated - initial
            for updated, initial in zip(updated_weights, initial_weights)
        ]

        return {
            'client_id': self.client_id,
            'weight_updates': weight_updates,
            'num_samples': len(local_data),
            'metrics': history.history
        }

class FederatedLearningServer:
    def __init__(self, global_model):
        self.global_model = global_model
        self.client_updates = []

    def aggregate_updates(
        self,
        client_updates: List[Dict],
        aggregation: str = 'weighted_avg'
    ):
        """聚合客户端更新"""
        if aggregation == 'weighted_avg':
            # 加权平均聚合（FedAvg）
            total_samples = sum(
                update['num_samples']
                for update in client_updates
            )

            aggregated_weights = []

            for layer_idx in range(len(self.global_model.get_weights())):
                layer_weights = []
                for update in client_updates:
                    weight = update['weight_updates'][layer_idx]
                    weight = weight * (update['num_samples'] / total_samples)
                    layer_weights.append(weight)

                aggregated_weights.append(
                    np.sum(layer_weights, axis=0)
                )

            # 更新全局模型
            current_weights = self.global_model.get_weights()
            new_weights = [
                current + update
                for current, update in zip(current_weights, aggregated_weights)
            ]

            self.global_model.set_weights(new_weights)

        return self.global_model

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// 智能数据路由
class DataRouter {
  async routeData(data: SensorData): Promise<RouteDecision> {
    const analysis = await this.analyzeData(data);

    // 高优先级、需要全局视角的 -> 云端
    if (analysis.requiresGlobalView || analysis.priority === 'high') {
      return {
        destination: 'cloud',
        reason: analysis.reason,
        compressed: false
      };
    }

    // 中等优先级 -> 边缘聚合
    if (analysis.priority === 'medium') {
      return {
        destination: 'edge',
        action: 'aggregate',
        windowSize: 60 // 60秒聚合窗口
      };
    }

    // 低优先级 -> 本地处理
    return {
      destination: 'local',
      action: 'process',
      ttl: 3600 // 1小时过期
    };
  }

  private async analyzeData(data: SensorData): Promise<DataAnalysis> {
    // 计算数据特征
    const features = this.extractFeatures(data);

    // 本地模型推理
    const prediction = await this.localModel.predict(features);

    return {
      priority: this.calculatePriority(data, prediction),
      requiresGlobalView: this.needsGlobalContext(prediction),
      reason: this.explainDecision(prediction)
    };
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

// 模型版本管理
class ModelSyncManager {
  private localVersion: number;
  private cloudVersion: number;
  private syncInProgress: boolean = false;

  async syncModel(): Promise<SyncResult> {
    if (this.syncInProgress) {
      throw new Error('Sync already in progress');
    }

    this.syncInProgress = true;

    try {
      // 1. 检查云端版本
      const cloudInfo = await this.checkCloudVersion();

      if (cloudInfo.version <= this.localVersion) {
        return {
          status: 'up_to_date',
          localVersion: this.localVersion
        };
      }

      // 2. 下载增量更新
      const delta = await this.downloadDelta(
        this.localVersion,
        cloudInfo.version
      );

      // 3. 应用更新
      await this.applyDelta(delta);

      // 4. 验证模型
      const isValid = await this.validateModel();

      if (!isValid) {
        await this.rollback();
        throw new Error('Model validation failed');
      }

      // 5. 更新版本号
      this.localVersion = cloudInfo.version;

      return {
        status: 'updated',
        localVersion: this.localVersion,
        previousVersion: this.localVersion - 1
      };

    } finally {
      this.syncInProgress = false;
    }
  }

  private async applyDelta(delta: ModelDelta): Promise<void> {
    // 增量更新模型权重
    const currentWeights = await this.model.getWeights();

    const updatedWeights = currentWeights.map((layer, index) => {
      if (delta.updates[index]) {
        return layer + delta.updates[index];
      }
      return layer;
    });

    await this.model.setWeights(updatedWeights);
  }
}