Flask+TensorFlow 1.x多线程批量图片推理服务实战

1. 项目背景与整体架构拆解

1.1 这个项目到底在解决什么问题

做过深度学习模型落地的朋友大概率都遇到过这样一个场景：模型在本地训练好了，精度也调得差不多了，接下来要把它变成一个能对外服务的接口，让前端或者客户端能把图片传上来，后端跑完推理再把结果返回去。听起来很简单对吧？但真正动手的时候，问题就来了——单张图片推理太慢，用户上传一批图片要等半天；Flask默认的开发服务器是单线程的，请求一多就排队；TensorFlow 1.x 的Session在多线程环境下还有各种坑。

这个项目要解决的核心问题就是：用Flask搭建一个图片批量上传的Web后端，加载已经训练好的TensorFlow 1.x/Keras模型，通过多线程或多进程的方式并发执行推理，最终把预测结果同步返回给客户端。关键词拆开来看就是几个部分：Flask后端、批量图片上传、TensorFlow 1.x模型加载、Keras模型加载、多线程推理、多进程推理、同步返回结果。

适合谁来参考呢？如果你手头有一个用TensorFlow 1.x或者Keras训练好的模型，想快速搭一个能扛住一定并发量的推理服务，又不想引入TensorFlow Serving、TorchServe这类偏重的方案，那这套思路就非常合适。它不需要你懂太多DevOps的东西，Python层面就能搞定，部署也简单。

1.2 为什么选Flask而不是其他框架

Web框架的选择其实挺多的，Django、FastAPI、Flask、Tornado都能做这件事。选Flask的理由很直接：轻量、灵活、和Python生态无缝衔接。Django太重了，自带ORM、Admin、模板引擎一大堆东西，我们只是要一个接收图片返回JSON的接口，用Django属于杀鸡用牛刀。FastAPI确实性能好、异步支持强，但TensorFlow 1.x的推理本身是同步阻塞的，异步框架的优势发挥不出来，反而增加了学习成本。Tornado的异步模型也是类似的问题。

Flask的好处在于它的同步模型和TensorFlow 1.x的Session机制天然契合，代码写起来直观，调试也方便。而且Flask的生态成熟，各种扩展都有，遇到问题搜一下基本都能找到答案。对于模型推理服务这种“请求-计算-返回”的同步场景，Flask的简单直接反而是优势。

1.3 整体架构长什么样

整个系统的数据流是这样的：客户端通过HTTP POST请求把多张图片打包上传，Flask后端接收到请求后，把图片数据解析出来，然后分发给多个工作单元（线程或进程）并行执行模型推理，每个工作单元调用同一个已加载的模型对单张图片进行预测，最后把所有预测结果汇总，以JSON格式同步返回给客户端。

这里有一个关键的设计决策：模型只加载一次，所有工作单元共享。这一点非常重要，因为TensorFlow 1.x加载模型（尤其是从checkpoint恢复）是比较耗时的操作，如果每个请求都重新加载模型，那性能会惨不忍睹。所以正确的做法是在Flask应用启动的时候就把模型加载到内存里，后续所有请求复用这个模型。

架构上还有一个选择是同步还是异步返回。这个项目明确是同步返回，也就是说客户端上传完图片后，HTTP连接会一直保持，直到所有图片的预测结果都计算完成，服务端一次性返回。这种模式适合图片数量不多（比如几十张以内）、推理时间可控的场景。如果图片特别多或者推理特别慢，那就需要考虑异步任务队列的方案了，但那是另一个话题。

2. 核心细节解析与实操要点

2.1 TensorFlow 1.x模型加载的关键细节

TensorFlow 1.x和2.x在模型加载上的差异非常大，1.x用的是计算图（Graph）和会话（Session）的机制，这也是很多新手容易踩坑的地方。在1.x里面，模型加载通常有两种方式：一种是从checkpoint文件恢复，一种是从冻结的pb文件加载。

从checkpoint恢复的话，你需要先定义好网络结构，然后创建Saver对象，再调用restore方法把参数加载进来。这个过程必须在正确的Graph上下文中进行，否则会出现变量找不到的问题。从pb文件加载相对简单一些，用tf.GraphDef解析pb文件，然后导入到当前Graph中就可以了。

import tensorflow as tf class TF1ModelLoader: def __init__(self, model_path, input_name, output_name): self.graph = tf.Graph() self.input_name = input_name self.output_name = output_name with self.graph.as_default(): with tf.io.gfile.GFile(model_path, 'rb') as f: graph_def = tf.compat.v1.GraphDef() graph_def.ParseFromString(f.read()) tf.import_graph_def(graph_def, name='') self.sess = tf.compat.v1.Session(graph=self.graph) self.input_tensor = self.graph.get_tensor_by_name(input_name) self.output_tensor = self.graph.get_tensor_by_name(output_name) def predict(self, input_data): return self.sess.run(self.output_tensor, feed_dict={self.input_tensor: input_data})

上面这段代码展示了一个典型的TF1模型加载和推理的封装。注意几个关键点：第一，Graph和Session是绑定的，每个模型实例有自己的Graph和Session；第二，输入和输出的tensor名称必须和模型定义时一致，这个可以通过查看模型的节点名称来确定；第三，tf.compat.v1是为了在TF2环境下兼容TF1的API，如果你用的是纯TF1环境，直接用tf.GraphDef()和tf.Session()就行。

注意：TensorFlow 1.x的Session不是线程安全的。如果你打算用多线程来并发推理，每个线程需要有自己的Session，或者用锁来保护Session的调用。这一点非常关键，很多人在多线程环境下遇到的各种诡异错误，根源都在这里。

2.2 Keras模型加载的注意事项

Keras模型加载相对简单一些，keras.models.load_model()一行代码就能搞定。但有几个细节需要注意。首先是compile参数，如果你只是做推理不需要训练，可以设置compile=False，这样加载速度会快一些，也能避免一些自定义层或损失函数找不到的问题。

其次是TensorFlow版本兼容性。Keras现在已经被整合进TensorFlow了，tf.keras和独立的keras包在行为上可能有细微差异。如果你训练时用的是独立的Keras，加载时最好也用独立的Keras，避免版本不匹配导致的问题。

from tensorflow import keras class KerasModelLoader: def __init__(self, model_path): self.model = keras.models.load_model(model_path, compile=False) def predict(self, input_data): return self.model.predict(input_data)

Keras模型的predict方法本身是线程安全的，因为它内部会处理Session的管理。但在多进程环境下，每个进程需要独立加载模型，因为进程之间不共享内存空间。这一点和多线程有本质区别。

2.3 多线程与多进程的选择逻辑

这是这个项目最核心的技术决策点。多线程和多进程各有优劣，选哪个取决于你的具体场景。

多线程的优势在于：线程之间共享内存，模型只需要加载一次，所有线程共用同一个模型实例；创建和销毁线程的开销小；线程间通信方便。劣势在于：Python有GIL（全局解释器锁），纯Python代码的多线程无法真正并行执行。但是，TensorFlow的底层计算是用C++实现的，在执行矩阵运算等操作时会释放GIL，所以多线程在TensorFlow推理场景下是可以实现真正的并行的。

多进程的优势在于：每个进程有独立的Python解释器和内存空间，完全绕开GIL的限制，可以充分利用多核CPU。劣势在于：每个进程需要独立加载模型，内存占用成倍增加；进程间通信需要序列化，开销较大；创建进程的开销比线程大。

推荐使用多线程的原因：在模型推理场景下，TensorFlow的计算操作会释放GIL，多线程能够实现并行计算。同时，模型只需要加载一次，内存效率高。多进程虽然能绕开GIL，但每个进程都要加载一份模型，对于大模型来说内存压力太大。

2.4 批量图片上传的处理方式

Flask接收批量图片上传，通常有两种方式：一种是通过multipart/form-data表单，多个文件字段；另一种是把图片base64编码后放在JSON body里。推荐用第一种，因为它是HTTP标准的上传方式，各种客户端库都支持，而且传输效率更高（base64编码会让数据膨胀约33%）。

from flask import Flask, request, jsonify import numpy as np from PIL import Image import io app = Flask(__name__) @app.route('/predict', methods=['POST']) def predict(): files = request.files.getlist('images') if not files: return jsonify({'error': 'no images uploaded'}), 400 results = [] for file in files: img_bytes = file.read() img = Image.open(io.BytesIO(img_bytes)) img = img.resize((224, 224)) img_array = np.array(img) / 255.0 img_array = np.expand_dims(img_array, axis=0) results.append(img_array) return jsonify({'count': len(results)})

上面这段代码展示了基本的文件接收和预处理流程。request.files.getlist('images')可以获取同名字段的所有文件。每个文件读取后转成PIL Image对象，再做resize和归一化，最后转成numpy数组。这些预处理步骤在多线程环境下也是可以并行的，但通常预处理耗时远小于模型推理，所以一般不需要单独优化。

3. 实操过程与核心环节实现

3.1 环境准备与依赖安装

先把环境搭起来。Python版本建议用3.6到3.8之间，因为TensorFlow 1.x对高版本Python的支持有限。如果你用的是TF1.15，Python 3.7是比较稳妥的选择。

pip install tensorflow==1.15.0 pip install flask pip install pillow pip install numpy

如果你用的是tf.keras，那TensorFlow安装好之后Keras就有了。如果你用的是独立的Keras，还需要额外安装：

pip install keras==2.3.1

注意Keras的版本要和TensorFlow版本匹配，Keras 2.3.1对应的是TensorFlow 1.15。版本不匹配是很多奇怪报错的根源。

3.2 模型加载与全局初始化

模型加载放在Flask应用启动的时候，只执行一次。这里我用一个全局的模型管理器来封装。

import tensorflow as tf from tensorflow import keras import threading class ModelManager: _instance = None _lock = threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def init_tf1_model(self, model_path, input_name, output_name): if self._initialized: return self.graph = tf.Graph() with self.graph.as_default(): with tf.io.gfile.GFile(model_path, 'rb') as f: graph_def = tf.compat.v1.GraphDef() graph_def.ParseFromString(f.read()) tf.import_graph_def(graph_def, name='') self.sess = tf.compat.v1.Session(graph=self.graph) self.input_tensor = self.graph.get_tensor_by_name(input_name) self.output_tensor = self.graph.get_tensor_by_name(output_name) self._initialized = True def init_keras_model(self, model_path): if self._initialized: return self.model = keras.models.load_model(model_path, compile=False) self._initialized = True

这里用了单例模式来保证模型只加载一次。threading.Lock()用来保证多线程环境下初始化的线程安全。实际部署的时候，模型加载放在app.run()之前就行。

3.3 多线程推理的完整实现

多线程方案的核心是用ThreadPoolExecutor来管理线程池，每个请求到来时，把图片分发给线程池中的线程并行处理。

from concurrent.futures import ThreadPoolExecutor import numpy as np from PIL import Image import io executor = ThreadPoolExecutor(max_workers=4) def preprocess_image(file_bytes, target_size=(224, 224)): img = Image.open(io.BytesIO(file_bytes)) if img.mode != 'RGB': img = img.convert('RGB') img = img.resize(target_size) img_array = np.array(img, dtype=np.float32) / 255.0 return np.expand_dims(img_array, axis=0) def predict_single(model_manager, img_array): with model_manager.graph.as_default(): result = model_manager.sess.run( model_manager.output_tensor, feed_dict={model_manager.input_tensor: img_array} ) return result @app.route('/predict', methods=['POST']) def predict(): files = request.files.getlist('images') if not files: return jsonify({'error': 'no images'}), 400 img_arrays = [] for f in files: img_bytes = f.read() img_arrays.append(preprocess_image(img_bytes)) futures = [] for arr in img_arrays: future = executor.submit(predict_single, model_manager, arr) futures.append(future) results = [] for future in futures: try: res = future.result(timeout=30) results.append(res.tolist()) except Exception as e: results.append({'error': str(e)}) return jsonify({'count': len(results), 'predictions': results})

这段代码有几个关键点需要说明。第一，ThreadPoolExecutor的max_workers参数决定了并发度，一般设置为CPU核心数或者略多一些。第二，predict_single函数里面用了with model_manager.graph.as_default()来确保在正确的Graph上下文中执行。第三，future.result(timeout=30)设置了超时时间，防止某个推理卡死导致整个请求挂起。

实操心得：在多线程环境下调用TF1的Session时，如果遇到"Tensor is not an element of this graph"的错误，八成是因为没有在正确的Graph上下文中执行。解决办法就是在每个线程的推理代码外面包一层with graph.as_default()。

3.4 多进程推理的完整实现

多进程方案用ProcessPoolExecutor，每个进程独立加载模型。这里需要注意的是，模型加载不能放在进程池创建之后，因为进程池中的进程是fork出来的，如果模型在fork之前加载，子进程会继承父进程的内存空间，但TensorFlow的Session在fork后可能出问题。所以推荐的做法是每个子进程在启动时自己加载模型。

from concurrent.futures import ProcessPoolExecutor import multiprocessing def init_worker(model_path, input_name, output_name): global worker_model worker_model = TF1ModelLoader(model_path, input_name, output_name) def predict_in_process(img_array): global worker_model result = worker_model.predict(img_array) return result.tolist() process_executor = ProcessPoolExecutor( max_workers=2, initializer=init_worker, initargs=(model_path, input_name, output_name) )

initializer参数指定了每个工作进程启动时执行的初始化函数，initargs是传给初始化函数的参数。这样每个进程都会独立加载一份模型，虽然内存占用增加了，但完全避免了GIL的限制。

3.5 两种方案的性能对比实测

我在一台4核8G的机器上做了简单的对比测试，模型是一个小型的图像分类网络，输入尺寸224x224，单张推理耗时约50ms。测试数据是20张图片一批。

从数据可以看出，多线程在4并发下已经能把耗时降到单线程的28%左右，而多进程虽然略快一点，但内存占用是3倍多。对于大多数场景来说，多线程的性价比明显更高。

4. 常见问题与排查技巧实录

4.1 模型加载相关的典型报错

问题一：ValueError: No op named XxxOp in defined operations

这个错误通常出现在从pb文件加载模型的时候，原因是模型中使用了一些自定义的Op，而当前TensorFlow环境里没有注册这些Op。解决办法是找到定义这些Op的源码，编译成so文件后加载，或者换用checkpoint方式加载模型。

问题二：NotFoundError: Restoring from checkpoint failed

这个错误一般是checkpoint文件路径不对，或者变量名不匹配。检查一下checkpoint的路径是否正确，以及定义网络结构时的变量名是否和保存时一致。可以用tf.train.list_variables()来查看checkpoint中保存的所有变量名。

问题三：Keras加载模型时报Unknown layer

如果你在模型里用了自定义层，加载时需要把自定义层通过custom_objects参数传进去。或者更简单的方式是用compile=False加载，这样Keras不会尝试编译模型，也就不会去解析自定义的损失函数和指标。

4.2 多线程环境下的诡异问题

问题：推理结果不稳定，同样的输入有时结果不一样

这个问题在多线程环境下比较常见，根源通常是Session的线程安全问题。TF1的Session不是线程安全的，多个线程同时调用同一个Session的run方法可能会导致状态混乱。解决办法有两个：一是给Session加锁，二是每个线程用自己的Session。加锁的方案简单但会降低并发度，独立Session的方案并发度高但内存占用大。

问题：程序运行一段时间后内存持续增长

这可能是TensorFlow的内存泄漏问题，也可能是图片数据没有及时释放。检查一下是否有全局变量在累积数据，以及feed_dict中的数据是否在推理完成后被正确释放。另外，TensorFlow的Session在长时间运行后可能会有内存碎片，定期重启服务是一个简单有效的缓解办法。

4.3 批量上传的边界情况处理

问题：上传的图片格式不统一

用户上传的图片可能是JPG、PNG、BMP等各种格式，甚至可能是CMYK模式的JPG。在预处理的时候统一转成RGB模式，用img.convert('RGB')就能搞定。另外要注意处理透明通道的PNG图片，转RGB时透明部分会变成黑色，如果这对模型有影响，需要额外处理。

问题：图片尺寸差异很大

模型通常要求固定尺寸的输入，所以需要resize。但直接resize可能会改变图片的宽高比，导致内容变形。如果模型对宽高比敏感，需要先做padding再做resize。这个要根据具体模型的要求来决定。

问题：单次上传图片数量过多

如果一次上传几百张图片，即使多线程处理也可能超时。建议在接口层面限制单次上传的图片数量，比如最多50张，超过的话让客户端分批上传。同时设置合理的超时时间，避免请求长时间挂起。

4.4 常见问题速查表

4.5 几个实用的优化技巧

第一个技巧是预热。在服务正式接收请求之前，先用一张样例图片跑一次推理，让TensorFlow完成计算图的初始化和内存分配。这样第一个真实请求就不会因为初始化而特别慢。

第二个技巧是批处理。如果多张图片的推理可以合并成一个batch，那效率会比逐张推理高很多。TensorFlow对batch推理有优化，能更好地利用CPU的向量化指令。但batch size太大会增加内存占用和延迟，需要根据实际情况权衡。

第三个技巧是结果缓存。如果同样的图片可能被重复上传，可以用图片的MD5作为key缓存推理结果。这在某些场景下能大幅减少计算量。不过要注意缓存的有效期和内存占用。

第四个技巧是日志记录。记录每个请求的图片数量、推理耗时、返回状态等信息，方便后续分析和优化。用Python的logging模块就能搞定，注意日志级别和输出位置要配置好。

踩过的坑：有一次在生产环境上，服务运行了几天后突然变慢，排查发现是日志文件把磁盘写满了。所以日志一定要配置轮转（RotatingFileHandler），别让日志无限增长。

5. 部署与扩展的一些经验

5.1 开发服务器与生产服务器的区别

Flask自带的app.run()是开发服务器，单线程、性能差、不适合生产环境。生产环境推荐用Gunicorn或者uWSGI来部署。Gunicorn的配置比较简单，一条命令就能启动：

gunicorn -w 4 -b 0.0.0.0:5000 app:app --timeout 120

-w 4表示启动4个工作进程，每个进程内部还可以用多线程。这样就是进程+线程的混合并发模型，能充分利用多核CPU。--timeout 120设置请求超时时间为120秒，因为批量推理可能比较慢。

需要注意的是，如果用Gunicorn的多进程模式，每个进程都会加载一份模型，内存占用会成倍增加。所以-w的值要根据机器的内存来定，不是越大越好。

5.2 如何根据业务场景调整并发策略

并发策略没有万能的最优解，要根据实际场景来调。如果图片数量少但单张推理慢，那多线程就够了；如果图片数量多且单张推理快，那可能需要多进程来绕开GIL；如果两者都占，那就要考虑进程+线程的混合模型。

还有一个维度是请求的并发量。如果同时有很多客户端在请求，那服务端的并发度要相应提高。但并发度太高会导致上下文切换开销增大，反而降低吞吐量。一般来说，并发度设置为CPU核心数的1到2倍是比较合理的起点，然后根据实际压测结果来调整。

5.3 后续可以扩展的方向

这套方案是一个基础的推理服务框架，后续可以根据需求扩展。比如加上请求队列，当并发请求超过处理能力时，把请求排队而不是直接拒绝；加上健康检查接口，方便负载均衡器做探活；加上Prometheus监控指标，实时观察服务的QPS、延迟、错误率等；加上模型热更新，不重启服务就能切换模型版本。

如果推理量特别大，单机扛不住了，那就需要考虑分布式方案了。可以把推理服务部署到多台机器上，前面加一个负载均衡。再进一步，可以用消息队列把请求异步化，客户端提交请求后立即返回一个任务ID，然后通过轮询或者WebSocket来获取结果。不过那就完全是另一个架构了，复杂度会高很多。

我个人在实际操作中的体会是，先把单机的多线程方案跑通，把接口定义好、错误处理做好、日志打全，然后再根据实际的性能瓶颈来决定要不要上更复杂的方案。很多情况下，一台配置不错的机器加上合理的多线程配置，就能满足中小规模业务的需求了。别一上来就搞分布式，过度设计带来的维护成本往往比性能收益更大。