当前位置：首页 > news >正文

NLP模型工程化部署

news 2025/6/20 23:23:50

文章目录

- 一、理论-微服务、测试与GPU
- - 1）微服务架构
  - 2）代码测试
  - 3）GPU使用
- 二、实践-封装微服务，编写测试用例和脚本，并观察GPU
- - 1）微服务封装(RestFul和RPC)
  - 2）测试编写(unit_test\api_test\load_test)
  - 3）GPU使用和监控
- 三、理论-docker、CICD与K8S
- 四、实践-构建镜像与CICD脚本

一、理论-微服务、测试与GPU

学习目标与成果

1）微服务架构

①单服务架构
在这里插入图片描述
②微服务架构

③微服务特点

④主要技术栈

在这里插入图片描述
1）HTTP协议做RESTFul接口

基于flask web框架实现restful接口

2）远程RPC调用

基于grpc

restful还是rpc

2）代码测试

在这里插入图片描述

具体实践举例

代码测试

压力测试

压力测试工具
在这里插入图片描述

3）GPU使用

在这里插入图片描述
可以使用命令，每一秒刷新一次

watch -n 1 nvidia-smi

二、实践-封装微服务，编写测试用例和脚本，并观察GPU

1）微服务封装(RestFul和RPC)

①RestFul接口
（1）配置日志logger
在这里插入图片描述

import logging.handlers# 获取logger
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)# 生成文件handler，打印到文件
# 按天滚动的log，一天滚动一次，只保留最近7个日志文件（即保留最近7天）
file_handler = logging.handlers.TimedRotatingFileHandler('./logs/root.log', 'D', 1, 7, encoding='utf-8')
file_handler.setLevel(logging.DEBUG)# 设置formatter
# 打印日志时间、级别、文件名、行号、函数名字、内容
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(filename)s:%(lineno)s - %(funcName)s() - %(message)s'
)# 将formatter设置到两个handler
file_handler.setFormatter(formatter)# 将handler设置到logger
logger.addHandler(file_handler)

（2）编写RESTFUL API
1>编写hello_resource，配置健康检查路由

#hello_resource.py
from flask_restful import Resourceclass HelloResource(Resource):"""hello路由快速检查服务是否健康"""def get(self):return 'hello'

2>编写seg_resource,配置分词路由

import json
import timefrom flask import request
from flask_restful import Resourcefrom online import loggerclass SegResource(Resource):"""分词路由主要调用segment.seg"""def __init__(self, segment):# 使用传过来的segment对象，进行后面的分词self.segment = segmentdef post(self):data = request.get_json()  # 解析输入json为一个dictinit_time = time.time()result = {'status': 'OK',  # 本次请求返回状态'msg': ''  # 额外说明}request_id = data.get('request_id')  # 支持传入request_id，便于线上追踪请求try:assert data, "请确保输入不为空"# 从data取用户输入的各种参数content, model, enable_offset, enable_stop_word, use_ner = \data['content'], data.get('model'), data.get('enable_offset', False), \data.get('enable_stop_word', False), data.get('use_ner', False)logger.info('request_id: {}, model: {}, enable_offset: {}, enable_stop_word: {}, use_ner: {}, ''content: {} ...'.format(request_id, model, enable_offset, enable_stop_word, use_ner,content[:100]))# 调用segment对象的seg方法r = self.segment.seg(content, model=model, enable_offset=enable_offset,enable_stop_word=enable_stop_word, use_ner=use_ner)result['result'] = list(r)  # 将分词结果存放在result里面except Exception as e:# 出现异常，打印异常栈，更改本次请求状态为ERRORlogger.exception(e)result['status'] = 'ERROR'result['msg'] = str(e)logger.info('request_id: {}, result: {} ..., cost time: {}s'.format(request_id, json.dumps(result, ensure_ascii=False)[:200], time.time() - init_time))return result

3>编写其余resource
4>编写http server

#server.py
import sysfrom flask import Flask
from flask_restful import Apifrom online import logger
from online.http.resources.dict_resource import DictResource
from online.http.resources.hello_resource import HelloResource
from online.http.resources.pos_resource import PosResource
from online.http.resources.seg_resource import SegResource
from segment.segment import Segmentdef start_server(port=8000):# 如果输入第1个参数，将第1个参数解析为端口号if len(sys.argv) > 1:port = int(sys.argv[1])# 实例化flask appapp = Flask(__name__)app.config.update(RESTFUL_JSON=dict(ensure_ascii=False))  # 设置ensure_ascii=False，确保接口返回的中文正常api = Api(app)# 实例化segment对象，准备传入到各个resource里面segment = Segment()resource_class_kwargs = {'segment': segment}# 为api添加hello路由、seg路由、pos路由、dict路由api.add_resource(HelloResource, '/')  # hello路由用于快速检查服务可用性api.add_resource(SegResource, '/seg', resource_class_kwargs=resource_class_kwargs)  # seg路由用于分词api.add_resource(PosResource, '/pos', resource_class_kwargs=resource_class_kwargs)  # pos路由用于词性标注api.add_resource(DictResource, '/dict', resource_class_kwargs=resource_class_kwargs)  # dict路由用于管理词典# 启动服务，设置host port# host='0.0.0.0'，表示外部机器可以访问，必须设置为0.0.0.0# threaded=False，表示我们的主程序是单线程模式，需要一个一个处理请求# （我们的word_graph对象不是线程安全的）logger.info('server starts port {}'.format(port))app.run(debug=False, host='0.0.0.0', port=port, threaded=False)if __name__ == '__main__':start_server()