边缘 AI 与端侧 ML 完全指南：TFLite、ONNX、Core ML、llama.cpp

import tensorflow as tf

# 方法 1：从 SavedModel 转换
converter = tf.lite.TFLiteConverter.from_saved_model('saved_model_dir')
tflite_model = converter.convert()

with open('model.tflite', 'wb') as f:
    f.write(tflite_model)

# 方法 2：直接从 Keras 模型转换
model = tf.keras.applications.MobileNetV2(weights='imagenet')
converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()

with open('mobilenetv2.tflite', 'wb') as f:
    f.write(tflite_model)

# 方法 3：从 concrete function 转换
@tf.function(input_signature=[tf.TensorSpec(shape=[1, 224, 224, 3], dtype=tf.float32)])
def predict(x):
    return model(x)

converter = tf.lite.TFLiteConverter.from_concrete_functions(
    [predict.get_concrete_function()]
)
tflite_model = converter.convert()

converter = tf.lite.TFLiteConverter.from_saved_model('saved_model_dir')
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.target_spec.supported_types = [tf.float16]
tflite_model = converter.convert()
# 模型体积减少约 50%，精度几乎保持不变

import numpy as np

def representative_dataset():
    # 使用 100~1000 个真实数据样本
    for _ in range(100):
        data = np.random.rand(1, 224, 224, 3).astype(np.float32)
        yield [data]

converter = tf.lite.TFLiteConverter.from_saved_model('saved_model_dir')
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.representative_dataset = representative_dataset
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()
# 模型体积减少约 75%，推理速度提升 2-4 倍

converter = tf.lite.TFLiteConverter.from_saved_model('saved_model_dir')
converter.optimizations = [tf.lite.Optimize.DEFAULT]
# 不需要 representative_dataset 也可以 - 只量化权重
tflite_model = converter.convert()

import tensorflow as tf
import numpy as np
from PIL import Image

# 初始化 interpreter
interpreter = tf.lite.Interpreter(model_path='mobilenetv2.tflite')
interpreter.allocate_tensors()

# 查看输入输出张量信息
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()
print(f"输入 shape: {input_details[0]['shape']}")
print(f"输入 dtype: {input_details[0]['dtype']}")
print(f"输出 shape: {output_details[0]['shape']}")

# 图像预处理
img = Image.open('test_image.jpg').resize((224, 224))
input_data = np.expand_dims(np.array(img, dtype=np.float32) / 255.0, axis=0)

# 执行推理
interpreter.set_tensor(input_details[0]['index'], input_data)
interpreter.invoke()

# 提取结果
output_data = interpreter.get_tensor(output_details[0]['index'])
predicted_class = np.argmax(output_data[0])
confidence = output_data[0][predicted_class]
print(f"预测类别: {predicted_class}, 置信度: {confidence:.4f}")

# 多线程设置
interpreter = tf.lite.Interpreter(
    model_path='model.tflite',
    num_threads=4
)

# GPU 委托 (Android/iOS)
try:
    from tensorflow.lite.python.interpreter import load_delegate
    gpu_delegate = load_delegate('libdelegate.so')
    interpreter = tf.lite.Interpreter(
        model_path='model.tflite',
        experimental_delegates=[gpu_delegate]
    )
    print("GPU 委托已启用")
except Exception as e:
    print(f"GPU 委托失败，改用 CPU: {e}")
    interpreter = tf.lite.Interpreter(model_path='model.tflite')

interpreter.allocate_tensors()

dependencies {
    implementation 'org.tensorflow:tensorflow-lite:2.13.0'
    implementation 'org.tensorflow:tensorflow-lite-gpu:2.13.0'
    implementation 'org.tensorflow:tensorflow-lite-support:0.4.4'
}

import org.tensorflow.lite.Interpreter
import org.tensorflow.lite.support.image.TensorImage
import org.tensorflow.lite.support.tensorbuffer.TensorBuffer
import java.nio.ByteBuffer
import java.nio.ByteOrder

class TFLiteClassifier(private val context: Context) {

    private lateinit var interpreter: Interpreter
    private val inputSize = 224
    private val numClasses = 1000

    fun initialize() {
        val model = loadModelFile("mobilenetv2.tflite")
        val options = Interpreter.Options().apply {
            numThreads = 4
            useNNAPI = true  // Android Neural Networks API
        }
        interpreter = Interpreter(model, options)
    }

    private fun loadModelFile(filename: String): ByteBuffer {
        val assetFileDescriptor = context.assets.openFd(filename)
        val fileInputStream = FileInputStream(assetFileDescriptor.fileDescriptor)
        val fileChannel = fileInputStream.channel
        val startOffset = assetFileDescriptor.startOffset
        val declaredLength = assetFileDescriptor.declaredLength
        return fileChannel.map(FileChannel.MapMode.READ_ONLY, startOffset, declaredLength)
    }

    fun classify(bitmap: Bitmap): FloatArray {
        val resized = Bitmap.createScaledBitmap(bitmap, inputSize, inputSize, true)
        val inputBuffer = ByteBuffer.allocateDirect(1 * inputSize * inputSize * 3 * 4)
        inputBuffer.order(ByteOrder.nativeOrder())

        for (y in 0 until inputSize) {
            for (x in 0 until inputSize) {
                val pixel = resized.getPixel(x, y)
                inputBuffer.putFloat(((pixel shr 16 and 0xFF) / 255.0f))
                inputBuffer.putFloat(((pixel shr 8 and 0xFF) / 255.0f))
                inputBuffer.putFloat(((pixel and 0xFF) / 255.0f))
            }
        }

        val outputBuffer = Array(1) { FloatArray(numClasses) }
        interpreter.run(inputBuffer, outputBuffer)
        return outputBuffer[0]
    }
}

import TensorFlowLite
import UIKit

class TFLiteImageClassifier {
    private var interpreter: Interpreter
    private let inputWidth = 224
    private let inputHeight = 224

    init(modelName: String) throws {
        guard let modelPath = Bundle.main.path(forResource: modelName, ofType: "tflite") else {
            throw NSError(domain: "ModelNotFound", code: 0, userInfo: nil)
        }
        var options = Interpreter.Options()
        options.threadCount = 4
        interpreter = try Interpreter(modelPath: modelPath, options: options)
        try interpreter.allocateTensors()
    }

    func classify(image: UIImage) throws -> [Float] {
        guard let cgImage = image.cgImage else { return [] }

        let inputTensor = try interpreter.input(at: 0)
        let batchSize = 1
        let inputChannels = 3
        let inputData = preprocessImage(cgImage: cgImage)

        try interpreter.copy(inputData, toInputAt: 0)
        try interpreter.invoke()

        let outputTensor = try interpreter.output(at: 0)
        let results: [Float] = [Float](unsafeData: outputTensor.data) ?? []
        return results
    }

    private func preprocessImage(cgImage: CGImage) -> Data {
        var data = Data(count: inputWidth * inputHeight * 3 * 4)
        // 图像缩放及归一化处理
        return data
    }
}

import torch
import torch.nn as nn
import torchvision.models as models

# 加载模型
model = models.resnet50(pretrained=True)
model.eval()

# 生成 dummy 输入
dummy_input = torch.randn(1, 3, 224, 224)

# 导出 ONNX
torch.onnx.export(
    model,
    dummy_input,
    "resnet50.onnx",
    export_params=True,
    opset_version=17,
    do_constant_folding=True,
    input_names=['input'],
    output_names=['output'],
    dynamic_axes={
        'input': {0: 'batch_size'},
        'output': {0: 'batch_size'}
    }
)
print("ONNX 转换完成！")

# 校验 ONNX 模型
import onnx
onnx_model = onnx.load("resnet50.onnx")
onnx.checker.check_model(onnx_model)
print(f"ONNX IR 版本: {onnx_model.ir_version}")
print(f"Opset 版本: {onnx_model.opset_import[0].version}")

import onnxruntime as ort
import numpy as np
from PIL import Image
import torchvision.transforms as transforms

# 创建会话并设置 provider
providers = ['CUDAExecutionProvider', 'CPUExecutionProvider']
session = ort.InferenceSession("resnet50.onnx", providers=providers)

# 确认实际使用的 provider
print(f"使用的 provider: {session.get_providers()}")

# 查看输入输出信息
input_name = session.get_inputs()[0].name
output_name = session.get_outputs()[0].name
input_shape = session.get_inputs()[0].shape
print(f"输入名称: {input_name}, shape: {input_shape}")

# 图像预处理
transform = transforms.Compose([
    transforms.Resize(256),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])

img = Image.open('test.jpg')
input_tensor = transform(img).unsqueeze(0).numpy()

# 推理
outputs = session.run([output_name], {input_name: input_tensor})
logits = outputs[0]
predicted_class = np.argmax(logits[0])
print(f"预测类别: {predicted_class}")

from onnxruntime.transformers import optimizer
from onnxruntime.quantization import quantize_dynamic, QuantType

# 图优化（Transformer 模型）
optimized_model = optimizer.optimize_model(
    'bert_base.onnx',
    model_type='bert',
    num_heads=12,
    hidden_size=768,
    optimization_options=None
)
optimized_model.save_model_to_file('bert_optimized.onnx')

# 动态量化 (INT8)
quantize_dynamic(
    model_input='bert_optimized.onnx',
    model_output='bert_quantized_int8.onnx',
    weight_type=QuantType.QInt8,
    per_channel=True
)
print("量化完成！")

# 会话选项调优
so = ort.SessionOptions()
so.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
so.intra_op_num_threads = 4
so.inter_op_num_threads = 2
so.execution_mode = ort.ExecutionMode.ORT_SEQUENTIAL

session = ort.InferenceSession('model.onnx', sess_options=so)

// npm install onnxruntime-web
import * as ort from 'onnxruntime-web'

async function runInference() {
  // 设置 WebAssembly 后端
  ort.env.wasm.wasmPaths = '/static/'
  ort.env.wasm.numThreads = 4

  // 创建会话
  const session = await ort.InferenceSession.create('/models/mobilenet.onnx', {
    executionProviders: ['webgpu', 'wasm'],
    graphOptimizationLevel: 'all',
  })

  // 创建输入张量 (1, 3, 224, 224)
  const inputData = new Float32Array(1 * 3 * 224 * 224).fill(0.5)
  const inputTensor = new ort.Tensor('float32', inputData, [1, 3, 224, 224])

  // 执行推理
  const feeds = { input: inputTensor }
  const results = await session.run(feeds)

  // 处理结果
  const outputData = results.output.data
  const maxIndex = Array.from(outputData).indexOf(Math.max(...outputData))
  console.log('预测类别:', maxIndex)
}

runInference()

import coremltools as ct
import torch
import torchvision.models as models

# 转换 PyTorch 模型
torch_model = models.mobilenet_v2(pretrained=True)
torch_model.eval()

example_input = torch.rand(1, 3, 224, 224)
traced_model = torch.jit.trace(torch_model, example_input)

# 转换为 Core ML
mlmodel = ct.convert(
    traced_model,
    inputs=[ct.TensorType(name='input', shape=(1, 3, 224, 224))],
    compute_units=ct.ComputeUnit.ALL,  # 同时使用 CPU + GPU + Neural Engine
    minimum_deployment_target=ct.target.iOS16
)

# 添加元数据
mlmodel.short_description = "MobileNetV2 图像分类器"
mlmodel.author = "YJ Blog"
mlmodel.version = "1.0"

mlmodel.save("MobileNetV2.mlpackage")
print("Core ML 转换完成！")

import coremltools as ct
from coremltools.optimize.coreml import (
    OpLinearQuantizerConfig,
    OptimizationConfig,
    linearly_quantize_weights
)

# 加载原始模型
mlmodel = ct.models.MLModel("MobileNetV2.mlpackage")

# 线性权重量化（8 位）
op_config = OpLinearQuantizerConfig(mode="linear_symmetric", dtype="int8")
config = OptimizationConfig(global_config=op_config)

compressed_model = linearly_quantize_weights(mlmodel, config)
compressed_model.save("MobileNetV2_int8.mlpackage")

# 调色板量化（4 位）
from coremltools.optimize.coreml import palettize_weights, OpPalettizerConfig

palette_config = OptimizationConfig(
    global_config=OpPalettizerConfig(mode="kmeans", nbits=4)
)
palette_model = palettize_weights(mlmodel, palette_config)
palette_model.save("MobileNetV2_4bit.mlpackage")

import CoreML
import Vision
import UIKit

class CoreMLClassifier {
    private var model: VNCoreMLModel?

    func loadModel() {
        guard let modelURL = Bundle.main.url(forResource: "MobileNetV2", withExtension: "mlpackage") else {
            print("找不到模型文件")
            return
        }

        let config = MLModelConfiguration()
        config.computeUnits = .all  // CPU + GPU + Neural Engine

        do {
            let coreMLModel = try MLModel(contentsOf: modelURL, configuration: config)
            model = try VNCoreMLModel(for: coreMLModel)
            print("模型加载成功")
        } catch {
            print("模型加载失败: \(error)")
        }
    }

    func classify(image: UIImage, completion: @escaping ([VNClassificationObservation]?) -> Void) {
        guard let model = model,
              let cgImage = image.cgImage else {
            completion(nil)
            return
        }

        let request = VNCoreMLRequest(model: model) { request, error in
            guard let results = request.results as? [VNClassificationObservation] else {
                completion(nil)
                return
            }
            let topResults = results.prefix(5)
            completion(Array(topResults))
        }

        request.imageCropAndScaleOption = .centerCrop

        let handler = VNImageRequestHandler(cgImage: cgImage, options: [:])
        DispatchQueue.global(qos: .userInteractive).async {
            do {
                try handler.perform([request])
            } catch {
                print("推理失败: \(error)")
                completion(nil)
            }
        }
    }
}

// 使用示例
let classifier = CoreMLClassifier()
classifier.loadModel()
classifier.classify(image: UIImage(named: "test")!) { results in
    results?.forEach { result in
        print("\(result.identifier): \(String(format: "%.2f", result.confidence * 100))%")
    }
}

import CreateML
import Foundation

// 训练图像分类模型
let trainingData = MLImageClassifier.DataSource.labeledDirectories(
    at: URL(fileURLWithPath: "/path/to/training_data")
)

let parameters = MLImageClassifier.ModelParameters(
    featureExtractor: .scenePrint(revision: 2),
    maxIterations: 25,
    augmentation: [.flip, .crop, .rotation]
)

let classifier = try MLImageClassifier(
    trainingData: trainingData,
    parameters: parameters
)

// 评估
let evaluationData = MLImageClassifier.DataSource.labeledDirectories(
    at: URL(fileURLWithPath: "/path/to/test_data")
)
let metrics = classifier.evaluation(on: evaluationData)
print("准确率: \(metrics.classificationError)")

// 保存
try classifier.write(to: URL(fileURLWithPath: "MyClassifier.mlmodel"))

import tensorrt as trt
import numpy as np
import pycuda.driver as cuda
import pycuda.autoinit

# 创建 TensorRT logger
TRT_LOGGER = trt.Logger(trt.Logger.WARNING)

def build_engine_from_onnx(onnx_path, engine_path, fp16=True, int8=False):
    with trt.Builder(TRT_LOGGER) as builder, \
         builder.create_network(1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)) as network, \
         trt.OnnxParser(network, TRT_LOGGER) as parser:

        config = builder.create_builder_config()
        config.max_workspace_size = 1 << 30  # 1GB

        if fp16:
            config.set_flag(trt.BuilderFlag.FP16)
        if int8:
            config.set_flag(trt.BuilderFlag.INT8)

        # 解析 ONNX
        with open(onnx_path, 'rb') as f:
            if not parser.parse(f.read()):
                for error in range(parser.num_errors):
                    print(parser.get_error(error))
                return None

        # 构建引擎
        print("正在构建 TensorRT 引擎...（需要几分钟）")
        serialized_engine = builder.build_serialized_network(network, config)

        # 保存
        with open(engine_path, 'wb') as f:
            f.write(serialized_engine)
        print(f"引擎保存完成: {engine_path}")

build_engine_from_onnx('resnet50.onnx', 'resnet50_fp16.trt', fp16=True)

import tensorrt as trt
import pycuda.driver as cuda
import pycuda.autoinit
import numpy as np

class TRTInference:
    def __init__(self, engine_path):
        TRT_LOGGER = trt.Logger(trt.Logger.WARNING)

        with open(engine_path, 'rb') as f:
            runtime = trt.Runtime(TRT_LOGGER)
            self.engine = runtime.deserialize_cuda_engine(f.read())

        self.context = self.engine.create_execution_context()
        self.inputs, self.outputs, self.bindings, self.stream = self._allocate_buffers()

    def _allocate_buffers(self):
        inputs, outputs, bindings = [], [], []
        stream = cuda.Stream()

        for binding in self.engine:
            size = trt.volume(self.engine.get_binding_shape(binding))
            dtype = trt.nptype(self.engine.get_binding_dtype(binding))
            host_mem = cuda.pagelocked_empty(size, dtype)
            device_mem = cuda.mem_alloc(host_mem.nbytes)
            bindings.append(int(device_mem))

            if self.engine.binding_is_input(binding):
                inputs.append({'host': host_mem, 'device': device_mem})
            else:
                outputs.append({'host': host_mem, 'device': device_mem})

        return inputs, outputs, bindings, stream

    def infer(self, input_data):
        np.copyto(self.inputs[0]['host'], input_data.ravel())
        cuda.memcpy_htod_async(self.inputs[0]['device'], self.inputs[0]['host'], self.stream)
        self.context.execute_async_v2(bindings=self.bindings, stream_handle=self.stream.handle)
        cuda.memcpy_dtoh_async(self.outputs[0]['host'], self.outputs[0]['device'], self.stream)
        self.stream.synchronize()
        return self.outputs[0]['host']

# 使用
trt_model = TRTInference('resnet50_fp16.trt')
input_array = np.random.rand(1, 3, 224, 224).astype(np.float32)
result = trt_model.infer(input_array)
print(f"预测类别: {np.argmax(result)}")

# 使用 DeepStream Python 绑定构建视频流水线
import gi
gi.require_version('Gst', '1.0')
from gi.repository import GObject, Gst, GLib

Gst.init(None)

def create_pipeline():
    pipeline = Gst.Pipeline()

    # 源: USB 摄像头
    source = Gst.ElementFactory.make("v4l2src", "usb-cam-source")
    source.set_property("device", "/dev/video0")

    # capsfilter
    caps = Gst.ElementFactory.make("capsfilter", "capsfilter")
    caps.set_property("caps", Gst.Caps.from_string("video/x-raw,width=1280,height=720,framerate=30/1"))

    # NVARGUS（摄像头 ISP）
    nvconv = Gst.ElementFactory.make("nvvideoconvert", "convertor")

    # nvinfer (TensorRT 推理)
    nvinfer = Gst.ElementFactory.make("nvinfer", "primary-inference")
    nvinfer.set_property("config-file-path", "config_infer_primary.txt")

    # 跟踪器
    tracker = Gst.ElementFactory.make("nvtracker", "tracker")
    tracker.set_property("ll-lib-file", "/opt/nvidia/deepstream/deepstream/lib/libnvds_nvmultiobjecttracker.so")

    # OSD (屏幕显示叠加)
    osd = Gst.ElementFactory.make("nvdsosd", "onscreendisplay")

    # 输出
    sink = Gst.ElementFactory.make("nveglglessink", "nvvideo-renderer")

    for element in [source, caps, nvconv, nvinfer, tracker, osd, sink]:
        pipeline.add(element)

    source.link(caps)
    caps.link(nvconv)
    nvconv.link(nvinfer)
    nvinfer.link(tracker)
    tracker.link(osd)
    osd.link(sink)

    return pipeline

pipeline = create_pipeline()
pipeline.set_state(Gst.State.PLAYING)

# 安装 Hailo SDK
pip install hailort

# 转换模型 (ONNX -> HEF)
# Hailo Model Zoo 提供预先转换好的模型
wget https://hailo-model-zoo.s3.eu-west-2.amazonaws.com/ModelZoo/Compiled/v2.11.0/hailo8/resnet_v1_50.hef

import hailo_platform as hp
import numpy as np

# 加载 HEF 并推理
with hp.VDevice() as vdevice:
    hef = hp.Hef("resnet_v1_50.hef")
    network_groups = vdevice.configure(hef)
    network_group = network_groups[0]

    input_vstreams_params = hp.InputVStreamParams.make_from_network_group(
        network_group, quantized=False, format_type=hp.FormatType.FLOAT32
    )
    output_vstreams_params = hp.OutputVStreamParams.make_from_network_group(
        network_group, quantized=False, format_type=hp.FormatType.FLOAT32
    )

    with hp.InferVStreams(network_group, input_vstreams_params, output_vstreams_params) as infer_pipeline:
        input_data = {"input_layer1": np.random.rand(1, 224, 224, 3).astype(np.float32)}
        with network_group.activate():
            infer_results = infer_pipeline.infer(input_data)
        print(f"结果: {np.argmax(infer_results['resnet_v1_50/softmax1'])}")

import cv2
import numpy as np
import tflite_runtime.interpreter as tflite

# TFLite Interpreter（在树莓派上使用轻量版）
interpreter = tflite.Interpreter(
    model_path='ssd_mobilenet_v2.tflite',
    num_threads=4
)
interpreter.allocate_tensors()
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()

# 初始化摄像头（树莓派摄像头 v2）
cap = cv2.VideoCapture(0)
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)

while True:
    ret, frame = cap.read()
    if not ret:
        break

    # 预处理
    input_size = (input_details[0]['shape'][2], input_details[0]['shape'][1])
    rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    resized = cv2.resize(rgb_frame, input_size)
    input_data = np.expand_dims(resized, axis=0).astype(np.uint8)

    # 推理
    interpreter.set_tensor(input_details[0]['index'], input_data)
    interpreter.invoke()

    # 提取结果 (SSD MobileNet)
    boxes = interpreter.get_tensor(output_details[0]['index'])[0]
    classes = interpreter.get_tensor(output_details[1]['index'])[0]
    scores = interpreter.get_tensor(output_details[2]['index'])[0]

    # 结果可视化
    h, w = frame.shape[:2]
    for i in range(len(scores)):
        if scores[i] > 0.5:
            ymin, xmin, ymax, xmax = boxes[i]
            cv2.rectangle(frame,
                         (int(xmin * w), int(ymin * h)),
                         (int(xmax * w), int(ymax * h)),
                         (0, 255, 0), 2)
            label = f"类别 {int(classes[i])}: {scores[i]:.2f}"
            cv2.putText(frame, label, (int(xmin * w), int(ymin * h) - 10),
                       cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)

    cv2.imshow('树莓派 AI', frame)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()

import mediapipe as mp
import cv2
import numpy as np

mp_hands = mp.solutions.hands
mp_drawing = mp.solutions.drawing_utils
mp_drawing_styles = mp.solutions.drawing_styles

def run_hand_tracking():
    cap = cv2.VideoCapture(0)

    with mp_hands.Hands(
        static_image_mode=False,
        max_num_hands=2,
        min_detection_confidence=0.7,
        min_tracking_confidence=0.5
    ) as hands:
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break

            # BGR -> RGB 转换
            rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            rgb_frame.flags.writeable = False
            results = hands.process(rgb_frame)
            rgb_frame.flags.writeable = True

            frame = cv2.cvtColor(rgb_frame, cv2.COLOR_RGB2BGR)

            if results.multi_hand_landmarks:
                for hand_landmarks in results.multi_hand_landmarks:
                    mp_drawing.draw_landmarks(
                        frame,
                        hand_landmarks,
                        mp_hands.HAND_CONNECTIONS,
                        mp_drawing_styles.get_default_hand_landmarks_style(),
                        mp_drawing_styles.get_default_hand_connections_style()
                    )

                    # 提取关键点坐标（21 个关键点）
                    for idx, landmark in enumerate(hand_landmarks.landmark):
                        h, w, _ = frame.shape
                        cx, cy = int(landmark.x * w), int(landmark.y * h)
                        if idx == 8:  # 食指指尖
                            cv2.circle(frame, (cx, cy), 10, (255, 0, 0), -1)

            cv2.imshow('手部追踪', frame)
            if cv2.waitKey(5) & 0xFF == ord('q'):
                break

    cap.release()
    cv2.destroyAllWindows()

run_hand_tracking()

import mediapipe as mp
import cv2

mp_pose = mp.solutions.pose

def calculate_angle(a, b, c):
    """根据三个点计算角度"""
    import numpy as np
    a = np.array(a)
    b = np.array(b)
    c = np.array(c)

    radians = np.arctan2(c[1] - b[1], c[0] - b[0]) - \
              np.arctan2(a[1] - b[1], a[0] - b[0])
    angle = np.abs(radians * 180.0 / np.pi)

    if angle > 180.0:
        angle = 360 - angle
    return angle

cap = cv2.VideoCapture(0)

with mp_pose.Pose(
    min_detection_confidence=0.5,
    min_tracking_confidence=0.5,
    model_complexity=1  # 0: Lite, 1: Full, 2: Heavy
) as pose:
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break

        rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        results = pose.process(rgb_frame)
        frame = cv2.cvtColor(rgb_frame, cv2.COLOR_RGB2BGR)

        if results.pose_landmarks:
            landmarks = results.pose_landmarks.landmark
            h, w, _ = frame.shape

            # 计算肘部角度
            shoulder = [landmarks[mp_pose.PoseLandmark.LEFT_SHOULDER.value].x * w,
                       landmarks[mp_pose.PoseLandmark.LEFT_SHOULDER.value].y * h]
            elbow = [landmarks[mp_pose.PoseLandmark.LEFT_ELBOW.value].x * w,
                    landmarks[mp_pose.PoseLandmark.LEFT_ELBOW.value].y * h]
            wrist = [landmarks[mp_pose.PoseLandmark.LEFT_WRIST.value].x * w,
                    landmarks[mp_pose.PoseLandmark.LEFT_WRIST.value].y * h]

            angle = calculate_angle(shoulder, elbow, wrist)
            cv2.putText(frame, f"肘部角度: {angle:.1f}度",
                       (int(elbow[0]), int(elbow[1])),
                       cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2)

            mp.solutions.drawing_utils.draw_landmarks(
                frame, results.pose_landmarks, mp_pose.POSE_CONNECTIONS
            )

        cv2.imshow('姿态估计', frame)
        if cv2.waitKey(5) & 0xFF == ord('q'):
            break

cap.release()
cv2.destroyAllWindows()

import mediapipe as mp
from mediapipe.tasks import python
from mediapipe.tasks.python import vision

# 目标检测任务
base_options = python.BaseOptions(model_asset_path='efficientdet_lite0.tflite')
options = vision.ObjectDetectorOptions(
    base_options=base_options,
    running_mode=vision.RunningMode.IMAGE,
    max_results=5,
    score_threshold=0.5
)

with vision.ObjectDetector.create_from_options(options) as detector:
    image = mp.Image.create_from_file('test_image.jpg')
    detection_result = detector.detect(image)

    for detection in detection_result.detections:
        category = detection.categories[0]
        print(f"目标: {category.category_name}, 置信度: {category.score:.2f}")
        bbox = detection.bounding_box
        print(f"  位置: ({bbox.origin_x}, {bbox.origin_y}), 尺寸: {bbox.width}x{bbox.height}")

# 源码编译
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp

# 仅 CPU
make -j4

# Apple Silicon（Metal GPU 加速）
make LLAMA_METAL=1 -j4

# NVIDIA CUDA
make LLAMA_CUDA=1 -j4

# 下载 GGUF 模型 (例如 Llama 3.2 3B)
huggingface-cli download \
    bartowski/Llama-3.2-3B-Instruct-GGUF \
    Llama-3.2-3B-Instruct-Q4_K_M.gguf \
    --local-dir ./models

# 交互式对话
./llama-cli \
    -m models/Llama-3.2-3B-Instruct-Q4_K_M.gguf \
    -n 512 \
    -p "你是一个友善的 AI 助手。" \
    --repeat-penalty 1.1 \
    -t 8 \
    --color

from llama_cpp import Llama

# 加载模型
llm = Llama(
    model_path="./models/Llama-3.2-3B-Instruct-Q4_K_M.gguf",
    n_ctx=4096,          # 上下文窗口
    n_threads=8,         # CPU 线程数
    n_gpu_layers=35,     # 放到 GPU 上的层数（-1 表示全部）
    verbose=False
)

# 基本文本生成
output = llm(
    "韩国的首都是哪里？",
    max_tokens=128,
    temperature=0.7,
    top_p=0.95,
    top_k=40,
    repeat_penalty=1.1
)
print(output['choices'][0]['text'])

# 对话格式
messages = [
    {"role": "system", "content": "你是一个友善的 AI 助手。请用中文回答。"},
    {"role": "user", "content": "请写一段用 Python 求斐波那契数列的代码。"}
]

response = llm.create_chat_completion(
    messages=messages,
    max_tokens=512,
    temperature=0.7
)
print(response['choices'][0]['message']['content'])

# 流式输出
stream = llm.create_chat_completion(
    messages=messages,
    max_tokens=512,
    stream=True
)

for chunk in stream:
    delta = chunk['choices'][0]['delta']
    if 'content' in delta:
        print(delta['content'], end='', flush=True)
print()

# 运行 llama.cpp 服务器
./llama-server \
    -m models/Llama-3.2-3B-Instruct-Q4_K_M.gguf \
    --port 8080 \
    --host 0.0.0.0 \
    -n 2048 \
    -t 8 \
    --n-gpu-layers 35

# 用 OpenAI SDK 调用本地服务器
from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:8080/v1",
    api_key="none"
)

response = client.chat.completions.create(
    model="local-model",
    messages=[
        {"role": "user", "content": "请说明机器学习与深度学习的区别。"}
    ],
    max_tokens=512,
    temperature=0.7
)
print(response.choices[0].message.content)

# HuggingFace 模型 → GGUF
cd llama.cpp

# 安装 Python 依赖
pip install -r requirements.txt

# 转换
python convert_hf_to_gguf.py \
    /path/to/hf_model \
    --outfile models/my_model.gguf \
    --outtype f16

# 量化
./quantize models/my_model.gguf models/my_model_q4km.gguf Q4_K_M

# 编译
git clone https://github.com/ggerganov/whisper.cpp
cd whisper.cpp
make -j4

# Apple Silicon
make WHISPER_METAL=1 -j4

# 下载模型
bash ./models/download-ggml-model.sh base.en    # 仅英语，142MB
bash ./models/download-ggml-model.sh medium     # 多语言，1.5GB
bash ./models/download-ggml-model.sh large-v3   # 最高质量，3.1GB

# 转录音频文件
./main -m models/ggml-medium.bin \
       -f audio.wav \
       -l zh \
       --output-txt \
       -of output

# 实时麦克风输入
./stream -m models/ggml-medium.bin \
         -t 8 \
         --step 500 \
         --length 5000 \
         -l zh

import whisper_cpp
import numpy as np
import soundfile as sf

# 加载模型
model = whisper_cpp.Whisper.from_pretrained("medium")

# 转录 WAV 文件
audio, sr = sf.read("audio.wav", dtype="float32")
if audio.ndim > 1:
    audio = audio.mean(axis=1)  # 立体声 -> 单声道

# 转换采样率（需要 16kHz）
if sr != 16000:
    import librosa
    audio = librosa.resample(audio, orig_sr=sr, target_sr=16000)

result = model.transcribe(audio, language="zh")
print(f"转录结果:\n{result['text']}")

# 包含时间戳
for segment in result['segments']:
    start = segment['start']
    end = segment['end']
    text = segment['text']
    print(f"[{start:.2f}s -> {end:.2f}s] {text}")

# Whisper GGML 模型量化
./quantize models/ggml-medium.bin models/ggml-medium-q5_0.bin q5_0

# 体积对比
ls -lh models/ggml-medium*.bin
# ggml-medium.bin: 1.5GB
# ggml-medium-q5_0.bin: 约 900MB

// 在 iOS 上使用 whisper.cpp（WhisperKit 库）
import WhisperKit

class SpeechRecognizer {
    var whisperKit: WhisperKit?

    func initialize() async {
        do {
            whisperKit = try await WhisperKit(
                model: "openai_whisper-medium",
                computeOptions: ModelComputeOptions(melCompute: .cpuAndGPU)
            )
            print("Whisper 模型加载完成")
        } catch {
            print("初始化失败: \(error)")
        }
    }

    func transcribe(audioURL: URL) async -> String? {
        guard let whisperKit = whisperKit else { return nil }

        do {
            let result = try await whisperKit.transcribe(
                audioPath: audioURL.path,
                decodeOptions: DecodingOptions(language: "zh")
            )
            return result.map(\.text).joined(separator: " ")
        } catch {
            print("转录失败: \(error)")
            return nil
        }
    }
}

<!DOCTYPE html>
<html>
  <head>
    <title>浏览器图像分类</title>
    <script src="https://cdn.jsdelivr.net/npm/@tensorflow/tfjs@latest"></script>
    <script src="https://cdn.jsdelivr.net/npm/@tensorflow-models/mobilenet@2.1.1"></script>
  </head>
  <body>
    <input type="file" id="imageInput" accept="image/*" />
    <img id="preview" style="max-width: 400px;" />
    <div id="result"></div>

    <script>
      let model

      async function loadModel() {
        model = await mobilenet.load({ version: 2, alpha: 1.0 })
        console.log('模型加载完成！')
        document.getElementById('result').textContent = '模型已就绪，请选择图片。'
      }

      document.getElementById('imageInput').addEventListener('change', async (e) => {
        const file = e.target.files[0]
        if (!file) return

        const img = document.getElementById('preview')
        img.src = URL.createObjectURL(file)
        img.onload = async () => {
          const predictions = await model.classify(img, 5)
          const resultDiv = document.getElementById('result')
          resultDiv.innerHTML = '<h3>预测结果:</h3>'
          predictions.forEach((pred) => {
            resultDiv.innerHTML += `
                    <p>${pred.className}: ${(pred.probability * 100).toFixed(2)}%</p>
                `
          })
        }
      })

      loadModel()
    </script>
  </body>
</html>

import { pipeline, env } from '@xenova/transformers'

// 设置 WASM 路径
env.backends.onnx.wasm.wasmPaths = 'https://cdn.jsdelivr.net/npm/onnxruntime-web/dist/'

// 文本分类流水线
async function runTextClassification() {
  const classifier = await pipeline(
    'sentiment-analysis',
    'Xenova/distilbert-base-uncased-finetuned-sst-2-english'
  )

  const results = await classifier(['I love machine learning!', 'This is terrible.'])
  results.forEach((result, i) => {
    console.log(`文本 ${i + 1}: ${result.label} (${(result.score * 100).toFixed(2)}%)`)
  })
}

// 图像分类
async function runImageClassification() {
  const classifier = await pipeline('image-classification', 'Xenova/vit-base-patch16-224')
  const result = await classifier('https://example.com/image.jpg')
  console.log('图像分类结果:', result)
}

// 文本生成（小型 LLM）
async function runTextGeneration() {
  const generator = await pipeline('text-generation', 'Xenova/gpt2')
  const output = await generator('人工智能的未来是', {
    max_new_tokens: 100,
    temperature: 0.7,
  })
  console.log('生成的文本:', output[0].generated_text)
}

runTextClassification()

import * as ort from 'onnxruntime-web'

async function runWithWebGPU() {
  // 检查 WebGPU 支持情况
  if (!navigator.gpu) {
    console.log('当前浏览器不支持 WebGPU。')
    return
  }

  const adapter = await navigator.gpu.requestAdapter()
  const device = await adapter.requestDevice()
  console.log('WebGPU adapter:', adapter.info)

  // 在 ONNX Runtime 中使用 WebGPU
  ort.env.wasm.wasmPaths = '/'
  const session = await ort.InferenceSession.create('/models/resnet50.onnx', {
    executionProviders: ['webgpu'],
    graphOptimizationLevel: 'all',
  })

  // 批量推理
  const batchSize = 4
  const inputData = new Float32Array(batchSize * 3 * 224 * 224)
  const inputTensor = new ort.Tensor('float32', inputData, [batchSize, 3, 224, 224])

  const startTime = performance.now()
  const output = await session.run({ input: inputTensor })
  const elapsed = performance.now() - startTime

  console.log(`WebGPU 推理时间: ${elapsed.toFixed(2)}ms`)
  console.log(`批处理吞吐量: ${((batchSize / elapsed) * 1000).toFixed(1)} images/sec`)
}

runWithWebGPU()

训练 (PyTorch/TF)
     ↓
剪枝（去除不必要的权重）
     ↓
知识蒸馏（Teacher-Student）
     ↓
量化感知训练 (QAT)
     ↓
格式转换 (ONNX/TFLite/GGUF)
     ↓
运行时优化 (TensorRT/OpenVINO)
     ↓
部署（移动端/边缘/网页）

import torch
import torch.nn.utils.prune as prune
import torchvision.models as models

model = models.resnet50(pretrained=True)

# 结构化剪枝：移除 Conv2d 层 50% 的过滤器
for name, module in model.named_modules():
    if isinstance(module, torch.nn.Conv2d):
        prune.l1_unstructured(module, name='weight', amount=0.3)
        prune.remove(module, 'weight')  # 永久应用掩码

# 确认模型体积
original_params = sum(p.numel() for p in models.resnet50().parameters())
pruned_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
print(f"原始参数: {original_params:,}")
print(f"剪枝后: {pruned_params:,}")
print(f"减少比例: {(1 - pruned_params/original_params)*100:.1f}%")

import torch
import torch.nn as nn
import torch.nn.functional as F

class DistillationLoss(nn.Module):
    def __init__(self, temperature=4.0, alpha=0.7):
        super().__init__()
        self.T = temperature
        self.alpha = alpha

    def forward(self, student_logits, teacher_logits, labels):
        # 软目标损失（蒸馏损失）
        soft_loss = F.kl_div(
            F.log_softmax(student_logits / self.T, dim=1),
            F.softmax(teacher_logits / self.T, dim=1),
            reduction='batchmean'
        ) * (self.T ** 2)

        # 硬目标损失（交叉熵）
        hard_loss = F.cross_entropy(student_logits, labels)

        return self.alpha * soft_loss + (1 - self.alpha) * hard_loss

def train_student(teacher, student, dataloader, epochs=10):
    teacher.eval()
    student.train()

    optimizer = torch.optim.AdamW(student.parameters(), lr=1e-4)
    criterion = DistillationLoss(temperature=4.0, alpha=0.7)

    for epoch in range(epochs):
        total_loss = 0
        for images, labels in dataloader:
            with torch.no_grad():
                teacher_logits = teacher(images)

            student_logits = student(images)
            loss = criterion(student_logits, teacher_logits, labels)

            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            total_loss += loss.item()

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

# 教师: ResNet50，学生: MobileNetV2
teacher = models.resnet50(pretrained=True)
student = models.mobilenet_v2(pretrained=False)

import torch
from torch.quantization import get_default_qat_qconfig, prepare_qat, convert

model = models.mobilenet_v2(pretrained=True)
model.train()

# QAT 设置
model.qconfig = get_default_qat_qconfig('qnnpack')  # ARM/移动端
# model.qconfig = get_default_qat_qconfig('fbgemm')  # x86

# 准备 QAT（插入伪量化）
model = prepare_qat(model, inplace=False)

# 微调训练（少量 epoch）
optimizer = torch.optim.SGD(model.parameters(), lr=0.0001)
model.train()
for epoch in range(5):
    for images, labels in dataloader:
        outputs = model(images)
        loss = nn.CrossEntropyLoss()(outputs, labels)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    print(f"QAT Epoch {epoch+1}/5 完成")

# 转换为 INT8 模型
model.eval()
quantized_model = convert(model.eval(), inplace=False)
torch.save(quantized_model.state_dict(), 'mobilenetv2_int8.pth')
print("QAT 完成！精度几乎没有下降，模型体积缩小了 4 倍")

import time
import numpy as np
import psutil
import os

class EdgeAIBenchmark:
    def __init__(self, model_path, framework='tflite'):
        self.model_path = model_path
        self.framework = framework
        self.results = {}

    def measure_latency(self, input_data, num_runs=100, warmup=10):
        """测量平均推理延迟"""
        import tensorflow as tf

        interpreter = tf.lite.Interpreter(model_path=self.model_path)
        interpreter.allocate_tensors()
        input_details = interpreter.get_input_details()
        output_details = interpreter.get_output_details()

        # 预热
        for _ in range(warmup):
            interpreter.set_tensor(input_details[0]['index'], input_data)
            interpreter.invoke()

        # 测量
        latencies = []
        for _ in range(num_runs):
            start = time.perf_counter()
            interpreter.set_tensor(input_details[0]['index'], input_data)
            interpreter.invoke()
            _ = interpreter.get_tensor(output_details[0]['index'])
            latencies.append((time.perf_counter() - start) * 1000)

        self.results['latency_mean_ms'] = np.mean(latencies)
        self.results['latency_p99_ms'] = np.percentile(latencies, 99)
        self.results['throughput_fps'] = 1000 / np.mean(latencies)
        return self.results

    def measure_memory(self):
        """测量内存使用量"""
        process = psutil.Process(os.getpid())
        before = process.memory_info().rss / 1024 / 1024

        import tensorflow as tf
        interpreter = tf.lite.Interpreter(model_path=self.model_path)
        interpreter.allocate_tensors()

        after = process.memory_info().rss / 1024 / 1024
        self.results['memory_mb'] = after - before
        return self.results

    def measure_model_size(self):
        """模型文件大小"""
        size_bytes = os.path.getsize(self.model_path)
        self.results['model_size_mb'] = size_bytes / 1024 / 1024
        return self.results

    def run_full_benchmark(self, input_data):
        self.measure_model_size()
        self.measure_memory()
        self.measure_latency(input_data)

        print(f"\n=== {self.model_path} 基准测试 ===")
        print(f"模型大小: {self.results.get('model_size_mb', 0):.2f} MB")
        print(f"内存使用: {self.results.get('memory_mb', 0):.2f} MB")
        print(f"平均延迟: {self.results.get('latency_mean_ms', 0):.2f} ms")
        print(f"P99 延迟: {self.results.get('latency_p99_ms', 0):.2f} ms")
        print(f"吞吐量: {self.results.get('throughput_fps', 0):.1f} FPS")
        return self.results


# 使用
import numpy as np
input_data = np.random.rand(1, 224, 224, 3).astype(np.float32)

bench = EdgeAIBenchmark('mobilenetv2.tflite')
results = bench.run_full_benchmark(input_data)

bench_q = EdgeAIBenchmark('mobilenetv2_int8.tflite')
results_q = bench_q.run_full_benchmark(input_data)

print("\n=== 量化效果 ===")
size_reduction = (1 - results_q['model_size_mb'] / results['model_size_mb']) * 100
speed_improvement = results['latency_mean_ms'] / results_q['latency_mean_ms']
print(f"体积减少: {size_reduction:.1f}%")
print(f"速度提升: {speed_improvement:.1f}倍")