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 101 102
| import json
class LayerAllocation: """ 分析模型各层在 NPU/CPU 上的分配 典型情况: - 标准卷积 → NPU(高效) - 特殊算子 → CPU 回退(慢) - 未知算子 → CPU 回退(极慢) """ HEXAGON_SUPPORTED = [ 'Conv2d', 'DepthwiseConv2d', 'Conv2dTranspose', 'BatchNorm2d', 'Relu', 'Relu6', 'Sigmoid', 'MaxPool2d', 'AvgPool2d', 'AdaptiveAvgPool2d', 'Concat', 'Add', 'Mul', 'Flatten', 'FullyConnected', 'Reshape', 'Sigmoid', 'Swish', 'HardSwish', 'ResizeNearest', 'ResizeBilinear', ] HEXAGON_UNSUPPORTED = [ 'TransformerAttention', 'LSTM', 'GRU', 'Einsum', 'ScatterAdd', 'CustomOp', ] def analyze_model(self, model_graph: list) -> dict: """ 分析模型层分配 Args: model_graph: 模型层列表 Returns: allocation: NPU/CPU 分配结果 """ npu_layers = 0 cpu_layers = 0 cpu_fallback = [] for layer in model_graph: op_type = layer['type'] if op_type in self.HEXAGON_SUPPORTED: npu_layers += 1 else: cpu_layers += 1 cpu_fallback.append({ 'name': layer['name'], 'type': op_type, 'params': layer.get('params', {}), 'estimated_ms': self._estimate_cpu_time(layer) }) total = npu_layers + cpu_layers npu_ratio = npu_layers / total if total > 0 else 0 npu_time = npu_layers * 0.1 cpu_time = sum(l['estimated_ms'] for l in cpu_fallback) return { 'npu_layers': npu_layers, 'cpu_layers': cpu_layers, 'npu_ratio': f"{npu_ratio:.1%}", 'estimated_npu_ms': npu_time, 'estimated_cpu_ms': cpu_time, 'estimated_total_ms': npu_time + cpu_time, 'cpu_fallbacks': cpu_fallback } def _estimate_cpu_time(self, layer): """估算 CPU 执行时间""" base = {'LSTM': 5.0, 'GRU': 4.0, 'TransformerAttention': 8.0} return base.get(layer['type'], 2.0)
if __name__ == "__main__": analyzer = LayerAllocation() model_graph = [ {'name': 'conv1', 'type': 'Conv2d'}, {'name': 'bn1', 'type': 'BatchNorm2d'}, {'name': 'relu1', 'type': 'Relu'}, {'name': 'conv2', 'type': 'Conv2d'}, {'name': 'lstm', 'type': 'LSTM'}, {'name': 'attn', 'type': 'TransformerAttention'}, {'name': 'fc', 'type': 'FullyConnected'}, ] result = analyzer.analyze_model(model_graph) print(f"NPU 层数: {result['npu_layers']}/{result['npu_layers']+result['cpu_layers']}") print(f"NPU 占比: {result['npu_ratio']}") print(f"估算总延迟: {result['estimated_total_ms']:.1f}ms") print(f"CPU 回退层: {[l['name'] for l in result['cpu_fallbacks']]}")
|