模型部署方案：Triton Inference Server 与 FastAPI 的选型对比与混合架构

一、模型部署的"最后一公里"困境：为什么推理服务总是瓶颈

模型训练完成后，部署到生产环境是"最后一公里"也是最容易出问题的环节。典型场景：训练时模型推理只需 20ms，部署为 HTTP 服务后端到端延迟变成 200ms——其中 180ms 花在了请求解析、序列化、排队等待和框架开销上。更常见的问题是吞吐量不足：单实例 QPS 只有 50，而业务需求是 500，水平扩展又受限于 GPU 资源。

部署方案的选择直接影响推理服务的性能和运维成本。FastAPI 是轻量级 Python Web 框架，开发简单但性能受限；Triton Inference Server 是 NVIDIA 专为模型推理设计的高性能服务，支持多框架、动态 Batch 和 GPU 共享，但配置复杂、学习曲线陡峭。理解两者的适用边界是做出正确选型的前提。

二、两种方案的架构差异：通用 Web 服务 vs. 专用推理服务

FastAPI 将模型推理封装为 HTTP 接口，请求处理流程与普通 Web 服务一致；Triton 将模型作为一等公民管理，提供模型级别的调度、批处理和资源隔离。

flowchart TB subgraph FastAPI方案 FA1[HTTP 请求] --> FA2[FastAPI 路由] FA2 --> FA3[Uvicorn ASGI] FA3 --> FA4[PyTorch 推理] FA4 --> FA5[JSON 序列化] FA5 --> FA6[HTTP 响应] end subgraph Triton方案 T1[HTTP/gRPC 请求] --> T2[Triton 请求调度器] T2 --> T3[动态 Batch 合并] T3 --> T4[GPU 推理引擎] T4 --> T5[模型流水线] T5 --> T6[响应返回] T2 --> T7[多模型并发调度] T7 --> T4 end subgraph 关键差异 D1[FastAPI: 逐请求处理, 无 Batch] D2[Triton: 动态 Batch, GPU 利用率高] D3[FastAPI: 单模型部署] D4[Triton: 多模型共存, GPU 共享] end

Triton 的核心优势是动态 Batch——将多个并发请求合并为一个 Batch 执行，充分利用 GPU 的并行计算能力。在 QPS 较高时，动态 Batch 可以将吞吐量提升 5-10 倍，同时单请求延迟仅增加数毫秒。

三、两种方案的部署实现

3.1 FastAPI 推理服务

""" FastAPI 模型推理服务 适合快速验证和低 QPS 场景 """ from fastapi import FastAPI, HTTPException from pydantic import BaseModel import torch from transformers import AutoModelForSequenceClassification, AutoTokenizer import time import logging app = FastAPI(title="Text Classification Service") logger = logging.getLogger(__name__) # 全局模型和 Tokenizer model = None tokenizer = None class PredictRequest(BaseModel): """推理请求模型""" text: str max_length: int = 128 class PredictResponse(BaseModel): """推理响应模型""" label: str confidence: float latency_ms: float @app.on_event("startup") async def load_model(): """应用启动时加载模型""" global model, tokenizer model_name = "bert-base-chinese" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForSequenceClassification.from_pretrained(model_name) model.eval() if torch.cuda.is_available(): model = model.cuda() logger.info("模型加载完成") @app.post("/predict", response_model=PredictResponse) async def predict(request: PredictRequest): """单条文本推理""" if model is None: raise HTTPException(status_code=503, detail="模型未加载") start = time.perf_counter() # Tokenize inputs = tokenizer( request.text, max_length=request.max_length, truncation=True, padding=True, return_tensors="pt" ) if torch.cuda.is_available(): inputs = {k: v.cuda() for k, v in inputs.items()} # 推理 with torch.no_grad(): outputs = model(**inputs) probs = torch.softmax(outputs.logits, dim=-1) pred_idx = torch.argmax(probs, dim=-1).item() confidence = probs[0, pred_idx].item() latency = (time.perf_counter() - start) * 1000 return PredictResponse( label=str(pred_idx), confidence=confidence, latency_ms=latency ) @app.post("/batch_predict") async def batch_predict(texts: list[str]): """批量推理（手动 Batch）""" if model is None: raise HTTPException(status_code=503, detail="模型未加载") start = time.perf_counter() inputs = tokenizer( texts, max_length=128, truncation=True, padding=True, return_tensors="pt" ) if torch.cuda.is_available(): inputs = {k: v.cuda() for k, v in inputs.items()} with torch.no_grad(): outputs = model(**inputs) probs = torch.softmax(outputs.logits, dim=-1) pred_indices = torch.argmax(probs, dim=-1).tolist() confidences = probs.max(dim=-1).values.tolist() latency = (time.perf_counter() - start) * 1000 return { "predictions": [ {"label": str(idx), "confidence": conf} for idx, conf in zip(pred_indices, confidences) ], "latency_ms": latency, "batch_size": len(texts) }

3.2 Triton 推理服务配置

""" Triton Inference Server 部署配置 需要准备模型仓库目录结构 """ # 模型仓库目录结构: # model_repository/ # └── text_classifier/ # ├── config.pbtxt # 模型配置 # └── 1/ # 版本号 # └── model.onnx # ONNX 模型文件 # config.pbtxt 配置文件内容: """ name: "text_classifier" platform: "onnxruntime_onnx" max_batch_size: 32 input [ { name: "input_ids" data_type: TYPE_INT64 dims: [128] }, { name: "attention_mask" data_type: TYPE_INT64 dims: [128] } ] output [ { name: "logits" data_type: TYPE_FP32 dims: [2] } ] # 动态 Batch 配置 dynamic_batching { preferred_batch_size: [8, 16, 32] max_queue_delay_microseconds: 5000 } # 实例组配置（GPU 部署） instance_group [ { count: 2 kind: KIND_GPU gpus: [0] } ] # 优化配置 optimization { execution_accelerators { gpu_execution_accelerator: [ { name: "tensorrt" parameters { key: "precision_mode" value: "FP16" } } ] } } """ # Triton 客户端调用 import tritonclient.http as http_client import numpy as np class TritonPredictor: """Triton 推理客户端""" def __init__(self, url: str = "localhost:8000"): self.client = http_client.InferenceServerClient(url=url) def predict(self, input_ids: np.ndarray, attention_mask: np.ndarray) -> dict: """单条推理""" inputs = [ http_client.InferInput("input_ids", input_ids.shape, "INT64"), http_client.InferInput("attention_mask", attention_mask.shape, "INT64"), ] inputs[0].set_data_from_numpy(input_ids) inputs[1].set_data_from_numpy(attention_mask) outputs = [ http_client.InferRequestedOutput("logits"), ] result = self.client.infer("text_classifier", inputs, outputs) logits = result.as_numpy("logits") return {"logits": logits} def batch_predict(self, input_ids_list: list, attention_mask_list: list) -> list: """批量推理（Triton 自动 Batch）""" input_ids = np.stack(input_ids_list) attention_mask = np.stack(attention_mask_list) inputs = [ http_client.InferInput("input_ids", input_ids.shape, "INT64"), http_client.InferInput("attention_mask", attention_mask.shape, "INT64"), ] inputs[0].set_data_from_numpy(input_ids) inputs[1].set_data_from_numpy(attention_mask) outputs = [http_client.InferRequestedOutput("logits")] result = self.client.infer("text_classifier", inputs, outputs) logits = result.as_numpy("logits") return logits.tolist()

四、两种方案的适用边界与运维成本

FastAPI 的 GPU 利用率问题：FastAPI 逐请求处理模式下，GPU 在每次推理时只处理 Batch Size = 1 的输入，计算单元利用率通常低于 10%。即使使用batch_predict接口，也需要客户端自行凑 Batch，增加了调用复杂度。Triton 的动态 Batch 在服务端自动合并请求，对客户端透明，GPU 利用率可达 60-80%。

Triton 的模型格式限制：Triton 支持多种模型格式（ONNX、TensorRT、PyTorch、TensorFlow），但每种格式的配置方式不同。PyTorch 格式需要额外编写模型导出脚本，TensorRT 格式需要预构建引擎。模型更新时需要重启 Triton 或触发热加载，不如 FastAPI 灵活（FastAPI 只需替换模型文件）。

FastAPI 的开发效率优势：FastAPI 的开发周期通常为 1-2 天，包括模型加载、接口定义和基本测试。Triton 的部署周期通常为 3-5 天，包括模型导出、配置调优、性能测试和监控接入。对于快速验证场景（如 A/B 测试新模型），FastAPI 的迭代速度远快于 Triton。

混合架构的实践：生产环境中可以采用"FastAPI 前端 + Triton 后端"的混合架构。FastAPI 负责请求解析、认证、限流等业务逻辑，Triton 负责模型推理。两者通过 gRPC 通信，FastAPI 将预处理后的 Tensor 发送给 Triton，Triton 返回推理结果。这种架构兼顾了开发效率和推理性能。

五、总结

模型部署方案的选型核心在于"QPS 与开发效率"的权衡。FastAPI 适合低 QPS（< 100）和快速验证场景，开发周期短、迭代快；Triton 适合高 QPS（> 100）和 GPU 密集场景，动态 Batch 可将吞吐量提升 5-10 倍。对于中等规模场景，建议采用混合架构——FastAPI 处理业务逻辑，Triton 负责模型推理。Triton 的动态 Batch 延迟参数（max_queue_delay_microseconds）需要根据业务 SLA 调整：延迟敏感场景设为 1-2ms，吞吐优先场景设为 5-10ms。