企业官网建设流程全解析

模型预测跑通只是第一步，把模型交到别人手里才算真上线。这篇Keras入门第七篇，我用Flask把Keras-bert训练出来的文本分类模型包成一个HTTP预测服务，让业务方直接通过接口调用，而不是把模型文件丢过去让人家自己跑脚本。适合已经训好模型、想快速提供预测接口的同学，也适合刚接触服务化部署、想搞懂整套流程的人。

接口服务这事听起来不复杂，但真正落地时会遇到一堆“训练时没想过”的问题：模型怎么加载一次、请求来了怎么传参、并发上来会不会崩、报错怎么回给调用方。这篇我会从方案选型、模型组装、接口实现到部署排查完整走一遍，写到能直接抄作业的程度。

1. 为什么非要用Flask包一层服务

先说清楚这件事的起点。我手头有一个用Keras-bert训练好的文本分类模型，权重文件、token字典、类别映射都在。模型在Jupyter里验证过，预测效果没问题，但业务方不可能打开Notebook去调模型，他们需要的是一句话：给我个接口，我传文本进去，你把分类结果返回给我。

直接把模型文件丢给对方肯定不行，对方没有Keras环境，没有GPU，甚至不一定装了Python。就算理论上能跑，也没人愿意为了一个分类功能去维护一套深度学习环境。用Flask包成HTTP服务，等于把模型封装成一个黑盒，调用方只需要发一个POST请求，拿JSON响应，其他什么都不用管。

为什么最后选Flask而不是别的方案，我列过一张对比表：

选Flask不是因为别的方案不好，而是这个场景下它最合适。TensorFlow Serving确实强大，但为了一个分类模型去搭一套独立服务，前期投入太大，配置项多，排查问题也更绕。FastAPI同样优秀，不过如果团队里其他人更熟悉Flask，维护成本会更低；而且预测服务瓶颈通常在模型推理，不在HTTP框架本身，Flask在单机小并发场景下完全够用。

Flask还有一个隐性优势，就是和Keras的Python生态天然贴合。模型加载、预处理、预测、后处理可以写在同一个进程里，不用跨语言、跨协议去绕。代码结构保持简单，出问题的时候打开日志就知道卡在哪一步。

这个方案适合什么体量的项目？我实践下来，日请求量几千到几万、单次推理耗时几百毫秒以内的文本分类服务，Flask加多进程部署完全hold住。真正到了每天百万级请求、需要弹性伸缩的时候，再考虑更重的服务化方案不迟。

2. Keras-bert模型构建的核心细节

2.1 tokenizer是第一个不能出错的环节

Keras-bert的模型输入不是原始字符串，而是经过切词、转id、分段之后的一组张量。很多人刚接触BERT类模型，最容易在这个环节踩坑。

BERT的输入通常包含两部分：Token ids和Segment ids。Token ids把句子切分成词或子词单元，每个单元对应一个编号；Segment ids用来区分两个句子，比如文本对任务里第一句标记为0，第二句标记为1。做单文本分类时，整个句子都标记为0。

Keras-bert里Tokenizer的核心代码逻辑大概是这样的：

from keras_bert import Tokenizer token_dict = token_dict # 训练时用的字典 tokenizer = Tokenizer(token_dict) # 单条文本转输入 text = "这部电影真的很精彩" result = tokenizer.encode(first=text, max_len=128)

这里的max_len要和训练模型时保持一致，BERT类模型对输入长度有固定上限，超过部分会被截断。如果训练时用的是128，预测时却用了64，相当于模型看到的文本信息变少了，效果会出现肉眼可见的下降。

Tokenizer的字典必须和加载模型时配套。BERT预训练模型自带一个词表，但如果你自己在业务数据上fine-tune过，新增了业务词汇，就得保证预测时用的字典是训练后保存的那份。最稳妥的做法是在训练阶段结束后，把token字典保存到文件，部署时直接读这个文件，不要自己重新构建。

这里我吃过一次亏：训练时用了包含业务词汇的扩展字典，部署时图省事直接加载了原版BERT词表，结果模型预测出来的结果完全不对，概率分布乱成一团。排查了很久才发现是字典不一致导致的。

2.2 把BERT层组装成下游分类模型

Keras-bert提供了加载预训练权重并转成Keras Layer的方法。单文本分类任务里，标准的组装方式是在BERT输出上接取第一个Token位置的特征，也就是[CLS]位置的输出，然后接全连接层和Softmax。

from keras.models import Model from keras.layers import Input, Dense, Lambda from keras_bert import load_trained_model_from_checkpoint # 加载预训练BERT模型 bert_model = load_trained_model_from_checkpoint( config_file, checkpoint_file, seq_len=128 ) # 取BERT输出的CLS位置向量 cls_output = Lambda(lambda x: x[:, 0], name="cls_output")(bert_model.output) # 下游分类层 dense = Dense(units=128, activation="relu")(cls_output) output = Dense(units=num_classes, activation="softmax")(dense) model = Model(bert_model.inputs, output)

注意这里不是直接拿bert_model.output去接全连接，BERT输出的形状是(batch_size, seq_len, hidden_size)，序列每个位置都有一个向量。文本分类任务里通常把第一个位置（CLS）的向量作为整句表示，所以要用Lambda层把它抽出来。

如果你的下游任务不是句子分类，而是序列标注，比如实体识别，那就不用取CLS，而是对每个位置的输出都接一层分类，这是后话，场景不同结构也不同。

多分类和多标签任务的输出层也有区别。多分类用Softmax，所有类别概率和为1；多标签场景每个类别独立判断，一般用Sigmoid。训练时的损失函数也要对应调整，多分类用categorical_crossentropy，多标签常用binary_crossentropy。

2.3 加载权重的坑：pickle与自定义层

Keras模型训练完之后，保存和加载也有不少讲究。BERT模型本身结构复杂，如果只是简单调用model.save("model.h5")，加载时通常没问题；但如果下游结构里用了自定义Layer，比如自己写的Attention层或Pooling层，直接加载就会报错，提示找不到自定义层。

解决办法有两个，一是在加载前注册自定义层：

from keras.utils.generic_utils import get_custom_objects get_custom_objects().update({"AttentionLayer": AttentionLayer})

二是保存的时候不要保存模型结构，只保存权重，加载时先搭好结构再load weights。我习惯用第二种，因为结构代码本身就是确定的，不用依赖序列化出来的架构，出问题更少，排查也容易。

model.load_weights("model_weights.h5")

权重的保存还有一个细节：Keras-bert加载的BERT层，参数名有固定前缀，保存权重文件时这些前缀会原样存下来。加载时模型的层名必须和保存时一致，否则load_weights会提示有些权重没匹配上。如果你在搭建模型时给BERT层的名字改了，权重加载就会失败，这也是个隐蔽的坑。

3. Flask预测服务完整实现

3.1 工程结构先行设计

部署代码不要全堆在一个文件里，我习惯按这个结构组织：

. ├── app.py ├── model.py ├── requirements.txt ├── models/ │ ├── bert_config.json │ ├── bert_model.ckpt.index │ ├── bert_model.ckpt.data-00000-of-00001 │ ├── token_dict.json │ └── class_to_idx.json

model.py负责模型加载和预测逻辑，app.py只负责HTTP接口，models目录放权重和配置文件。这样职责分开，后续如果要做模型热更新，只需要替换models目录下的文件再重载模型，不动路由代码。

class_to_idx.json是类别映射文件，内容是{"0": "负面", "1": "正面"}这种格式。训练时模型输出的数字类别，接口返回时要把数字转成人类可读的标签，这个映射文件在部署阶段不可或缺。

3.2 model.py：模型封装类

封装的思想很简单：模型加载只做一次，后续请求都复用同一个模型实例。第一次加载很慢，因为BERT权重文件通常几百MB，还要初始化图结构，如果每次请求都加载一次，服务基本没法用。

import json import numpy as np from keras_bert import Tokenizer, load_trained_model_from_checkpoint class BertClassifier: def __init__(self, config_path, checkpoint_path, token_dict_path, class_map_path, max_len=128): self.max_len = max_len self.tokenizer = Tokenizer(json.load(open(token_dict_path, "r", encoding="utf-8"))) self.class_map = json.load(open(class_map_path, "r", encoding="utf-8")) # BERT模型加载，一次到位 self.model = load_trained_model_from_checkpoint( config_file=config_path, checkpoint_file=checkpoint_path, seq_len=max_len ) # 加载下游训练好的分类权重 # 注意：这里假定model.py里已经搭好了分类模型结构 # 如果单独保存了权重，则把load_weights注释打开 def predict(self, text): # 文本转模型输入 token_ids, segment_ids = self.tokenizer.encode( first=text, max_len=self.max_len ) inputs = [ np.array([token_ids]), np.array([segment_ids]) ] # 预测 probs = self.model.predict(inputs)[0] pred_idx = int(np.argmax(probs)) confidence = float(probs[pred_idx]) label = self.class_map.get(str(pred_idx), str(pred_idx)) return { "label": label, "confidence": confidence, "prob": [round(float(p), 6) for p in probs] }

这里有几个关键点值得说明。

第一，tokenizer.encode返回的token_ids和segment_ids形状都是(seq_len,)，进入模型前要包一层变成(1, seq_len)，因为模型推理是按batch维度处理的。即使只有一条文本，也要保留batch维度。

第二，model.predict返回的概率数组形状是(1, num_classes)，所以取[0]拿到第一条样本的结果。

第三，confidence只取最大概率值，prob返回完整概率分布。接口返回完整概率是有意义的，调用方可能不只是想拿一个类别，还想知道模型对新类别的置信度，或者做阈值判断。比如某条文本预测为A类的概率是0.43，调用方可以把该条文本标记为“待人工确认”，而不是硬着头皮接受分类结果。

3.3 app.py：路由与请求处理

from flask import Flask, request, jsonify from model import BertClassifier app = Flask(__name__) # 启动时加载模型，全局复用 model = BertClassifier( config_path="models/bert_config.json", checkpoint_path="models/bert_model.ckpt", token_dict_path="models/token_dict.json", class_map_path="models/class_to_idx.json", max_len=128 ) @app.route("/health", methods=["GET"]) def health(): return jsonify({"status": "ok"}) @app.route("/predict", methods=["POST"]) def predict(): try: data = request.get_json(force=True) text = data.get("text", "").strip() if not text: return jsonify({"error": "text不能为空"}), 400 result = model.predict(text) return jsonify(result) except Exception as e: app.logger.error("predict error: %s", e) return jsonify({"error": "internal error"}), 500 if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, threaded=True)

路由设计得很简单，/predict接收JSON格式的POST请求，/health做存活检查。生产部署时负载均衡器或者监控系统会定时请求健康检查接口，如果模型挂了或者服务起不来，这个接口要能及时反馈。

再说说几个容易忽略的点。

一是request.get_json(force=True)。部分调用方发请求时没有正确设置Content-Type: application/json，加了force=True可以强制按JSON解析，避免因为请求头问题报错。

二是空文本的处理。很多线上事故都是空字符串、纯空格或者None引发的，输入校验放第一层，比模型内部去兜底要省事很多。

三是异常捕获要写在业务逻辑外层。模型预测阶段如果出错，不能直接把堆栈抛给调用方，应返回一个统一格式的错误JSON，同时把详细堆栈记到服务端日志里。调用方不需要知道你的模型内部发生了什么，只需要知道接口失败了、原因是什么。返回错误信息尽量中立一些，不要暴露太多内部实现细节。

启动服务之后，先用curl验证：

curl -X POST http://127.0.0.1:5000/predict \ -H "Content-Type: application/json" \ -d '{"text": "这部电影真的很精彩"}'

返回结果长这样：

{ "label": "正面", "confidence": 0.992, "prob": [0.003, 0.992] }

4. 部署中的性能与稳定性问题

4.1 模型加载慢和显存占用不是玄学

BERT模型文件动辄几百MB，加载一次可能需要几十秒甚至更久。第一次启动服务时，健康检查接口可能在模型还没加载完时就已经被调到，这时候返回的status应该是什么？我建议健康检查只检查进程是否存活，不检查模型是否ready，避免启动后几十秒内被误认为服务异常而反复重启。

显存占用问题在预测阶段也很现实。BERT模型加载后占用的显存远大于单次预测时的实际需求，因为模型默认按最大序列长度分配了临时存储。如果GPU显存不大，而部署机器上还跑着其他任务，建议在加载模型前显式限制GPU显存增长，或者在只需要少量并发预测时直接用CPU推理。

实际项目里我遇到过这样的情况：训练时用的GPU机器有16G显存，部署机器只有8G显存，模型加载直接OOM。最后排查发现，训练机器显存大，跑起来没事；但同一模型在8G显存上加载后，前向推理还要额外的激活值和临时张量空间。解决办法是换机器，或者降低max_len重新导出模型。所以建议在部署前先在目标机器上做一次加载和推理测试，确认资源够用再上。

4.2 并发性能与GIL是要讲清楚的

Flask默认的多线程模式有个隐藏问题：Python解释器的GIL导致多线程在CPU密集型任务上无法真正并行。模型推理是CPU密集操作，开10个线程同时跑预测，效果远不如想象中好。

我自己实测同一个BERT模型，单线程下QPS大约在50左右，开8个线程后QPS只涨到70左右，因为没有真正的并行计算，线程切换成本反而拉低了整体效率。

如果想真正提升并发上限，推荐用多进程部署，比如用gunicorn起多个worker，每个worker是一个独立Python进程，模型会在每个进程里各自加载一份。代价是内存占用成倍增加，一个模型占2G内存，4个worker就是8G，部署前要确认内存够不够。

另一个思路是维持单进程，但把预测请求放到批量队列里，每次取多条文本一起predict。因为Keras-bert底层batch预测比单条预测快很多，把一条一条地for循环预测改成一次性传一个batch，吞吐能明显提升。

probs = self.model.predict([batch_token_ids, batch_segment_ids])

batch模式并不是要等请求达到一定数量才开始，而是接收方把多条消息放队列里，攒到一定阈值或到时间窗口再统一处理。这个方案适合调用方本身有批量请求的场景，不太适合要求在几毫秒内快速响应的实时接口。

4.3 服务重启与模型更新

模型更新是部署绕不开的话题。业务方隔段时间就让重新训练模型，新权重文件需要上线而不影响调用方。最简单的做法是停服务、换权重文件、重启服务，小小地抖动一下，如果内部服务可以接受，这个方案最省事。

如果要求不中断服务，可以先加载新模型到另一个进程中，等新模型加载完成后，把流量切换过去，老进程等正在处理的请求结束后再退出。实际操作中可以用一个配置开关控制当前要使用的模型路径，切换时加载新实例，把旧实例引用替换掉。不过这种实现要处理好并发访问，模型实例在切换时可能有短暂的不一致，需要考虑加锁或者用双buffer的方式，代码复杂度会增加不少，非必要先从简单方案做起。

5. 常见报错排查速查表

下面列几个我在实际部署中遇到的报错和定位思路，给新手排障用。

这里面最容易被忽略的是“先跑一遍warmup”。BERT模型刚加载完，第一次predict会触发一些图初始化和缓冲区申请，耗时可能明显高于后续请求。如果不做warmup，服务刚起来时的第一个请求可能超时，被监控误判为服务不可用。在模型加载完成后调一次predict，把耗时的初始化放在服务启动阶段，而不是留给第一个真实请求，这个小动作对稳定性帮助很大。

另外一个坑是Keras和TensorFlow版本匹配问题。Keras-bert依赖的底层接口在不同版本下表现不太一样，经常出现module 'tensorflow' has no attribute 'get_default_session'这类报错。遇到版本问题，最快的方式是新建一个干净的虚拟环境，按requirements.txt里固定的版本安装，不要图省事在训练环境里直接跑服务，训练环境里一堆自定义模块和实验残留，很容易踩乱七八糟的坑。

我的requirements.txt大概长这样，仅供参考：

flask==1.1.4 tensorflow==1.15.0 keras==2.3.1 keras-bert==0.89.0

为什么锁版本？因为Keras-bert是一个比较早期的开源项目，对TensorFlow 2.x的兼容性并不好，如果不锁版本，pip升级一下底层依赖，服务可能直接跑不起来。

6. 踩过的坑和一点实际心得

整条链路走下来，我印象最深的是类别映射文件的重要性。模型输出的永远是一个整数索引，它本身没有业务含义。如果你在训练环境里用的是label_to_idx，部署时忘了把idx_to_label同步过去，接口返回的label全是数字，调用方根本不知道那是什么意思。这个文件别看它小，缺了它整个接口可用性直接归零。

还有一点是关于日志的。预测服务上线后，一定要记录每一条请求的耗时。我最初没在意，后来业务方反馈某个时段接口响应特别慢，我看日志也看不出原因，因为没有记录耗时。后来加了一行：

start = time.time() result = model.predict(text) cost = time.time() - start app.logger.info("predict text len=%s cost=%.3fs", len(text), cost)

有了这个日志，再遇到响应慢的问题，就能快速判断是模型推理慢还是HTTP传输慢，不用靠猜。

模型预测服务的整个体检下来，其实比训练模型本身更考验工程化思维。你要考虑的不是准确率，而是服务能不能稳定跑、接口规不规范、错误返回是否合理、出了问题能不能快速定位。很多训练做得不错的同学，第一次把模型接出去时都会被这些问题绊一下。希望这篇能帮你少走这些弯路，把Keras-bert模型的最后一步落好。