Torchserve批量响应不匹配错误排查与多线程处理方案咨询

一、核心错误原因 #

number of batch response mismatched错误本质是handler脚本处理批量请求时，返回的结果数量与输入请求的批量大小不匹配。当Torchserve启用批量处理后，handler必须严格保证：输入N个请求的批量，输出必须是N个对应的结果，否则就会触发该503错误。

二、Handler脚本修改（关键步骤） #

以下是适配批量处理的NER模型handler示例，需根据你的模型结构调整细节：

import torch
from transformers import AutoModelForTokenClassification, AutoTokenizer
from ts.torch_handler.base_handler import BaseHandler

class NERHandler(BaseHandler):
    def initialize(self, context):
        properties = context.system_properties
        model_dir = properties.get("model_dir")
        self.device = torch.device("cpu")  # 无GPU环境指定CPU
        # 加载模型与tokenizer
        self.tokenizer = AutoTokenizer.from_pretrained(model_dir)
        self.model = AutoModelForTokenClassification.from_pretrained(model_dir)
        self.model.to(self.device)
        self.model.eval()
        self.label_list = ["O", "B-LOC", "I-LOC", "B-ORG", "I-ORG"]  # 替换为你的NER标签

    def preprocess(self, requests):
        # 批量解析请求文本
        texts = []
        for req in requests:
            # 适配请求格式，这里假设请求body包含"text"字段
            input_text = req.get("body").get("text")
            texts.append(input_text)
        # 批量tokenize，生成模型可处理的张量
        return self.tokenizer(
            texts,
            padding=True,
            truncation=True,
            return_tensors="pt"
        ).to(self.device)

    def inference(self, model_input):
        # 批量推理，禁用梯度计算
        with torch.no_grad():
            outputs = self.model(**model_input)
        logits = outputs.logits
        predictions = torch.argmax(logits, dim=-1)
        return predictions, model_input["attention_mask"]

    def postprocess(self, inference_output):
        predictions, attention_mask = inference_output
        results = []
        # 为每个请求生成对应结果，保证数量匹配
        for pred, mask in zip(predictions, attention_mask):
            token_labels = []
            # 过滤padding token的标签
            for idx, label_idx in enumerate(pred):
                if mask[idx] == 1:
                    token_labels.append(self.label_list[label_idx])
            # 提取实体（根据你的NER逻辑调整）
            entities = self._parse_entities(token_labels)
            results.append({"entities": entities})
        # 必须返回与输入批量数量一致的结果列表
        return results

    def _parse_entities(self, token_labels):
        # 从token标签中提取完整实体
        entities = []
        current_entity = None
        for idx, label in enumerate(token_labels):
            if label.startswith("B-"):
                if current_entity:
                    entities.append(current_entity)
                ent_type = label.split("-")[1]
                current_entity = {"type": ent_type, "tokens": [idx]}
            elif label.startswith("I-") and current_entity:
                current_entity["tokens"].append(idx)
            else:
                if current_entity:
                    entities.append(current_entity)
                    current_entity = None
        if current_entity:
            entities.append(current_entity)
        return entities

Handler修改要点： #

• preprocess：必须接收requests列表（批量请求），将所有请求的输入整理成模型可处理的批量张量。
• inference：模型必须支持批量输入（如shape为[batch_size, seq_len]的张量），输出批量预测结果。
• postprocess：必须将批量预测结果拆分为与输入请求数量完全一致的结果列表，每个元素对应单个请求的输出。

三、Torchserve配置调整 #

针对n1-standard-8（8核CPU）的无GPU环境，调整配置如下：

inference_address=http://0.0.0.0:8080
management_address=http://0.0.0.0:8081
metrics_address=http://0.0.0.0:8082
install_py_dep_per_model=true
prefer_direct_buffer=true
job_queue_size=10000
async_logging=true
number_of_netty_threads=8  # 匹配CPU核数，处理网络请求
netty_client_threads=8
default_workers_per_model=1
models={\
  "model": {\
    "1.0": {\
        "defaultVersion": true,\
        "marName": "legal_description.mar",\
        "minWorkers": 2,\
        "maxWorkers": 8,  # 与CPU核数一致，避免进程过载
        "batchSize": 8,  # 先从8开始测试，根据CPU负载调整
        "maxBatchDelay": 30,  # 攒批等待时间（ms），不要过长影响响应
        "responseTimeout": 300  # 延长超时时间，适配批量处理耗时
    }\
  }\
}

配置调整逻辑： #

• maxWorkers：设置为CPU核数（8），每个worker对应一个进程，最大化利用CPU资源。
• batchSize：从8开始测试，若CPU负载低于70%可逐步提升到16，避免CPU满载导致处理超时。
• maxBatchDelay：控制Torchserve等待攒批的最长时间，平衡批量效率与响应延迟。

四、验证与部署步骤 #

1. 重新打包MAR文件：确保修改后的handler被包含在内

   torch-model-archiver --model-name legal_description --version 1.0 --handler ner_handler.py --model-dir ./your_model_path --export-path model_store
   

2. 重启Torchserve服务，加载新的MAR文件
3. 并发测试：用curl/postman发送批量请求，检查返回结果数量与请求数量一致，无503错误
4. 监控CPU负载：用htop查看CPU使用率，若负载过高则调小batchSize或maxWorkers

内容的提问来源于stack exchange，提问作者RajeshM