边缘计算与轻量化 AI Agent Harness Engineering

1. 标题 (Title)

• 边缘计算新篇章：构建高效轻量化 AI Agent 的 Harness 工程实践
• 当 AI 遇到边缘：深入浅出轻量化 Agent 部署与优化全攻略
• 告别云端延迟：边缘计算下的 AI Agent Harness 工程从入门到精通
• 轻量化 AI Agent 实战：边缘计算环境下的工程化部署指南

---

2. 引言 (Introduction)

痛点引入 (Hook)

你是否曾经遇到过这样的场景：在工厂车间里，智能质检系统需要实时检测产品缺陷，但由于数据传输到云端处理存在延迟，导致生产线不得不降低速度；在智能安防监控中，关键事件的响应时间因为网络波动而错过了最佳处理时机；在自动驾驶汽车中，AI决策系统必须在毫秒内做出反应，但依赖云端计算是根本不现实的。

随着人工智能技术的快速发展，越来越多的智能应用需要实时、高效地处理数据。然而，传统的云端计算模式在面对这些场景时，往往面临着网络延迟、带宽限制、数据隐私安全等诸多挑战。据统计，到2025年，全球将有超过75%的数据需要在网络边缘进行处理。这意味着，我们需要一种新的计算范式来应对这些挑战。

文章内容概述 (What)

本文将带你深入探讨边缘计算与轻量化AI Agent的结合点，重点讲解如何通过Harness工程方法，构建、部署和管理高效的边缘AI系统。我们将从基础概念开始，逐步深入到实际工程实践，包括环境搭建、模型轻量化、Agent设计、部署策略、监控优化等全流程。

读者收益 (Why)

读完本文，你将能够：

• 理解边缘计算与轻量化AI Agent的核心概念和价值
• 掌握模型轻量化的主要技术和方法
• 学会设计和实现边缘AI Agent的Harness工程
• 具备在实际项目中部署和优化边缘AI系统的能力
• 了解边缘计算与AI结合的最新发展趋势和最佳实践

---

3. 准备工作 (Prerequisites)

在开始本文的学习之前，建议你具备以下知识和环境：

技术栈/知识：

• 编程基础： 熟练掌握Python编程语言，了解基本的软件工程概念
• AI/ML基础： 了解机器学习和深度学习的基本概念，熟悉常见的神经网络架构
• 系统知识： 了解操作系统、网络基础、容器化技术（如Docker）的基本概念
• 边缘计算基础： 对边缘计算概念有初步了解者更佳

环境/工具：

• 开发环境： 一台配置较好的开发机（建议16GB以上内存，现代多核处理器）
• 操作系统： Linux（推荐Ubuntu 20.04+）或macOS，Windows用户建议使用WSL2
• 软件安装：
  • Python 3.8+
  • Docker 和 Docker Compose
  • Git版本控制工具
  • 常用的机器学习库（PyTorch/TensorFlow）
• 边缘设备（可选但推荐）： 如树莓派、Jetson Nano或其他边缘计算设备，用于实际部署测试

---

4. 核心内容：手把手实战 (Step-by-Step Tutorial)

步骤一：理解边缘计算与轻量化AI Agent基础

核心概念

在开始实战之前，我们首先需要明确几个核心概念：

边缘计算（Edge Computing）：边缘计算是一种分布式计算范式，它将数据处理和存储任务从云端数据中心下移到网络边缘，即靠近数据源或终端用户的位置。这样可以减少数据传输延迟，降低带宽消耗，提高数据隐私安全性。

轻量化AI（Lightweight AI）：轻量化AI是指通过模型压缩、量化、剪枝等技术，减小AI模型的体积和计算复杂度，使其能够在资源受限的设备上高效运行。

AI Agent：AI Agent是一种能够感知环境、做出决策并执行行动的智能实体。在边缘计算场景中，AI Agent通常部署在边缘设备上，负责处理本地数据并做出实时决策。

Harness Engineering：Harness在这里指的是一套完整的工程化框架和方法论，用于高效地开发、部署、监控和管理AI Agent。它包括持续集成/持续部署（CI/CD）、版本控制、监控告警、故障恢复等一系列工程实践。

问题背景

随着物联网（IoT）设备的普及，海量数据在网络边缘产生。根据Gartner的预测，到2025年，75%的企业数据将在传统数据中心或云端之外创建和处理。同时，AI技术的快速发展使得我们能够从这些数据中提取有价值的信息。然而，将所有数据传输到云端处理面临以下挑战：

1. 延迟问题：对于需要实时响应的应用（如自动驾驶、工业控制），云端处理的延迟是不可接受的。
2. 带宽限制：传输海量原始数据需要巨大的网络带宽，成本高昂。
3. 隐私安全：敏感数据（如医疗记录、个人影像）传输到云端存在安全风险。
4. 可靠性问题：依赖网络连接意味着一旦网络中断，系统将无法工作。

边缘计算与轻量化AI的结合，正是为了解决这些问题而提出的解决方案。

概念结构与核心要素组成

边缘计算系统通常由以下几个核心层次组成：

1. 终端设备层：包括各种传感器、摄像头、智能终端等，负责数据采集和初步处理。
2. 边缘节点层：包括边缘服务器、网关等，负责本地数据处理和决策。
3. 边缘聚合层：负责多个边缘节点的数据聚合和协同处理。
4. 云端层：负责全局数据分析、模型训练和长期存储。

轻量化AI Agent通常包含以下核心组件：

1. 感知模块：负责采集和预处理环境数据。
2. 推理引擎：运行轻量化AI模型，进行数据分析和决策。
3. 执行模块：根据决策结果执行相应的操作。
4. 通信模块：负责与其他Agent或云端进行通信。
5. 管理模块：负责Agent的生命周期管理、配置更新等。

概念之间的关系

为了更好地理解这些概念之间的关系，我们可以通过以下表格和图表来展示：

边缘计算 vs 云端计算对比

特性	边缘计算	云端计算
数据处理位置	靠近数据源	远程数据中心
延迟	低（毫秒级）	高（秒级）
带宽需求	低	高
隐私安全性	高（数据本地处理）	中（数据传输到云端）
可靠性	高（不依赖网络）	中（依赖网络）
计算能力	有限	强大
可扩展性	有限	强
适用场景	实时响应、数据敏感	批量处理、复杂计算

边缘AI系统架构图

边缘计算与AI Agent交互关系图

数学模型：边缘计算中的延迟优化

在边缘计算系统中，延迟优化是一个核心问题。我们可以建立一个简单的数学模型来描述这个问题。

假设我们有一个计算任务，可以选择在边缘节点或云端执行。我们定义：

• $D$：数据大小（单位：MB）
• $C$：计算复杂度（单位：CPU cycles）
• $f_e$：边缘节点的计算能力（单位：CPU cycles/s）
• $f_c$：云端的计算能力（单位：CPU cycles/s）
• $r$：网络传输速率（单位：MB/s）
• $p_e$：边缘节点的功率消耗（单位：W）
• $p_c$：云端的功率消耗（单位：W）
• $p_t$：数据传输的功率消耗（单位：W）

那么，在边缘节点执行任务的总延迟为：
$$T_e=\frac{C}{f_e}$$

在云端执行任务的总延迟为：
$$T_c=\frac{D}{r}+\frac{C}{f_c}$$

我们的目标是最小化延迟，因此决策规则为：
$$\text{如果 }{T}_e<T_c\text{，则在边缘执行；否则在云端执行}$$

即：
$$\frac{C}{f_e}<\frac{D}{r}+\frac{C}{f_c}$$

整理得：
$$C\left(\frac{1}{f_e}-\frac{1}{f_c}\right)<\frac{D}{r}$$

这个简单的模型告诉我们，当计算复杂度相对较低、数据量相对较大、网络传输速率相对较慢时，边缘计算更有优势。

如果同时考虑功率消耗，总成本（以延迟和功耗的加权和表示）可以表示为：
$${\text{Cost}}_e=\alpha T_e+\beta p_e{T}_e$$
$${\text{Cost}}_c=\alpha T_c+\beta (p_t\frac{D}{r}+p_c\frac{C}{f_c})$$

其中，$\alpha$和$\beta$分别是延迟和功耗的权重因子。我们可以根据实际应用场景调整这些权重，做出最优的计算卸载决策。

---

步骤二：环境搭建与工具准备

项目介绍

在本文中，我们将构建一个实际的边缘AI应用场景：智能监控系统。这个系统将能够在边缘设备上实时检测视频流中的异常事件（如人员闯入、物品移动等），并在本地进行初步处理，只在必要时将关键信息上传到云端。

通过这个项目，我们将学习如何：

1. 搭建边缘AI开发环境
2. 选择和准备合适的预训练模型
3. 对模型进行轻量化处理
4. 设计和实现AI Agent
5. 部署到边缘设备
6. 监控和优化系统性能

环境安装

首先，我们需要设置开发环境。我们将使用Python作为主要编程语言，结合Docker进行容器化部署。

1. 基础环境准备

确保你的系统已安装Python 3.8或更高版本：

python3 --version

如果没有安装，请从Python官网下载安装。

接下来，安装Docker和Docker Compose：

# Ubuntu/Debian
curl -fsSL https://get.docker.com -o get-docker.sh
sudo sh get-docker.sh
sudo usermod -aG docker $USER

# 安装Docker Compose
sudo curl -L "https://github.com/docker/compose/releases/latest/download/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose

验证安装：

docker --version
docker-compose --version

2. 创建项目目录结构

我们将按照以下结构组织项目：

edge-ai-agent/
├── data/                   # 数据目录
│   ├── raw/                # 原始数据
│   ├── processed/          # 处理后的数据
│   └── models/             # 模型文件
├── src/                    # 源代码
│   ├── agent/              # AI Agent实现
│   ├── model/              # 模型相关代码
│   ├── utils/              # 工具函数
│   └── api/                # API接口
├── tests/                  # 测试代码
├── config/                 # 配置文件
├── docker/                 # Docker相关文件
├── docs/                   # 文档
├── requirements.txt        # Python依赖
├── docker-compose.yml      # Docker Compose配置
└── README.md               # 项目说明

创建目录结构：

mkdir -p edge-ai-agent/{data/{raw,processed,models},src/{agent,model,utils,api},tests,config,docker,docs}
cd edge-ai-agent

3. 安装Python依赖

创建requirements.txt文件：

# 基础依赖
numpy>=1.21.0
pandas>=1.3.0
matplotlib>=3.4.0
seaborn>=0.11.0
scikit-learn>=0.24.0

# 深度学习框架
torch>=1.9.0
torchvision>=0.10.0
tensorflow>=2.6.0
onnx>=1.10.0
onnxruntime>=1.8.0

# 模型轻量化工具
torch-pruning>=0.2.7
distiller>=0.3.1
tensorflow-model-optimization>=0.7.0

# 边缘计算相关
paho-mqtt>=1.5.1
opencv-python>=4.5.0
flask>=2.0.0
fastapi>=0.68.0
uvicorn>=0.15.0

# 监控和日志
prometheus-client>=0.11.0
python-json-logger>=2.0.1
structlog>=21.1.0

# 其他工具
python-dotenv>=0.19.0
pyyaml>=5.4.0
click>=8.0.0
tqdm>=4.62.0

安装依赖：

python3 -m venv venv
source venv/bin/activate  # Linux/macOS
# 或
.\venv\Scripts\activate  # Windows

pip install -r requirements.txt

4. 配置Docker环境

创建Dockerfile文件（放在docker/目录下）：

# 基础镜像
FROM nvidia/cuda:11.4.0-runtime-ubuntu20.04

# 设置环境变量
ENV DEBIAN_FRONTEND=noninteractive
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    python3.8 \
    python3-pip \
    python3-dev \
    git \
    curl \
    libgl1-mesa-glx \
    libglib2.0-0 \
    && rm -rf /var/lib/apt/lists/*

# 设置工作目录
WORKDIR /app

# 复制项目文件
COPY requirements.txt .
COPY src ./src
COPY config ./config

# 安装Python依赖
RUN pip3 install --no-cache-dir -r requirements.txt

# 创建非root用户
RUN useradd -m edgeai && chown -R edgeai /app
USER edgeai

# 暴露端口
EXPOSE 8000

# 运行命令
CMD ["uvicorn", "src.api.main:app", "--host", "0.0.0.0", "--port", "8000"]

创建docker-compose.yml文件：

version: '3.8'

services:
  edge-agent:
    build:
      context: .
      dockerfile: docker/Dockerfile
    container_name: edge-ai-agent
    ports:
      - "8000:8000"
    volumes:
      - ./data:/app/data
      - ./config:/app/config
    environment:
      - CONFIG_PATH=/app/config/config.yaml
    restart: unless-stopped
    networks:
      - edge-net

  mqtt-broker:
    image: eclipse-mosquitto:2.0
    container_name: edge-mqtt-broker
    ports:
      - "1883:1883"
      - "9001:9001"
    volumes:
      - ./config/mosquitto.conf:/mosquitto/config/mosquitto.conf
      - ./data/mosquitto/data:/mosquitto/data
      - ./data/mosquitto/log:/mosquitto/log
    restart: unless-stopped
    networks:
      - edge-net

  prometheus:
    image: prom/prometheus:latest
    container_name: edge-prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
      - ./data/prometheus:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.path=/prometheus'
      - '--web.console.libraries=/usr/share/prometheus/console_libraries'
      - '--web.console.templates=/usr/share/prometheus/consoles'
    restart: unless-stopped
    networks:
      - edge-net

  grafana:
    image: grafana/grafana:latest
    container_name: edge-grafana
    ports:
      - "3000:3000"
    volumes:
      - ./data/grafana:/var/lib/grafana
    environment:
      - GF_SECURITY_ADMIN_USER=admin
      - GF_SECURITY_ADMIN_PASSWORD=admin123
    depends_on:
      - prometheus
    restart: unless-stopped
    networks:
      - edge-net

networks:
  edge-net:
    driver: bridge

现在，我们已经完成了基础环境的搭建。接下来，我们将开始准备模型和实现AI Agent。

---

步骤三：模型准备与轻量化

问题背景

在边缘计算环境中，由于计算资源和能源受限，我们不能直接使用大型、复杂的深度学习模型。我们需要对模型进行轻量化处理，使其能够在边缘设备上高效运行，同时保持尽可能高的准确率。

模型轻量化技术概述

模型轻量化主要有以下几种技术：

1. 模型剪枝（Pruning）：移除模型中不重要的权重或神经元。
2. 量化（Quantization）：将模型参数从高精度（如FP32）转换为低精度（如INT8）。
3. 知识蒸馏（Knowledge Distillation）：训练一个小模型（学生模型）来模仿大模型（教师模型）的行为。
4. 神经架构搜索（Neural Architecture Search, NAS）：自动搜索适合边缘设备的高效模型架构。
5. 模型压缩（Model Compression）：使用编码技术进一步减小模型大小。

算法流程图：模型轻量化流程

选择预训练模型

对于我们的智能监控场景，我们将使用目标检测模型。YOLO（You Only Look Once）系列模型因其速度和准确率的平衡而广受欢迎。我们将选择YOLOv5作为基础模型，因为它有较好的社区支持，并且已经有一些针对边缘设备优化的版本。

首先，让我们下载YOLOv5模型：

# src/model/model_loader.py
import os
import torch
from pathlib import Path

def load_pretrained_model(model_name='yolov5s', device='cpu'):
    """
    加载预训练的YOLOv5模型
    
    参数:
        model_name: 模型名称 (yolov5n, yolov5s, yolov5m, yolov5l, yolov5x)
        device: 运行设备 (cpu, cuda)
    
    返回:
        加载的模型
    """
    # 检查是否有本地模型文件
    model_path = Path(f'data/models/{model_name}.pt')
    
    if not model_path.exists():
        print(f"下载模型 {model_name}...")
        # 从torch hub加载模型
        model = torch.hub.load('ultralytics/yolov5', model_name, pretrained=True)
        # 保存模型到本地
        os.makedirs(model_path.parent, exist_ok=True)
        torch.save(model, model_path)
    else:
        print(f"从本地加载模型 {model_path}...")
        model = torch.load(model_path, map_location=device)
    
    # 设置模型为评估模式
    model.eval()
    model.to(device)
    
    return model

if __name__ == "__main__":
    # 测试加载模型
    model = load_pretrained_model('yolov5s', 'cpu')
    print("模型加载成功!")
    print(f"模型结构: {model}")

模型剪枝

模型剪枝是一种通过移除模型中不重要的连接或神经元来减小模型大小的技术。我们将使用结构化剪枝，即移除整个通道或层，这样可以更好地利用硬件加速。

# src/model/pruning.py
import torch
import torch.nn as nn
import torch_pruning as tp
from pathlib import Path

def prune_model(model, example_inputs, pruning_ratio=0.5, save_path=None):
    """
    对模型进行剪枝
    
    参数:
        model: 原始模型
        example_inputs: 示例输入，用于分析模型结构
        pruning_ratio: 剪枝比例 (0-1)
        save_path: 剪枝后模型保存路径
    
    返回:
        剪枝后的模型
    """
    print(f"开始剪枝，剪枝比例: {pruning_ratio}")
    
    # 原始模型大小
    orig_params = sum(p.numel() for p in model.parameters())
    print(f"原始模型参数数量: {orig_params:,}")
    
    # 定义要忽略的层（通常是最后的分类层）
    ignored_layers = []
    for m in model.modules():
        if isinstance(m, nn.Linear) and m.out_features == 80:  # COCO数据集有80个类别
            ignored_layers.append(m)
    
    # 创建剪枝器
    pruner = tp.pruner.MagnitudePruner(
        model,
        example_inputs,
        global_pruning=False,  # 每层独立剪枝
        pruning_ratio=pruning_ratio,
        iterative_steps=1,  # 一次完成剪枝
        ignored_layers=ignored_layers,
    )
    
    # 执行剪枝
    pruner.step()
    
    # 剪枝后模型大小
    pruned_params = sum(p.numel() for p in model.parameters())
    print(f"剪枝后模型参数数量: {pruned_params:,}")
    print(f"参数减少: {(1 - pruned_params/orig_params)*100:.2f}%")
    
    # 保存剪枝后的模型
    if save_path:
        Path(save_path).parent.mkdir(parents=True, exist_ok=True)
        torch.save(model, save_path)
        print(f"剪枝后模型已保存到: {save_path}")
    
    return model

if __name__ == "__main__":
    # 测试剪枝功能
    from model_loader import load_pretrained_model
    
    # 加载模型
    model = load_pretrained_model('yolov5s', 'cpu')
    
    # 创建示例输入
    example_inputs = torch.randn(1, 3, 640, 640)
    
    # 剪枝模型
    pruned_model = prune_model(
        model, 
        example_inputs, 
        pruning_ratio=0.3,
        save_path='data/models/yolov5s_pruned.pt'
    )

模型量化

模型量化是将模型参数从高精度浮点数（如FP32）转换为低精度表示（如INT8）的过程，可以显著减小模型大小并提高推理速度。

# src/model/quantization.py
import torch
import torch.quantization
from pathlib import Path

def quantize_model(model, example_inputs, save_path=None):
    """
    对模型进行量化
    
    参数:
        model: 原始模型
        example_inputs: 示例输入，用于校准
        save_path: 量化后模型保存路径
    
    返回:
        量化后的模型
    """
    print("开始量化模型...")
    
    # 原始模型大小
    orig_params = sum(p.numel() for p in model.parameters())
    orig_size = sum(p.numel() * p.element_size() for p in model.parameters())
    print(f"原始模型参数数量: {orig_params:,}")
    print(f"原始模型大小: {orig_size / 1024 / 1024:.2f} MB")
    
    # 设置模型为评估模式
    model.eval()
    
    # 融合层（提高性能和数值精度）
    model = torch.quantization.fuse_modules(model, [['conv', 'bn', 'relu']], inplace=False)
    
    # 设置量化配置
    model.qconfig = torch.quantization.get_default_qconfig('qnnpack')
    
    # 准备量化
    torch.quantization.prepare(model, inplace=True)
    
    # 校准（使用示例输入）
    print("进行模型校准...")
    with torch.no_grad():
        model(example_inputs)
    
    # 转换为量化模型
    quantized_model = torch.quantization.convert(model, inplace=False)
    
    # 量化后模型大小
    quantized_size = 0
    for param in quantized_model.parameters():
        quantized_size += param.numel() * param.element_size()
    # 还需要考虑量化后的参数（不在parameters()中）
    for name, module in quantized_model.named_modules():
        if hasattr(module, 'weight'):
            if hasattr(module.weight, 'q_scale'):
                # 这是一个量化的权重
                quantized_size += module.weight().numel() * module.weight().element_size()
    
    print(f"量化后模型大小: {quantized_size / 1024 / 1024:.2f} MB")
    print(f"模型大小减少: {(1 - quantized_size/orig_size)*100:.2f}%")
    
    # 保存量化后的模型
    if save_path:
        Path(save_path).parent.mkdir(parents=True, exist_ok=True)
        torch.save(quantized_model, save_path)
        print(f"量化后模型已保存到: {save_path}")
    
    return quantized_model

if __name__ == "__main__":
    # 测试量化功能
    from model_loader import load_pretrained_model
    
    # 加载模型
    model = load_pretrained_model('yolov5s', 'cpu')
    
    # 创建示例输入
    example_inputs = torch.randn(1, 3, 640, 640)
    
    # 量化模型
    quantized_model = quantize_model(
        model, 
        example_inputs,
        save_path='data/models/yolov5s_quantized.pt'
    )

知识蒸馏

知识蒸馏是一种通过训练一个小模型（学生模型）来模仿大模型（教师模型）的行为的技术。这种方法可以在保持较高准确率的同时，显著减小模型大小。

# src/model/distillation.py
import torch
import torch.nn as nn
import torch.optim as optim
from pathlib import Path
from tqdm import tqdm

def distill_knowledge(teacher_model, student_model, dataloader, 
                      temperature=3.0, alpha=0.7, epochs=10, 
                      device='cpu', save_path=None):
    """
    知识蒸馏：训练学生模型模仿教师模型
    
    参数:
        teacher_model: 教师模型（大模型）
        student_model: 学生模型（小模型）
        dataloader: 训练数据加载器
        temperature: 温度参数，用于软化教师模型输出
        alpha: 蒸馏损失权重
        epochs: 训练轮数
        device: 训练设备
        save_path: 保存路径
    
    返回:
        训练好的学生模型
    """
    print(f"开始知识蒸馏，温度: {temperature}, alpha: {alpha}")
    
    # 设置模型模式
    teacher_model.eval()
    teacher_model.to(device)
    student_model.train()
    student_model.to(device)
    
    # 定义损失函数
    hard_loss_fn = nn.CrossEntropyLoss()
    soft_loss_fn = nn.KLDivLoss(reduction='batchmean')
    
    # 优化器
    optimizer = optim.Adam(student_model.parameters(), lr=1e-4)
    
    # 训练循环
    for epoch in range(epochs):
        total_loss = 0.0
        for inputs, labels in tqdm(dataloader, desc=f"Epoch {epoch+1}/{epochs}"):
            inputs, labels = inputs.to(device), labels.to(device)
            
            # 前向传播
            with torch.no_grad():
                teacher_outputs = teacher_model(inputs)
            
            student_outputs = student_model(inputs)
            
            # 计算硬损失（学生模型与真实标签）
            hard_loss = hard_loss_fn(student_outputs, labels)
            
            # 计算软损失（学生模型与教师模型）
            # 使用温度参数软化输出
            soft_teacher = torch.softmax(teacher_outputs / temperature, dim=1)
            log_soft_student = torch.log_softmax(student_outputs / temperature, dim=1)
            soft_loss = soft_loss_fn(log_soft_student, soft_teacher) * (temperature ** 2)
            
            # 组合损失
            loss = alpha * soft_loss + (1 - alpha) * hard_loss
            
            # 反向传播和优化
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item() * inputs.size(0)
        
        epoch_loss = total_loss / len(dataloader.dataset)
        print(f"Epoch {epoch+1}/{epochs}, Loss: {epoch_loss:.4f}")
    
    # 保存学生模型
    if save_path:
        Path(save_path).parent.mkdir(parents=True, exist_ok=True)
        torch.save(student_model, save_path)
        print(f"蒸馏后模型已保存到: {save_path}")
    
    return student_model

if __name__ == "__main__":
    # 注意：这里只是示例，实际使用时需要准备适当的数据集和学生模型
    print("知识蒸馏示例 - 实际使用时需要准备数据集和学生模型")

模型评估与选择

在应用了各种轻量化技术后，我们需要评估模型的性能，包括准确率、推理速度、模型大小等指标，然后选择最适合我们边缘设备的模型。

# src/model/evaluation.py
import time
import torch
import numpy as np
from pathlib import Path
import matplotlib.pyplot as plt

def evaluate_model(model, test_loader, device='cpu'):
    """
    评估模型性能
    
    参数:
        model: 要评估的模型
        test_loader: 测试数据加载器
        device: 运行设备
    
    返回:
        包含评估指标的字典
    """
    model.eval()
    model.to(device)
    
    # 评估指标
    total_time = 0.0
    total_samples = 0
    all_predictions = []
    all_labels = []
    
    with torch.no_grad():
        for inputs, labels in test_loader:
            inputs, labels = inputs.to(device), labels.to(device)
            batch_size = inputs.size(0)
            
            # 测量推理时间
            start_time = time.time()
            outputs = model(inputs)
            end_time = time.time()
            
            # 记录时间和样本数
            total_time += end_time - start_time
            total_samples += batch_size
            
            # 收集预测和标签（用于计算准确率）
            _, predicted = torch.max(outputs.data, 1)
            all_predictions.extend(predicted.cpu().numpy())
            all_labels.extend(labels.cpu().numpy())
    
    # 计算指标
    avg_time_per_sample = total_time / total_samples
    fps = total_samples / total_time
    accuracy = np.mean(np.array(all_predictions) == np.array(all_labels))
    
    # 计算模型大小
    model_size = 0
    for param in model.parameters():
        model_size += param.numel() * param.element_size()
    model_size_mb = model_size / (1024 * 1024)
    
    metrics = {
        'accuracy': accuracy,
        'avg_time_per_sample_ms': avg_time_per_sample * 1000,
        'fps': fps,
        'model_size_mb': model_size_mb,
        'total_params': sum(p.numel() for p in model.parameters())
    }
    
    return metrics

def compare_models(model_metrics, save_path=None):
    """
    比较多个模型的性能
    
    参数:
        model_metrics: 字典，键为模型名称，值为评估指标字典
        save_path: 可视化结果保存路径
    """
    # 提取数据
    model_names = list(model_metrics.keys())
    accuracies = [m['accuracy'] for m in model_metrics.values()]
    fps = [m['fps'] for m in model_metrics.values()]
    sizes_mb = [m['model_size_mb'] for m in model_metrics.values()]
    
    # 创建可视化
    fig, axes = plt.subplots(1, 3, figsize=(18, 5))
    
    # 准确率对比
    axes[0].bar(model_names, accuracies, color='skyblue')
    axes[0].set_title('模型准确率对比')
    axes[0].set_ylabel('准确率')
    axes[0].tick_params(axis='x', rotation=45)
    
    # 速度对比
    axes[1].bar(model_names, fps, color='lightgreen')
    axes[1].set_title('模型推理速度对比 (FPS)')
    axes[1].set_ylabel('FPS')
    axes[1].tick_params(axis='x', rotation=45)
    
    # 模型大小对比
    axes[2].bar(model_names, sizes_mb, color='salmon')
    axes[2].set_title('模型大小对比 (MB)')
    axes[2].set_ylabel('大小 (MB)')
    axes[2].tick_params(axis='x', rotation=45)
    
    plt.tight_layout()
    
    if save_path:
        Path(save_path).parent.mkdir(parents=True, exist_ok=True)
        plt.savefig(save_path)
        print(f"模型对比图已保存到: {save_path}")
    
    plt.show()
    
    # 打印表格
    print("\n模型性能对比表:")
    print("-" * 80)
    print(f"{'模型名称':<20} {'准确率':<10} {'FPS':<10} {'大小(MB)':<10} {'参数量':<15}")
    print("-" * 80)
    for name, metrics in model_metrics.items():
        print(f"{name:<20} {metrics['accuracy']:<10.4f} {metrics['fps']:<10.2f} "
              f"{metrics['model_size_mb']:<10.2f} {metrics['total_params']:<15,}")
    print("-" * 80)

if __name__ == "__main__":
    # 示例：评估并比较模型
    print("模型评估示例 - 实际使用时需要准备测试数据")

通过上述步骤，我们可以准备和优化适用于边缘设备的AI模型。接下来，我们将设计和实现AI Agent。

---

步骤四：设计和实现AI Agent

核心概念

AI Agent是一种能够感知环境、做出决策并执行行动的智能实体。在边缘计算场景中，AI Agent需要具备以下特点：

1. 轻量级：能够在资源受限的设备上运行。
2. 实时性：能够快速响应环境变化。
3. 自主性：能够在没有云端干预的情况下自主运行。
4. 协作性：能够与其他Agent或云端进行协作。
5. 自适应性：能够根据环境变化调整自己的行为。

AI Agent架构设计

我们的边缘AI Agent将采用模块化设计，包含以下核心组件：

实现AI Agent核心组件

现在，让我们开始实现AI Agent的各个核心组件。

# src/agent/base_agent.py
import os
import time
import json
import logging
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Dict, Any, Optional, List
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class BaseAgent(ABC):
    """
    AI Agent 基类
    """
    
    def __init__(self, agent_id: str, config_path: Optional[str] = None):
        """
        初始化Agent
        
        参数:
            agent_id: Agent唯一标识符
            config_path: 配置文件路径
        """
        self.agent_id = agent_id
        self.config = self._load_config(config_path)
        self.state = "initialized"  # initialized, running, paused, stopped
        self.start_time = None
        self.metrics = {
            "inference_count": 0,
            "total_inference_time": 0,
            "decisions_made": 0,
            "actions_taken": 0,
            "messages_sent": 0,
            "messages_received": 0
        }
        
        # 初始化组件
        self.perception_module = None
        self.preprocessing_module = None
        self.inference_engine = None
        self.decision_module = None
        self.execution_module = None
        self.communication_module = None
        self.storage_module = None
        self.management_module = None
        
        logger.info(f"Agent {self.agent_id} 初始化完成")
    
    def _load_config(self, config_path: Optional[str]) -> Dict[str, Any]:
        """
        加载配置文件
        
        参数:
            config_path: 配置文件路径
        
        返回:
            配置字典
        """
        default_config = {
            "agent": {
                "name": "BaseAgent",
                "version": "1.0.0"
            },
            "perception": {
                "enabled": True,
                "sample_rate": 1.0  # Hz
            },
            "inference": {
                "device": "cpu",
                "model_path": None
            },
            "communication": {
                "enabled": False,
                "broker": "localhost",
                "port": 1883
            },
            "storage": {
                "data_dir": "./data/agent_data",
                "log_dir": "./data/logs"
            }
        }
        
        if config_path and Path(config_path).exists():
            try:
                with open(config_path, 'r') as f:
                    user_config = json.load(f)
                # 合并配置
                for key, value in user_config.items():
                    if key in default_config and isinstance(value, dict):
                        default_config[key].update(value)
                    else:
                        default_config[key] = value
                logger.info(f"配置文件已加载: {config_path}")
            except Exception as e:
                logger.error(f"加载配置文件失败: {e}")
        
        return default_config
    
    @abstractmethod
    def _init_components(self):
        """初始化各个组件，子类需要实现此方法"""
        pass
    
    def start(self):
        """启动Agent"""
        if self.state == "running":
            logger.warning(f"Agent {self.agent_id} 已经在运行中")
            return
        
        logger.info(f"正在启动 Agent {self.agent_id}")
        self.state = "running"
        self.start_time = time.time()
        self._init_components()
        self._run_loop()
    
    def pause(self):
        """暂停Agent"""
        if self.state != "running":
            logger.warning(f"Agent {self.agent_id} 不在运行状态")
            return
        
        logger.info(f"正在暂停 Agent {self.agent_id}")
        self.state = "paused"
    
    def resume(self):
        """恢复Agent"""
        if self.state != "paused":
            logger.warning(f"Agent {self.agent_id} 不在暂停状态")
            return
        
        logger.info(f"正在恢复 Agent {self.agent_id}")
        self.state = "running"
        self._run_loop()
    
    def stop(self):
        """停止Agent"""
        if self.state == "stopped":
            logger.warning(f"Agent {self.agent_id} 已经停止")
            return
        
        logger.info(f"正在停止 Agent {self.agent_id}")
        self.state = "stopped"
        self._cleanup()
    
    def _run_loop(self):
        """主运行循环"""
        sample_rate = self.config["perception"]["sample_rate"]
        sleep_time = 1.0 / sample_rate if sample_rate