大模型推理部署方案对比：vLLM、SGLang与TensorRT-LLM

from vllm import LLM, SamplingParams

llm = LLM(
    model="meta-llama/Meta-Llama-3-70B",
    tensor_parallel_size=4,      # 4卡张量并行
    gpu_memory_utilization=0.90,
    max_model_len=8192,
    enable_prefix_caching=True,   # 前缀缓存
    swap_space=16,                # CPU swap空间(GB)
)

sampling = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=512,
)

outputs = llm.generate(["你好，介绍一下自己"], sampling)

import sglang as sgl

@sgl.function
def multi_step_reasoning(s, question):
    s += "请分析以下问题：" + question
    s += "第一步：理解问题" + sgl.gen("understanding", max_tokens=200)
    s += "第二步：分析方案" + sgl.gen("analysis", max_tokens=300)
    s += "第三步：结论" + sgl.gen("conclusion", max_tokens=200)

# RadixAttention自动缓存前缀
engine = sgl.Engine(
    model_path="meta-llama/Meta-Llama-3-70B",
    tp_size=4,
)

result = multi_step_reasoning.run(question="AI对就业市场的影响")

import tensorrt_llm
from tensorrt_llm.runtime import ModelRunner

# 构建引擎（需先转换模型）
builder = tensorrt_llm.Builder()
config = builder.create_builder_config(
    max_batch_size=128,
    max_input_len=8192,
    max_output_len=1024,
    use_fp8=True,           # FP8量化
    use_paged_kv_cache=True,
    tokens_per_block=128,
)

engine = builder.build_engine(model, config)

# 运行推理
runner = ModelRunner(engine)
outputs = runner.generate(
    input_ids=input_ids,
    sampling_config=sampling_config
)

# 适用场景：快速部署、多模型支持、通用推理
# 推荐用于：大多数生产环境的首选

# 最佳配置
config = {
    "model": "your-model",
    "tensor_parallel_size": 4,
    "gpu_memory_utilization": 0.90,
    "max_num_seqs": 256,        # 最大并发序列
    "max_model_len": 32768,
    "enable_prefix_caching": True,
    "quantization": None,       # 或 "awq" / "fp8"
    "enforce_eager": False,     # 使用CUDA Graph优化
}

# 适用场景：需要结构化输出、多轮对话、复杂生成流程
# 推荐用于：Agent应用、工具调用、JSON输出

@sgl.function
def agent_reasoning(s, user_input, tools):
    s += system_prompt(user_input, tools)
    
    # 结构化生成：保证输出是合法JSON
    s += sgl.gen(
        "action",
        max_tokens=200,
        regex=r'\{"tool":\s*"[^"]+",\s*"params":\s*\{[^}]*\}\}'
    )
    
    action = json.loads(s["action"])
    
    if action["tool"] != "none":
        result = execute_tool(action["tool"], action["params"])
        s += f"工具结果：{result}\n"
        s += sgl.gen("response", max_tokens=500)
    
    return s

# 适用场景：对延迟和吞吐量有极致要求
# 推荐用于：大规模API服务、NVIDIA硬件环境

# 最佳使用方式：
# 1. 使用NVIDIA Triton Inference Server + TensorRT-LLM后端
# 2. 预先构建引擎（非动态加载）
# 3. 启用FP8（需H100及以上）

# 构建脚本
build_command = """
python build.py \
    --model_dir /models/llama-3-70b \
    --dtype float16 \
    --use_fp8 \
    --world_size 4 \
    --tp_size 4 \
    --pp_size 1 \
    --max_batch_size 128 \
    --max_input_len 8192 \
    --max_output_len 1024 \
    --use_paged_kv_cache \
    --tokens_per_block 128 \
    --output_dir /engines/llama-3-70b-fp8
"""

class ProductionDeployment:
    def __init__(self, model_config):
        self.config = model_config
    
    def deploy(self):
        return {
            "frontend": {
                "load_balancer": "Nginx / HAProxy",
                "api_gateway": "FastAPI + rate limiting"
            },
            "inference": {
                "engine": self._select_engine(),
                "replicas": self._compute_replicas(),
                "gpu_per_replica": self.config.gpu_per_replica,
            },
            "optimization": {
                "prefix_caching": True,
                "quantization": self.config.quantization,
                "speculative_decoding": self._setup_speculative(),
            },
            "monitoring": {
                "metrics": ["qps", "latency_p50", "latency_p99", "gpu_util"],
                "alerting": "Prometheus + AlertManager",
                "logging": "structured JSON logs"
            }
        }
    
    def _select_engine(self):
        if self.config.need_structured_output:
            return "SGLang"
        elif self.config.need_max_performance and self.config.gpu_type == "H100":
            return "TensorRT-LLM"
        else:
            return "vLLM"  # 默认选择

def setup_speculative_decoding(target_model, draft_model):
    """配置投机解码"""
    # vLLM中的投机解码
    config = {
        "model": target_model,
        "speculative_model": draft_model,  # 小模型做草稿
        "num_speculative_tokens": 5,       # 每次预测5个token
        "speculative_max_model_len": 1024, # 草稿模型的最大长度
    }
    
    # 实测效果：
    # Llama-3-70B + Llama-3-8B草稿
    # 吞吐量提升：1.8-2.2x
    # 延迟降低：1.5-1.8x
    # 准确接受率：85-90%

class CostOptimizer:
    def optimize(self, workload_pattern):
        """根据负载模式优化成本"""
        
        if workload_pattern["peak_to_avg_ratio"] > 3:
            # 高峰谷比：需要快速扩缩容
            return self._autoscaling_config()
        
        elif workload_pattern["concurrent_users"] > 1000:
            # 高并发：需要批处理优化
            return self._batching_config()
        
        else:
            # 低负载：成本优先
            return self._cost_first_config()
    
    def _autoscaling_config(self):
        return {
            "min_replicas": 2,
            "max_replicas": 10,
            "scale_up_threshold": 0.7,   # GPU利用率>70%扩容
            "scale_down_threshold": 0.3,  # GPU利用率<30%缩容
            "scale_up_delay": 30,         # 30秒后扩容
            "scale_down_delay": 300,      # 5分钟后缩容
        }

需要结构化输出(JSON/正则)？
  → SGLang

NVIDIA H100 + 追求极致性能？
  → TensorRT-LLM

其他情况？
  → vLLM（安全选择）