Llama-3.2V-11B-COT 部署：Triton 推理服务封装与压测

构建 Triton 模型仓库

Triton 通过目录结构管理模型，称为'模型仓库'。

创建仓库结构

mkdir -p triton_model_repository/llama_3_2v_11b_cot/1

• triton_model_repository: 根目录。
• llama_3_2v_11b_cot: 模型名称。
• 1: 版本号。

准备配置文件

将模型文件复制到版本目录中，并在模型名目录下创建 config.pbtxt。

name: "llama_3_2v_11b_cot"
platform: "onnxruntime_onnx"
max_batch_size: 4
input [
  { name: "pixel_values" data_type: TYPE_FP32 dims: [3, 336, 336] },
  { name: "input_ids" data_type: TYPE_INT64 dims: [-1] },
  { name: "attention_mask" data_type: TYPE_INT64 dims: [-1] }
]
output [
  { name: "output_0" data_type: TYPE_FP32 dims: [-1, -1] }
]
instance_group [
  { count: 1 kind: KIND_GPU gpus: [0] }
]
dynamic_batching {
  preferred_batch_size: [1, 2, 4]
  max_queue_delay_microseconds: 500000
}

配置告诉 Triton 输入输出维度、实例数量以及启用动态批处理策略。

集成模型流水线

视觉语言模型需要预处理、推理、后处理三个步骤。利用 Triton 的集成模型功能串联它们。

创建 ensemble_model 目录并配置 config.pbtxt：

name: "llama_3_2v_11b_cot_ensemble"
platform: "ensemble"
max_batch_size: 4
input [
  { name: "IMAGE" data_type: TYPE_UINT8 dims: [-1, -1, 3] },
  { name: "QUESTION" data_type: TYPE_STRING dims: [-1] }
]
output [
  { name: "ANSWER" data_type: TYPE_STRING dims: [-1] }
]
ensemble_scheduling {
  step [
    { model_name: "llama_3_2v_11b_cot_preprocess" ... }
    { model_name: "llama_3_2v_11b_cot" ... }
    { model_name: "llama_3_2v_11b_cot_postprocess" ... }
  ]
}

预处理模型 (model.py)：负责图像归一化和文本分词。

import triton_python_backend_utils as pb_utils
import numpy as np
from PIL import Image
import torch
from transformers import AutoProcessor

class TritonPythonModel:
    def initialize(self, args):
        self.processor = AutoProcessor.from_pretrained("/home/user/llama-3.2v-11b-cot")

    def execute(self, requests):
        responses = []
        for request in requests:
            image_np = pb_utils.get_input_tensor_by_name(request, "IMAGE").as_numpy()[0]
            question_text = pb_utils.get_input_tensor_by_name(request, "QUESTION").as_numpy()[0].decode('utf-8')
            
            image_pil = Image.fromarray(image_np)
            image_tensor = self.processor.image_processor(image_pil, return_tensors="pt")["pixel_values"]
            text_encoding = self.processor.tokenizer(question_text, return_tensors="pt", padding=True, truncation=True)
            
            out_pixel_values = pb_utils.Tensor("pixel_values", image_tensor.numpy().astype(np.float32))
            out_input_ids = pb_utils.Tensor("input_ids", text_encoding["input_ids"].numpy().astype(np.int64))
            out_attention_mask = pb_utils.Tensor("attention_mask", text_encoding["attention_mask"].numpy().astype(np.int64))
            
            inference_response = pb_utils.InferenceResponse(output_tensors=[out_pixel_values, out_input_ids, out_attention_mask])
            responses.append(inference_response)
        return responses

后处理模型 (model.py)：负责将 logits 解码为文本。

import triton_python_backend_utils as pb_utils
import numpy as np
import torch
from transformers import AutoProcessor

class TritonPythonModel:
    def initialize(self, args):
        self.processor = AutoProcessor.from_pretrained("/home/user/llama-3.2v-11b-cot")
        self.tokenizer = self.processor.tokenizer

    def execute(self, requests):
        responses = []
        for request in requests:
            logits_np = pb_utils.get_input_tensor_by_name(request, "logits").as_numpy()
            logits_tensor = torch.from_numpy(logits_np)
            predicted_token_ids = torch.argmax(logits_tensor, dim=-1)
            generated_text = self.tokenizer.batch_decode(predicted_token_ids, skip_special_tokens=True)
            
            out_answer = pb_utils.Tensor("answer", np.array(generated_text, dtype=object))
            inference_response = pb_utils.InferenceResponse(output_tensors=[out_answer])
            responses.append(inference_response)
        return responses

至此，完整的模型仓库已就绪，对外提供简洁的图片和问题输入接口。

启动与测试

Docker 启动

推荐使用 Docker 避免依赖冲突。

docker pull nvcr.io/nvidia/tritonserver:23.10-py3
docker run --gpus=all --rm -p 8000:8000 -p 8001:8001 -p 8002:8002 \
  -v /path/to/your/triton_model_repository:/models \
  nvcr.io/nvidia/tritonserver:23.10-py3 \
  tritonserver --model-repository=/models

日志显示 READY 状态即表示加载成功。

客户端测试

编写 Python 客户端发送请求。

import requests
import json
import base64

url = "http://localhost:8000/v2/models/llama_3_2v_11b_cot_ensemble/infer"
image_path = "test_image.jpg"
question = "What is in this image?"

with open(image_path, "rb") as f:
    image_data = f.read()

payload = {
    "inputs": [
        { "name": "IMAGE", "shape": [1], "datatype": "BYTES", "data": [image_data] },
        { "name": "QUESTION", "shape": [1], "datatype": "BYTES", "data": [question.encode('utf-8')] }
    ],
    "outputs": [{"name": "ANSWER"}]
}

response = requests.post(url, data=json.dumps(payload), headers={"Content-Type": "application/json"})
if response.status_code == 200:
    answer = response.json()['outputs'][0]['data'][0].decode('utf-8')
    print(answer)

性能压测与优化

使用 Perf Analyzer

Triton 自带 perf_analyzer 工具模拟并发请求。

perf_analyzer -m llama_3_2v_11b_cot_ensemble \
  -u localhost:8000 \
  --concurrency-range 1:8:2 \
  --measurement-mode count_windows \
  --measurement-request-count 100

需准备包含真实数据的 JSON 文件作为输入源。关注吞吐量 (Inferences/Second)、延迟 (Latency) 和 GPU 利用率。

优化方向

• 调整批处理大小：增大 max_batch_size 可提升吞吐量，但会增加单请求延迟，需根据业务权衡。
• 模型格式优化：尝试 TensorRT 引擎或 FP16/INT8 量化，减少显存占用并加速计算。
• 增加实例：在 instance_group 中设置 count > 1，利用多副本并行处理请求。
• 多 GPU 部署：指定 gpus: [0,1] 实现负载均衡。

优化是一个迭代过程，目标是找到满足业务需求的最优配置。

总结

从简单的脚本验证到生产级服务，关键在于构建稳定的推理流水线。通过 Triton 封装预处理、推理和后处理，实现了高并发与动态批处理能力。配合性能压测工具，可以持续调优以应对实际流量。这套方案让 AI 能力从'玩具'迈向了可靠的'工具'，为海量图片理解任务提供了坚实基础。