【Python大语言模型系列】triton通过动态批处理减少调用开销提高吞吐量（案例分析）
triton通过动态批处理减少调用开销提高吞吐量（案例分析）
这是我的第447篇原创文章。
一、引言
我们创建一个加法模型，它可以同时处理多个推理请求，每个请求包含两个数字。Triton 会将多个客户端请求合并成一个 batch。
吞吐量，是指在一次性能测试过程中网络上传输的数据量的总和。（承压能力）
系统的吞吐量与request对CPU的消耗、外部接口、IO等等紧密关联。单个request 对CPU消耗越高，外部系统接口、IO速度越慢，系统吞吐能力越低，反之越高。系统吞吐量几个重要参数：QPS（TPS）、并发数、响应时间。
-
吞吐率QPS（TPS）：特指 Web 服务器单位时间内处理的请求数。（描述其并发处理能力）
-
响应时间（RT）：执行一个请求从开始到最后收到响应数据所花费的总体时间,即从客户端发起请求到收到服务器响应结果的时间。响应时间RT(Response-time)，是一个系统最重要的指标之一，它的数值大小直接反应了系统的快慢。
-
并发数：是指系统同时能处理的请求数量，这个也是反应了系统的负载能力。
理解了上面三个要素的意义之后，就能推算出它们之间的关系：
QPS=并发数/响应时间
场景对比分析
|
响应时间 |
所需最小并发数 |
系统行为 |
发送TPS |
有效TPS |
|---|---|---|---|---|
| 5秒 | 20 × 5 = 100 |
100个worker同时工作 |
精确20 |
约20 |
| 0.2秒 | 20 × 0.2 = 4 |
仅需4个worker即可 |
精确20 |
约20 |
| 0.05秒 | 20 × 0.05 = 1 |
1个worker就能跟上 |
精确20 |
约20 |
单从定义来看，吞吐率描述了服务器在实际运行期间单位时间内处理的请求数，然而，我们更加关心的是服务器并发处理能力的上限，也就是单位时间内服务器能够处理的最大请求数，即最大吞吐率。
所以我们普遍使用 “压力测试” 的方法，通过模拟足够多数目的并发用户，分别持续发送一定的 HTTP 请求，并统计测试持续的总时间，计算出基于这种 “压力” 下的吞吐率，即为一个平均计算值。
二、实现过程
2.1 模型配置文件 config.pbtxt
代码：
name: "batch_add_model"
backend: "python"
max_batch_size: 4 # 最多合并4个请求
input [
{
name: "INPUT_A"
data_type: TYPE_FP32
dims: [1] # 每个请求包含1个数字
},
{
name: "INPUT_B"
data_type: TYPE_FP32
dims: [1]
}
]
output [
{
name: "OUTPUT_SUM"
data_type: TYPE_FP32
dims: [1]
}
]
dynamic_batching {
preferred_batch_size: [2, 4] # 优先组成2或4的batch
max_queue_delay_microseconds: 500
}2.2 Python 后端模型 model.py
代码：
import triton_python_backend_utils as pb_utils
import numpy as np
class TritonPythonModel:
def initialize(self, args):
print("模型初始化完成")
def execute(self, requests):
print(f"\n=== 收到新的 Batch，包含 {len(requests)} 个请求 ===")
responses = []
# 遍历每个请求
for i, request in enumerate(requests):
print(f"\n--- 处理 Batch 中的第 {i+1} 个请求 ---")
# 获取输入数据
input_a = pb_utils.get_input_tensor_by_name(request, "INPUT_A")
input_b = pb_utils.get_input_tensor_by_name(request, "INPUT_B")
# 转换为 numpy 并查看具体内容
a_values = input_a.as_numpy()
b_values = input_b.as_numpy()
print(f"INPUT_A 值: {a_values} (形状: {a_values.shape})")
print(f"INPUT_B 值: {b_values} (形状: {b_values.shape})")
# 执行计算
sum_result = a_values + b_values
print(f"计算结果: {sum_result}")
# 创建输出张量
output_tensor = pb_utils.Tensor("OUTPUT_SUM", sum_result.astype(np.float32))
# 创建响应
response = pb_utils.InferenceResponse(output_tensors=[output_tensor])
responses.append(response)
print(f"\n=== 返回 {len(responses)} 个响应 ===\n")
return responses
def finalize(self):
print("模型卸载")2.3 客户端并发请求代码 client.py
代码：
from tritonclient.http import InferenceServerClient, InferInput
import numpy as np
from concurrent.futures import ThreadPoolExecutor
import time
def send_request(client, a, b, request_id):
"""发送单个推理请求"""
# 准备输入数据
input_a = InferInput("INPUT_A", [1], "FP32")
input_b = InferInput("INPUT_B", [1], "FP32")
input_a.set_data_from_numpy(np.array([a], dtype=np.float32))
input_b.set_data_from_numpy(np.array([b], dtype=np.float32))
print(f"发送请求 {request_id}: {a} + {b}")
# 发送请求
response = client.infer(
model_name="batch_add_model",
inputs=[input_a, input_b],
request_id=str(request_id)
)
# 获取结果
result = response.as_numpy("OUTPUT_SUM")
print(f"收到响应 {request_id}: {a} + {b} = {result[0]}")
return result
def main():
# 连接到 Triton 服务器
client = InferenceServerClient(url="localhost:8000")
# 测试数据：3组数字
test_data = [
(10.5, 20.3), # 请求1
(15.0, 25.0), # 请求2
(30.0, 40.0) # 请求3
]
print("=== 场景 1：串行发送（间隔大）===")
for i, (a, b) in enumerate(test_data):
send_request(client, a, b, f"sync_{i}")
time.sleep(0.01) # 等待10ms，让Triton有时间处理
print("\n" + "="*50 + "\n")
print("=== 场景 2：并发发送（会触发批处理）===")
# 使用线程池同时发送3个请求
with ThreadPoolExecutor(max_workers=3) as executor:
futures = []
for i, (a, b) in enumerate(test_data):
future = executor.submit(send_request, client, a, b, f"async_{i}")
futures.append(future)
# 等待所有请求完成
for future in futures:
future.result()
if __name__ == "__main__":
main()2.4 启动 Triton 服务器
代码：
# 模型目录结构
model_repository/
└── batch_add_model/
├── config.pbtxt
└── 1/
└── model.py
# 启动服务器
docker run --rm -p 8000:8000 -v $(pwd)/model_repository:/models \
nvcr.io/nvidia/tritonserver:25.11-py3 tritonserver --model-repository=/models2.5 客户端输出（场景 1 - 串行发送）
结果：
=== 场景 1：串行发送（间隔大）===
发送请求 sync_0: 10.5 + 20.3
收到响应 sync_0: 10.5 + 20.3 = 30.8
发送请求 sync_1: 15.0 + 25.0
收到响应 sync_1: 15.0 + 25.0 = 40.0
发送请求 sync_2: 30.0 + 40.0
收到响应 sync_2: 30.0 + 40.0 = 70.02.6 Triton 服务器日志（场景 1 - 串行发送）
结果：
=== 收到新的 Batch，包含 1 个请求 ===
--- 处理 Batch 中的第 1 个请求 ---
INPUT_A 值: [10.5] (形状: (1,))
INPUT_B 值: [20.3] (形状: (1,))
计算结果: [30.8]
=== 返回 1 个响应 ===
=== 收到新的 Batch，包含 1 个请求 ===
--- 处理 Batch 中的第 1 个请求 ---
INPUT_A 值: [15.] (形状: (1,))
INPUT_B 值: [25.] (形状: (1,))
计算结果: [40.]
=== 返回 1 个响应 ===
=== 收到新的 Batch，包含 1 个请求 ===
--- 处理 Batch 中的第 1 个请求 ---
INPUT_A 值: [30.] (形状: (1,))
INPUT_B 值: [40.] (形状: (1,))
计算结果: [70.]
=== 返回 1 个响应 ===
注意：每个请求被单独处理，len(requests) 始终为 12.7 客户端输出（场景 2 - 并发）
结果：
=== 场景 2：并发发送（会触发批处理）===
发送请求 async_0: 10.5 + 20.3
发送请求 async_1: 15.0 + 25.0
发送请求 async_2: 30.0 + 40.0
收到响应 async_0: 10.5 + 20.3 = 30.8
收到响应 async_1: 15.0 + 25.0 = 40.0
收到响应 async_2: 30.0 + 40.0 = 70.02.8 Triton 服务器日志（场景 2 - 并发）
结果：
=== 收到新的 Batch，包含 3 个请求 ===
--- 处理 Batch 中的第 1 个请求 ---
INPUT_A 值: [10.5] (形状: (1,))
INPUT_B 值: [20.3] (形状: (1,))
计算结果: [30.8]
--- 处理 Batch 中的第 2 个请求 ---
INPUT_A 值: [15.] (形状: (1,))
INPUT_B 值: [25.] (形状: (1,))
计算结果: [40.]
--- 处理 Batch 中的第 3 个请求 ---
INPUT_A 值: [30.] (形状: (1,))
INPUT_B 值: [40.] (形状: (1,))
计算结果: [70.]
=== 返回 3 个响应 ===关键观察：
-
len(requests)变为 3，代表 3 个客户端请求被合并 -
但每个请求的数据仍然是独立的
[1]形状 -
Triton 自动完成了请求的聚合与响应的拆分
2.9 性能对比
|
场景 |
服务器调用次数 |
总耗时（约） |
吞吐量 |
|---|---|---|---|
| 串行发送 |
3 次 |
30ms |
低 |
| 并发发送 |
1 次 |
12ms |
高 2.5x |
这个例子清晰地展示了 Triton 动态批处理的价值：减少调用开销，提升 GPU 利用率。
技术层面
|
开销项 |
串行（3次） |
并发（1次） |
节省 |
|---|---|---|---|
| 网络通信 |
3 次 HTTP 往返 |
1 次 batch HTTP |
2x |
| Python 调用 |
3 次 |
1 次 |
2x |
| GPU 启动 |
3 次 kernel 启动 |
1 次 batched kernel |
2x |
| 日志打印 |
3 次 I/O |
1 次 I/O |
2x |
这些固定开销的减少，直接转化为吞吐量提升。
作者简介：
读研期间发表6篇SCI数据挖掘相关论文，现在某研究院从事数据算法相关科研工作，结合自身科研实践经历不定期分享关于Python、机器学习、深度学习、人工智能系列基础知识与应用案例。致力于只做原创，以最简单的方式理解和学习，关注我一起交流成长。需要数据集和源码的小伙伴可以关注底部公众号添加作者微信。
腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。
更多推荐
22
0- 0
已为社区贡献26条内容
所有评论(0)