模型推理的推理引擎切换vLLM、TGI 与 TensorRT-LLM 对比选推理引擎不是选最快的那个而是选最契合你业务模式的那个。一、场景痛点你的团队选了一个开源 LLM 做在线推理服务第一版用 HuggingFace Transformers 直接跑——延迟 2 秒用户直接关页面。然后你开始做优化发现可选方案有 vLLM、TGIText Generation Inference、TensorRT-LLM每个都说自己是最快。你读了三个项目的 README、跑了十几个 benchmark、看了几十篇技术博客最后发现最难的并不是技术选型而是搞清楚你的业务到底是吞吐优先还是延迟优先、是单一模型还是多模型共存、是一次性部署还是需要频繁更新。没有银弹。每个引擎都是为特定场景设计的选错的代价不是性能差 20%而是整个推理架构需要推倒重来。二、底层机制与原理剖析2.1 三大引擎的核心差异2.2 PagedAttentionvLLM 的杀手特性KV Cache 是推理时最占内存的东西。传统实现中每个请求的 KV Cache 是连续分配的一块显存——这意味着你需要预留max_seq_len × num_layers × 2 × hidden_dim的内存哪怕实际请求只有 50 个 token。PagedAttention 把 KV Cache 切成固定大小的页类似操作系统中的内存分页按需分配不需要预留整块内存。这带来了两个巨大收益显存利用率从 30% 提升到 90%——同样的 GPU 可以服务 3 倍的并发请求Prefix Sharing多个请求有共同前缀如 system prompt共享同一块 KV Cache 页传统 KV Cache: [####################################___________] 浪费 40% PagedAttention: [####][####][####][____] 按需分配 ↑ 未使用的块可以给其他请求2.3 In-flight Batching vs Continuous Batching特性Continuous Batching (vLLM)In-flight Batching (TRT-LLM)合并时机每次 forward 调用前检查和重组推理进行中动态插入/移除请求实现方式Python 层调度调用 GPU kernelC 运行时 CUDA graph对新请求的响应下一个 iteration 加入当前 iteration 就能插入适合场景批量离线推理实时在线推理三、生产级代码实现3.1 推理引擎抽象层 推理引擎抽象层统一 vLLM、TGI、TensorRT-LLM 的调用接口 支持运行时引擎切换让业务代码不绑定具体引擎 设计原则: 1. 引擎选择通过环境变量或配置不硬编码 2. 统一的 generate / generate_stream 接口 3. 引擎特定的参数通过 extra_params 透传 from abc import ABC, abstractmethod from typing import AsyncIterator, Optional from dataclasses import dataclass from enum import Enum import os class InferenceEngineType(Enum): 推理引擎类型枚举 VLLM vllm TGI tgi TENSORRT_LLM tensorrt_llm HUGGINGFACE huggingface dataclass class GenerationConfig: 生成配置与引擎无关 max_tokens: int 2048 temperature: float 0.7 top_p: float 0.95 top_k: int 50 repetition_penalty: float 1.1 stop_sequences: Optional[list[str]] None # 引擎特有参数透传 extra_params: Optional[dict] None dataclass class GenerationResult: 生成结果 text: str input_tokens: int output_tokens: int finish_reason: str # stop / length / error latency_ms: float # 首 token 延迟 total_latency_ms: float # 总延迟 tokens_per_second: float # 吞吐 class InferenceEngine(ABC): 推理引擎抽象基类 所有引擎实现必须实现 generate 和 generate_stream 两个方法 abstractmethod async def generate(self, prompt: str, config: GenerationConfig) - GenerationResult: 同步生成等待完整结果 pass abstractmethod async def generate_stream( self, prompt: str, config: GenerationConfig ) - AsyncIterator[str]: 流式生成逐 token 输出 pass abstractmethod async def health_check(self) - bool: 健康检查 pass abstractmethod async def get_metrics(self) - dict: 获取引擎指标 pass # vLLM 引擎实现 class VLLMEngine(InferenceEngine): vLLM 引擎封装 使用 OpenAI 兼容 API 调用 优势: PagedAttention 高吞吐、_prefix caching、continuous batching 劣势: Python 运行时开销、冷启动慢 def __init__(self, base_url: str, model_name: str): self.base_url base_url.rstrip(/) self.model_name model_name self.client None # openai.AsyncOpenAI 实例 async def generate(self, prompt: str, config: GenerationConfig) - GenerationResult: import time start time.perf_counter() # vLLM 的 OpenAI 兼容 API response await self.client.completions.create( modelself.model_name, promptprompt, max_tokensconfig.max_tokens, temperatureconfig.temperature, top_pconfig.top_p, extra_body{ top_k: config.top_k, repetition_penalty: config.repetition_penalty, stop: config.stop_sequences, **(config.extra_params or {}) } ) elapsed time.perf_counter() - start choice response.choices[0] return GenerationResult( textchoice.text, input_tokensresponse.usage.prompt_tokens, output_tokensresponse.usage.completion_tokens, finish_reasonchoice.finish_reason or stop, latency_mselapsed * 1000, # vLLM 暂不区分首 token total_latency_mselapsed * 1000, tokens_per_secondresponse.usage.completion_tokens / elapsed ) async def generate_stream( self, prompt: str, config: GenerationConfig ) - AsyncIterator[str]: stream await self.client.completions.create( modelself.model_name, promptprompt, max_tokensconfig.max_tokens, temperatureconfig.temperature, top_pconfig.top_p, streamTrue, extra_body{ top_k: config.top_k, repetition_penalty: config.repetition_penalty, stop: config.stop_sequences, **(config.extra_params or {}) } ) async for chunk in stream: if chunk.choices and chunk.choices[0].text: yield chunk.choices[0].text async def health_check(self) - bool: try: resp await self.client.models.list() return True except Exception: return False async def get_metrics(self) - dict: vLLM 指标通过 /metrics 端点暴露 import aiohttp async with aiohttp.ClientSession() as session: async with session.get(f{self.base_url}/metrics) as resp: raw_metrics await resp.text() # 解析 Prometheus 格式指标 return {raw: raw_metrics} # TensorRT-LLM 引擎实现 class TensorRTLLMEngine(InferenceEngine): TensorRT-LLM 引擎封装 通过 Triton Inference Server 的 gRPC 接口调用 优势: GPU kernel 级优化、FP8/INT4 硬件加速、C 运行时低延迟 劣势: 模型转换复杂、不支持所有模型架构、更新慢 def __init__(self, triton_url: str, model_name: str): self.triton_url triton_url self.model_name model_name # 使用 tritonclient.grpc 或 HTTP async def generate(self, prompt: str, config: GenerationConfig) - GenerationResult: import time start time.perf_counter() # TensorRT-LLM 通过 Triton 暴露协议为 gRPC 或 HTTP # 这里使用简化的 HTTP 调用示意 import aiohttp payload { text_input: prompt, max_tokens: config.max_tokens, temperature: config.temperature, top_p: config.top_p, top_k: config.top_k, stop_words: config.stop_sequences or [], bad_words: [], stream: False } async with aiohttp.ClientSession() as session: async with session.post( f{self.triton_url}/v2/models/{self.model_name}/generate, jsonpayload ) as resp: data await resp.json() elapsed time.perf_counter() - start return GenerationResult( textdata.get(text_output, ), input_tokensdata.get(input_tokens, 0), output_tokensdata.get(output_tokens, 0), finish_reasonstop, latency_mselapsed * 1000, total_latency_mselapsed * 1000, tokens_per_seconddata.get(output_tokens, 0) / elapsed ) async def generate_stream( self, prompt: str, config: GenerationConfig ) - AsyncIterator[str]: import aiohttp payload { text_input: prompt, max_tokens: config.max_tokens, temperature: config.temperature, top_p: config.top_p, top_k: config.top_k, stop_words: config.stop_sequences or [], stream: True } async with aiohttp.ClientSession() as session: async with session.post( f{self.triton_url}/v2/models/{self.model_name}/generate_stream, jsonpayload ) as resp: async for line in resp.content: if line: text line.decode(utf-8).strip() if text.startswith(data: ): yield text[6:] async def health_check(self) - bool: import aiohttp try: async with aiohttp.ClientSession() as session: async with session.get(f{self.triton_url}/v2/health/ready) as resp: return resp.status 200 except Exception: return False async def get_metrics(self) - dict: import aiohttp async with aiohttp.ClientSession() as session: async with session.get(f{self.triton_url}/v2/models/{self.model_name}/stats) as resp: return await resp.json() # 引擎工厂 class InferenceEngineFactory: 推理引擎工厂根据配置创建对应的引擎实例 使用方式: engine InferenceEngineFactory.create() result await engine.generate(Hello, GenerationConfig()) _registry { InferenceEngineType.VLLM: VLLMEngine, InferenceEngineType.TENSORRT_LLM: TensorRTLLMEngine, } classmethod def create(cls, engine_type: Optional[InferenceEngineType] None) - InferenceEngine: 创建推理引擎实例 优先级: 参数指定 环境变量 默认 vLLM if engine_type is None: engine_type InferenceEngineType( os.getenv(INFERENCE_ENGINE, vllm) ) engine_cls cls._registry.get(engine_type) if engine_cls is None: raise ValueError(f不支持的引擎类型: {engine_type}) # 从环境变量读取引擎连接信息 base_url os.getenv( f{engine_type.value.upper()}_URL, http://localhost:8000 ) model_name os.getenv(MODEL_NAME, default) return engine_cls(base_url, model_name) # 使用示例 async def main(): engine InferenceEngineFactory.create(InferenceEngineType.VLLM) # 流式生成 config GenerationConfig(max_tokens512, temperature0.7) async for token in engine.generate_stream(解释量子纠缠, config): print(token, end, flushTrue)3.2 引擎切换的决策逻辑 推理引擎选择策略根据请求特征自动路由到最合适的引擎 路由规则: - 实时对话 (streaming, 100 tokens/s) → TensorRT-LLM (低延迟) - 批量生成 (batch, 1000 tokens) → vLLM (高吞吐) - 离线批处理 → vLLM (PagedAttention 高并发) - 非主流模型架构 → vLLM (兼容性好) from dataclasses import dataclass from enum import Enum class RequestType(Enum): CHAT chat # 实时对话流式 BATCH batch # 批量生成 OFFLINE offline # 离线批处理 dataclass class RoutingRequest: prompt: str request_type: RequestType max_tokens: int stream: bool False def route_engine(request: RoutingRequest) - InferenceEngineType: 根据请求特征选择推理引擎 # 规则 1: 实时对话 → TensorRT-LLM低延迟in-flight batching if request.request_type RequestType.CHAT and request.stream: return InferenceEngineType.TENSORRT_LLM # 规则 2: 大批量生成 → vLLMPagedAttention 高并发 prefix caching if request.request_type in (RequestType.BATCH, RequestType.OFFLINE): return InferenceEngineType.VLLM # 默认: vLLM兼容性最好 return InferenceEngineType.VLLM async def execute_with_engine(request: RoutingRequest) - str: 自动路由到最合适的引擎执行推理 engine_type route_engine(request) engine InferenceEngineFactory.create(engine_type) config GenerationConfig(max_tokensrequest.max_tokens) if request.stream: result_parts [] async for token in engine.generate_stream(request.prompt, config): result_parts.append(token) return .join(result_parts) else: result await engine.generate(request.prompt, config) return result.text四、边界分析与架构权衡4.1 性能不是唯一维度维度vLLMTGITensorRT-LLM单请求延迟中中低并发吞吐高中中模型兼容性高支持 50 架构高低需 TRT 引擎量化支持GPTQ/AWQ/SqueezeLLMGPTQ/AWQ/BitsAndBytesFP8/INT4 (硬件加速)部署复杂度低 (pip install)低 (docker)高 (模型转换 Triton)更新频率极高周更中月更低季更社区生态Python 生态完整HuggingFace 官方NVIDIA 企业支持场景建议快速验证阶段→ vLLM。pip install 起一个服务5 分钟跑通生产在线服务→ TensorRT-LLM。花一周做模型转换和 kernel 调优换取 2-3 倍的推理性能大量模型混用→ vLLM。你不可能把 50 个模型都转成 TRT 引擎HuggingFace 重度用户→ TGI。和 HF Hub 的集成最紧密4.2 引擎迁移的隐性成本从 vLLM 切到 TensorRT-LLM 不只是改几行代码模型转换时间70B 模型转 TRT 引擎大约需要 2-4 小时A100TRT 引擎文件体积通常是原始权重的 1.5-2 倍因为包含了优化后的 kernel不支持的算子某些自定义 attention 或 activation 函数需要手写 TRT plugin运维复杂度Triton Server 的配置比 vLLM 的 OpenAI API 复杂一个数量级4.3 混合部署策略我们最终采用了双引擎并存的架构TensorRT-LLM 跑流量最大的核心模型3 个模型部署在 8×A100 节点上vLLM 跑长尾模型和实验模型20 个部署在 4×A100 节点上前端统一网关根据model_name路由到不同引擎这样既享受了 TRT 的性能优势又保留了 vLLM 的灵活性和生态兼容性。五、总结选推理引擎的三个原则先看业务需求再看 benchmark。延迟敏感型选 TRT-LLM吞吐敏感型选 vLLM和 HuggingFace 生态深度绑定选 TGI。不要追求一个引擎统治所有场景。双引擎甚至三引擎并存是合理的通过统一抽象层屏蔽差异。考虑全生命周期成本不只是推理延迟。TRT-LLM 的 2ms 延迟优势可能被几小时的模型转换时间、运维复杂度、和有限的模型兼容性所抵消。最后这三个引擎的竞争正在让整个生态快速进化——vLLM 在抄 TRT 的 in-flight batchingTRT 在兼容更多模型TGI 在优化吞吐。一年后的格局可能完全不同。所以把引擎选择做成可替换的比选对更重要。