我们在实际做项目的时候，单个模型是不够用的，即便是移动端也都是多个流程接一起：
- 先检测、用检测框扣图、拿扣图做个活体或者分类
- 算力实在有限，先调一下原始输入，再检测跟踪，最后抽特征
- 甚至考虑 3D 序列的，用队列攒 10 个再一起推理
需求是多样滴、实现是诡异滴。为了适应这些，MegFlow 用了图的表述结构：上面列出来的三种情况都可以往图里面套（哪怕语音识别也可以）。 又因为是搞计算的，所以叫它“计算图”。
这次从最简单的情况开始，把 resnet18 分类模型，部署成 HTTP 图片服务。
准备模型
MegEngine/models 有现成 imagenet 预训模型。这里把模型 dump 成 .mge，推理更快。
写个简单的 dump.py 按 [1, 3, 224, 224] 尺寸 trace forward 模型，打开推理优化选项，存为 model.mge。
def dump_static_graph(model, graph_name, shape):
model.eval()
data = mge.Tensor(np.random.random(shape))
@jit.trace(capture_as_const=True)
def pred_func(data):
# 下一期合并预处理教程将改这里，以后再也不用关心 preprocess 了
outputs = model(data)
return outputs
pred_func(data)
pred_func.dump(
graph_name,
arg_names=["data"],
optimize_for_inference=True,
enable_fuse_conv_bias_nonlinearity=True,
)PR 已经提交，等 merge 了就在 models repo 里了。用法很简单，支持 resnet 和 shufflenet：
$ python3 dump.py -a resnet18 -s 1 3 224 224
...
$ ll model.mge
...
$ python3 dump.py --help模型单测
软件开发中，单测可以减少整体的工期，仅仅会在大前期拖慢开发进度。
开发模型的推理封装，调用方传入一张或多张图片、直接获取结果，尽量避免关心内部实现。
def inference(self, mat):
img = self.preprocess(mat, input_size=(224,224), scale_im = False, mean=[103.530, 116.280, 123.675], std=[57.375, 57.120, 58.395])
# build input tensor
inp_data =self.net.get_io_tensor("data")
inp_data.set_data_by_share(img)
# forward
self.net.forward()
self.net.wait()
# postprocess
output_keys = self.net.get_all_output_name()
output = self.net.get_io_tensor(output_keys[0]).to_numpy()
return np.argmax(output[0])测试很简单：
$ python3 lite.py --model model.mge --path test.jpg
2021-09-14 11:45:02.406 | INFO | __main__:<module>:81 - 285\`285\` 是分类模型最后一层的 argmax，对应含义需要查 imagenet 分类表 ，这里是 “Egyptian cat”（下标从 0 开始）。 test.jpg 嘛就是这张
配置计算图
前面已经介绍了“计算图”的由来。这次只有 1 个结点，带一堆东西“显得”累赘，等后面的教程图复杂起来，就知道这种描述的好处了。
$ cat flow-python/examples/simple_classification/image_cpu.toml
main = "tutorial_01_image"
[[graphs]]
name = "subgraph"
inputs = [{ name = "inp", cap = 16, ports = ["classify:inp"] }] # 一、输入输出结点
outputs = [{ name = "out", cap = 16, ports = ["classify:out"] }]
connections = [
]
[[graphs.nodes]] # 二、结点参数
name = "classify"
ty = "Classify"
path = "resnet18.mge"
device = "cpu"
device_id = 0
...
[[graphs.nodes]] # 三、服务类型配置
name = "source"
ty = "ImageServer"
port = 8084 # 端口号 8084
response = "json"图片单模型服务只需要关心 3 处：
- 计算图输入、输出结点的名字。名字是自定义的，这里都是\`classify\`
- \`classify\` 结点的参数。
- 最重要的是 \`ty="Classify"\`指明了类名，MegFlow 将在当前目录搜索\`Classify\`类。
- 剩下的 path/device/device_id 分别是模型路径/用 CPU 推理/用哪个核，属于用户自定义配置
- 服务类型。这里想运行图片服务 \`ty = "ImageServer"\`，如果想运行视频解析服务改 \`ty = "VideoServer"\`
实现 classify node
\`classif.py\` 就是从 envelope.msg['data'] 取 BGR 输入，把之前单测调起来，结果送出去
@register(inputs=['inp'], outputs=['out'])
class Classify:
...
def exec(self):
envelope = self.inp.recv()
if envelope is None:
return
data = envelope.msg['data']
result = self._model.inference(data)
self.out.send(envelope.repack(json.dumps(str(result))))建议 \` init \` 加个 warmup，否则第一次 inference 比较慢影响体验。
运行测试
WebUI 方式
浏览器打开 8084 端口服务（例如 http://127.0.0.1:8084/docs ），熟悉的 Web UI，选择一张图“try it out”即可。
命令行方式
$ curl http://127.0.0.1:8081/analyze/image_name -X POST --header "Content-Type:image/*" --data-binary @test.jpegimage_name 是用户自定义参数，用在需要 POST 内容的场景。这里随便填即可；test.jpeg 是测试图片
Python Client
$ cat ${MegFlow_DIR}/flow-python/examples/simple_classification/client.py
import requests
import cv2
def test():
ip = 'localhost'
port = '8084'
url = 'http://{}:{}/analyze/any_content'.format(ip, port)
img = cv2.imread("./test.jpg")
_, data = cv2.imencode(".jpg", img)
data = data.tobytes()
headers = {'Content-Length': '%d' % len(data), 'Content-Type': 'image/*'}
res = requests.post(url, data=data, headers=headers)
print(res.content)
if __name__ == "__main__":
test()其他
一、代码
完整的实现已经放到了 MegFlow simple classification，运行方式和猫猫围栏很接近
$ cd MegFlow/flow-python/examples
$ run_with_plugins -c simple_classification/image_cpu.toml -p simple_classification注意：源码/docker 编译流程用的 cargo 是 Rust 的辅助工具。MegFlow 使用者，只要有
- 编译好的 run with_ plugins 可执行文件
- import megflow 正常
就可以直接使用，不用关心 cargo 和 rust
原文链接：知乎
作者： 白牛
推荐阅读
更多嵌入式AI技术相关内容请关注嵌入式AI专栏。
