2. 模型服务化基础：模型部署的常见方式与核心指标

将训练好的机器学习模型封装为可远程调用的服务，是低延迟评分系统的第一步。不同的服务化协议在延迟、吞吐量、连接管理和开发复杂度上差异显著。本章聚焦三种主流部署方式——REST API、gRPC、WebSocket，并定义衡量服务性能的核心指标。

2.1 三种部署方式的协议对比

2.2 延迟与吞吐量的核心指标

衡量模型服务性能时，不能只看单次推理时间。必须从网络传输、序列化、排队和推理四个环节综合评估。以下是必须掌握的核心指标：

• P50 / P99 延迟：50% 和 99% 分位的端到端响应时间。P99 反映尾部延迟，对 SLA 至关重要。
• 吞吐量 (Throughput)：单位时间内成功处理的请求数，通常以 QPS (Queries Per Second) 或 TPS (Transactions Per Second) 表示。
• 并发连接数：服务同时维持的活跃连接数量，影响资源消耗和排队延迟。
• 网络往返时间 (RTT)：客户端与服务端之间的网络延迟，在跨区域部署时可能成为瓶颈。
• 序列化/反序列化耗时：将请求数据转为传输格式 (JSON/Protobuf) 及反向转换的时间。
• 排队延迟：请求在服务端队列中等待处理的时间，与线程池/协程模型强相关。

2.3 三种方式的延迟与吞吐量对比 (SVG 架构图)

下图从连接模型和数据流角度，对比 REST API、gRPC 和 WebSocket 在典型评分场景下的延迟构成与吞吐量表现。

2.4 延迟与吞吐量的权衡关系

延迟和吞吐量并非独立指标，它们之间存在经典的排队论关系。当系统接近饱和时，延迟会急剧上升。理解以下权衡有助于设计服务：

• 低延迟 ≠ 高吞吐量：为追求极低延迟而限制并发 (如单线程推理)，会降低吞吐量。
• 批处理提升吞吐量但增加延迟：将多个请求合并为一批推理，可提高 GPU 利用率，但首个请求需等待批处理窗口。
• 连接复用降低延迟：gRPC 和 WebSocket 通过长连接避免重复握手，显著降低 P50 延迟。
• 序列化开销不可忽视：在 1KB 以下的小请求中，JSON 序列化可能占端到端延迟的 30%~50%。

2.5 选择部署方式的决策流程

以下流程图帮助你在实际项目中快速选择最合适的服务化方式：

2.6 代码示例：三种方式的最小实现

以下展示使用 Python 实现三种服务化方式的最小代码骨架，便于理解核心差异：

REST API (Flask)

from flask import Flask, request, jsonify
import numpy as np

app = Flask(__name__)

# 模拟模型推理
def predict(features: list) -> float:
    return float(np.mean(features))

@app.route('/score', methods=['POST'])
def score():
    data = request.get_json()
    features = data.get('features', [])
    result = predict(features)
    return jsonify({'score': result, 'latency_ms': 0.0})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, threaded=True)

gRPC (使用 protobuf)

# 定义 proto 文件
# service Scorer { rpc Score(ScoreRequest) returns (ScoreResponse); }
# message ScoreRequest { repeated float features = 1; }
# message ScoreResponse { float score = 1; }

# server.py
import grpc
from concurrent import futures
import scorer_pb2, scorer_pb2_grpc

class ScorerServicer(scorer_pb2_grpc.ScorerServicer):
    def Score(self, request, context):
        features = list(request.features)
        score = sum(features) / len(features)  # 模拟推理
        return scorer_pb2.ScoreResponse(score=score)

server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
scorer_pb2_grpc.add_ScorerServicer_to_server(ScorerServicer(), server)
server.add_insecure_port('[::]:50051')
server.start()
server.wait_for_termination()

WebSocket (FastAPI + WebSockets)

from fastapi import FastAPI, WebSocket
import numpy as np

app = FastAPI()

@app.websocket("/ws/score")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_json()
        features = data.get('features', [])
        score = float(np.mean(features))
        await websocket.send_json({'score': score})

2.7 延迟与吞吐量的测量方法

使用标准工具对三种方式进行基准测试，获取可对比的指标：