背景痛点：当大模型遇上效率瓶颈

在实际部署像ChatGLM和ChatGPT这类大语言模型时，我们常常会陷入一个两难境地：一方面，模型的强大能力令人向往；另一方面，其巨大的资源消耗又让人望而却步。单独部署一个模型，尤其是参数量庞大的版本，对计算资源，特别是GPU显存，构成了严峻挑战。

以典型的13B参数模型为例，在FP16精度下，仅模型参数本身就需要占用大约26GB的显存。这还没算上推理过程中产生的中间激活值（Activations）和用于加速自回归生成的KV Cache。当处理长序列（例如1024个token）时，KV Cache的显存占用会急剧增加，轻松再吃掉数GB甚至十几GB的显存。这就导致单个高端消费级显卡（如RTX 4090的24GB显存）部署一个稍大的模型都显得捉襟见肘，更不用说同时部署多个模型了。

推理延迟（Latency）是另一个核心痛点。延迟主要由两部分构成：计算延迟和内存访问延迟。大模型的计算图非常庞大，即使使用高度优化的推理引擎，单次前向传播也需要数十到数百毫秒。更关键的是，自回归生成是串行的，模型需要为生成的每一个新token运行一次完整的前向传播。这意味着生成一个100个token的回复，可能需要执行100次模型推理，总延迟很容易突破秒级，严重影响用户体验。P99延迟（99%的请求响应时间）在高并发下会进一步恶化，因为多个请求会争抢有限的计算和内存带宽资源。

因此，单纯地“堆硬件”并不是一个高效且经济的解决方案。我们需要从系统架构和软件优化的层面入手，探索如何让有限的资源发挥出最大的效能。

技术选型：并行化策略的权衡

为了应对大模型的部署挑战，业界提出了多种模型并行化策略。每种策略都有其特定的适用场景和代价。理解它们的差异是设计高效混合部署方案的基础。

并行方案	核心思想	显存优化效果	计算量分布	通信开销	适用场景
数据并行 (Data Parallelism)	同一模型副本在多卡上运行，处理不同批次的输入数据。	无优化，每卡需加载完整模型。	计算负载均衡。	中等（需同步梯度或聚合输出）。	多副本高吞吐训练，推理场景下对单请求延迟无帮助。
模型并行/张量并行 (Tensor Parallelism)	将单个模型的层内计算（如矩阵乘）拆分到多个设备上。	优秀。将参数、激活、KV Cache拆分到多卡，显著降低单卡显存压力。	计算被拆分，设备间需要频繁通信。	高。每层的前向/反向传播都需要设备间通信（All-Reduce）。	单个超大模型无法放入单卡时的必备方案。
流水线并行 (Pipeline Parallelism)	将模型的不同层组（stage）放置在不同设备上，数据像流水线一样依次通过各stage。	良好。每张卡只保存部分层的参数和激活值。	计算负载可能不均衡，存在“流水线气泡”空闲时间。	中等（相邻stage间传递激活值）。	模型层数极深，且无法用张量并行有效拆分时。
混合专家 (MoE)	模型由多个“专家”子网络构成，每层根据路由机制仅激活部分专家。	极佳（稀疏激活）。理论上参数规模大，但每次推理激活的计算量固定。	计算量由激活的专家数决定，动态变化。	高（需要路由和专家间数据分发）。	追求极大参数量（万亿级）同时保持可控计算成本的场景。

对于ChatGLM与ChatGPT的混合部署，我们的目标是在单台多卡服务器上同时服务两个（或多个）模型。此时，模型并行（张量并行） 是核心手段，它允许我们将每个模型拆分到多张GPU上，从而让原本放不下的模型变得可部署。同时，我们需要一个高效的服务网关来统一管理请求路由、批处理和负载均衡。

核心实现：构建高效混合推理服务

1. 使用FastAPI构建统一推理服务网关

网关是所有请求的入口，负责接收用户请求，并将其分发给后端的模型实例。我们选择FastAPI是因为它异步性能好，易于构建高性能API。

from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
from typing import Optional, List
import asyncio
import uuid
from enum import Enum

app = FastAPI(title="混合模型推理网关")

class ModelType(str, Enum):
    CHATGLM = “chatglm”
    CHATGPT = “chatgpt”

class InferenceRequest(BaseModel):
    model: ModelType
    prompt: str
    max_tokens: int = 512
    stream: bool = False
    priority: int = 1  # 优先级，数字越小优先级越高

class InferenceResponse(BaseModel):
    request_id: str
    text: str
    model_used: str
    latency_ms: float

# 全局请求队列和模型处理器（实际中应为独立服务）
request_queues = {
    ModelType.CHATGLM: asyncio.Queue(),
    ModelType.CHATGPT: asyncio.Queue()
}
model_processors = {}

@app.post(“/v1/completions”, response_model=InferenceResponse)
async def create_completion(request: InferenceRequest, background_tasks: BackgroundTasks):
    request_id = str(uuid.uuid4())
    # 将请求放入对应模型的队列
    try:
        await request_queues[request.model].put({
            “request_id”: request_id,
            “data”: request,
            “event”: asyncio.Event()  # 用于等待结果
        })
    except KeyError:
        raise HTTPException(status_code=400, detail=f“Unsupported model: {request.model}”)

    # 在实际场景中，这里会等待处理器返回结果
    # 为简化示例，我们模拟一个后台任务处理
    background_tasks.add_task(process_request_simulated, request.model, request_id)
    return InferenceResponse(
        request_id=request_id,
        text=“Request is being processed...”,
        model_used=request.model.value,
        latency_ms=0.0
    )

async def process_request_simulated(model_type: ModelType, req_id: str):
    await asyncio.sleep(0.1)  # 模拟处理耗时
    print(f“Processed {model_type} request: {req_id}”)

2. 动态批处理实现

动态批处理是提升GPU利用率和吞吐量的关键技术。它将短时间内到达的多个请求合并成一个批次进行推理，从而摊薄模型加载和内核启动的开销。

import threading
import time
from collections import deque
import torch

class DynamicBatchProcessor:
    def __init__(self, model_name, max_batch_size=8, max_wait_time=0.05):
        self.model_name = model_name
        self.max_batch_size = max_batch_size
        self.max_wait_time = max_wait_time  # 最大等待时间（秒）
        self.request_queue = deque()
        self.lock = threading.Lock()
        self.processing_thread = threading.Thread(target=self._process_loop, daemon=True)
        self.processing_thread.start()
        self.model = None  # 实际应加载对应的模型
        self._load_model()

    def _load_model(self):
        """加载模型，此处为示例"""
        print(f“Loading {self.model_name}...”)
        # 实际应使用深度学习框架加载模型，并设置正确的并行策略
        # 例如：使用 Hugging Face Transformers 并指定 device_map=“auto”
        pass

    def add_request(self, request_data: dict):
        """添加请求到队列"""
        with self.lock:
            self.request_queue.append(request_data)

    def _process_loop(self):
        """处理循环，不断检查并处理批次"""
        while True:
            batch = []
            batch_start_time = time.time()

            with self.lock:
                # 收集批次：要么达到最大批次大小，要么等待超时
                while len(batch) < self.max_batch_size and self.request_queue:
                    if not batch:
                        # 第一个请求，开始计时
                        batch_start_time = time.time()
                    req = self.request_queue.popleft()
                    batch.append(req)

                    # 如果等待时间已到，即使没满也立即处理
                    if time.time() - batch_start_time >= self.max_wait_time:
                        break

            if batch:
                try:
                    self._inference_batch(batch)
                except torch.cuda.OutOfMemoryError as e:
                    print(f“CUDA OOM during batch inference: {e}”)
                    # 降级策略：减小批次大小重试或返回错误
                    self._handle_oom(batch)
                except Exception as e:
                    print(f“Inference error: {e}”)
                    self._handle_inference_error(batch, e)
            else:
                time.sleep(0.001)  # 避免空转

    def _inference_batch(self, batch: list):
        """执行批次推理（示例骨架）"""
        # 1. 将batch中所有请求的prompt进行padding，组成一个张量
        # prompts = [req[‘data’].prompt for req in batch]
        # input_ids = tokenizer(prompts, padding=True, return_tensors=“pt”).to(“cuda”)

        # 2. 调用模型生成
        # with torch.no_grad():
        #     outputs = self.model.generate(**input_ids, max_new_tokens=...)

        # 3. 解码并返回结果给每个请求
        # for i, req in enumerate(batch):
        #     text = tokenizer.decode(outputs[i], skip_special_tokens=True)
        #     req[‘event’].set()  # 通知请求完成
        #     req[‘result’] = text

        print(f“Processing batch of size {len(batch)} for {self.model_name}”)
        for req in batch:
            req[‘event’].set()  # 模拟完成

    def _handle_oom(self, batch):
        """处理OOM：尝试更小的批次"""
        print(“Handling OOM by splitting batch...”)
        mid = len(batch) // 2
        if mid > 0:
            with self.lock:
                # 将后半部分放回队列前端
                self.request_queue.extendleft(reversed(batch[mid:]))
            # 递归处理前半部分
            self._inference_batch(batch[:mid])
        else:
            # 单个请求也OOM，返回错误
            for req in batch:
                req[‘error’] = “CUDA Out of Memory”
                req[‘event’].set()

    def _handle_inference_error(self, batch, error):
        """处理其他推理错误"""
        for req in batch:
            req[‘error’] = str(error)
            req[‘event’].set()

3. 模型分片加载与显存优化技巧

为了在单台服务器上同时部署ChatGLM和ChatGPT，我们必须利用多GPU，并通过模型分片来降低单卡显存压力。

使用 Hugging Face accelerate 进行自动模型分片： 这是最简便的方法。accelerate库能自动将模型的不同层分配到可用的GPU上。

pip install accelerate

from transformers import AutoModelForCausalLM, AutoTokenizer
from accelerate import init_empty_weights, load_checkpoint_and_dispatch

model_name = “THUDM/chatglm3-6b”  # 或 ChatGPT 对应的模型
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)

# 方案A：使用 device_map=“auto” (推荐)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.float16,
    device_map=“auto”,  # 关键参数，自动分配模型层到多GPU
    trust_remote_code=True
)

# 方案B：更精细的控制（对于超大模型或特殊架构）
with init_empty_weights():  # 在meta设备上初始化，不占用真实显存
    model = AutoModelForCausalLM.from_config(config)

# 然后将模型 checkpoint 加载并分发到多个设备上
model = load_checkpoint_and_dispatch(
    model,
    checkpoint=model_name,
    device_map=“balanced”,  # 尝试在可用设备上平衡内存使用
    no_split_module_classes=[“GLMBlock”]  # 指定哪些模块不应该被拆分（针对ChatGLM）
)

CUDA MPS (Multi-Process Service) 配置： 对于多进程推理场景（例如每个模型一个进程），启用MPS可以提高GPU利用率，减少上下文切换开销。但请注意，MPS与某些模型并行技术可能不兼容。

# 启动MPS守护进程
sudo nvidia-smi -i 0 -c EXCLUSIVE_PROCESS  # 将GPU 0设置为独占进程模式
export CUDA_VISIBLE_DEVICES=0
nvidia-cuda-mps-control -d  # 启动MPS守护进程

# 在你的推理进程中，正常使用CUDA即可

关键优化参数：

• torch_dtype=torch.float16 或 torch.bfloat16：使用半精度，显著减少显存占用。
• load_in_8bit / load_in_4bit (BitsAndBytes)：使用量化技术，进一步压缩模型。
• use_cache=True：启用KV Cache以加速自回归生成，但会占用额外显存。对于长文本，可考虑 PagedAttention（vLLM等推理引擎支持）来优化KV Cache管理。

性能测试：用数据说话

设计一个科学的性能测试是验证优化效果的关键。

测试环境配置：

• 硬件：单台服务器，配备2张 NVIDIA A100 80GB PCIe GPU。
• 软件：Python 3.9, PyTorch 2.1, CUDA 11.8, Transformers 4.35。
• 模型：ChatGLM3-6B, GPT-2 XL（1.5B，用于模拟ChatGPT的较小版本进行对比测试）。

测试方案：

1. 独立部署：每个模型独占一张A100 GPU。
2. 混合部署：使用device_map=“auto”，让两个模型共享两张GPU，由加速器自动分配层。

我们使用 locust 模拟并发请求，并采集以下指标：

• QPS (Queries Per Second)：每秒成功处理的请求数。
• P99 Latency：99%的请求响应时间。
• GPU Utilization：GPU计算和显存使用率。

Prometheus监控指标采集： 我们可以使用 prometheus_client 库在推理服务中暴露指标。

from prometheus_client import Counter, Histogram, start_http_server

# 定义指标
REQUEST_COUNT = Counter(‘inference_requests_total’, ‘Total inference requests’, [‘model’, ‘status’])
REQUEST_LATENCY = Histogram(‘inference_latency_seconds’, ‘Inference latency in seconds’, [‘model’])
GPU_MEMORY_USAGE = Gauge(‘gpu_memory_usage_bytes’, ‘GPU memory usage in bytes’, [‘device_id’])

@app.post(“/v1/completions”)
async def create_completion(request: InferenceRequest):
    start_time = time.time()
    REQUEST_COUNT.labels(model=request.model, status=“received”).inc()
    try:
        # ... 处理逻辑 ...
        latency = time.time() - start_time
        REQUEST_LATENCY.labels(model=request.model).observe(latency)
        REQUEST_COUNT.labels(model=request.model, status=“success”).inc()
        return result
    except Exception as e:
        REQUEST_COUNT.labels(model=request.model, status=“error”).inc()
        raise e

# 在另一个线程中定期收集GPU信息
def collect_gpu_metrics():
    import pynvml
    pynvml.nvmlInit()
    while True:
        for i in range(pynvml.nvmlDeviceGetCount()):
            handle = pynvml.nvmlDeviceGetHandleByIndex(i)
            mem_info = pynvml.nvmlDeviceGetMemoryInfo(handle)
            GPU_MEMORY_USAGE.labels(device_id=str(i)).set(mem_info.used)
        time.sleep(5)

# 启动指标采集服务器
start_http_server(8000)
threading.Thread(target=collect_gpu_metrics, daemon=True).start()

预期结果： 在合理的请求负载下，混合部署方案有望在总体吞吐量(QPS)上超越独立部署，因为GPU资源得到了更充分的利用。对于P99延迟，混合部署可能会因为资源争用而略有增加，但通过良好的动态批处理和优先级调度，可以将其控制在可接受的范围内（例如，增加50ms以内）。GPU显存的使用率将从独立部署时的“一张卡满，一张卡闲”变为“两张卡都处于较高利用率”。

避坑指南：实战中总结的经验

1. 解决CUDA OOM的3种实践方法

• 梯度累积与激活检查点（训练时）：对于微调场景，使用 gradient_checkpointing=True 可以以计算时间换取显存，大幅减少中间激活值的存储。
• 即时清理缓存：在推理循环中，定期使用 torch.cuda.empty_cache()。但注意，频繁调用此函数可能带来性能开销。
• 量化与卸载：
  • 8-bit/4-bit量化：使用 bitsandbytes 库进行模型量化，这是解决显存问题最有效的手段之一。

  from transformers import BitsAndBytesConfig
  bnb_config = BitsAndBytesConfig(
      load_in_4bit=True,
      bnb_4bit_compute_dtype=torch.float16,
      bnb_4bit_use_double_quant=True,
  )
  model = AutoModelForCausalLM.from_pretrained(model_name, quantization_config=bnb_config, device_map=“auto”)
  

  • CPU卸载：将暂时不用的层或优化器状态卸载到CPU内存，需要时再加载回GPU。accelerate 的 device_map 可以配置 “cpu” 或 “disk”。

2. 长文本输入的分块处理策略

当输入上下文远超模型最大长度时：

• 滑动窗口：只保留最近N个token的上下文。简单有效，但会丢失远期记忆。
• 关键信息提取/总结：使用另一个轻量模型或规则，将长文档总结成较短的提示。
• 使用支持长上下文的技术：
  • 位置插值（Position Interpolation）：如 ChatGLM3 支持 LongLoRA 技术，通过微调将上下文窗口从2K扩展到32K甚至更长。
  • 流式处理API：如果模型服务方提供流式长上下文接口，优先使用。

3. 模型热更新的零停机方案

业务需要更新模型版本时，如何避免服务中断？

1. 蓝绿部署：准备两套完全独立的环境（蓝组和绿组）。将新模型部署到绿组，并进行充分验证。然后，将网关的流量从蓝组切换到绿组。旧版本（蓝组）保持运行一段时间以备回滚。
2. 影子部署：新模型实例启动后，并不直接承接生产流量，而是复制一份生产流量（影子流量）对其进行测试。验证无误后，再通过更改负载均衡配置或网关路由，逐步将流量切至新实例。
3. 模型版本化与动态加载：在服务代码中，模型加载不绑定固定路径，而是通过版本号或配置中心决定加载哪个模型文件。更新时，先加载新模型到内存，待加载成功后，通过原子操作切换网关指向新模型的内存地址。此方案对架构设计要求较高。

延伸思考：基于负载预测的弹性伸缩

当线上流量波动剧烈时，固定的混合部署资源可能仍会面临压力或闲置。我们可以设计一个弹性伸缩方案：

1. 数据采集：持续收集历史QPS、延迟、GPU利用率、请求队列长度等指标。
2. 负载预测：使用时间序列预测模型（如Prophet、LSTM），对未来几分钟的请求流量进行预测。
3. 决策引擎：制定伸缩策略。例如：
   • 如果预测未来5分钟流量上涨，且当前GPU利用率持续高于80%，则触发水平扩展：在Kubernetes中拉起一个新的推理服务Pod（包含混合模型部署）。
   • 如果预测流量下降，且利用率持续低于30%超过10分钟，则触发水平收缩：优雅排空一个Pod的流量后将其关闭。
   • 垂直伸缩：在容器内，根据负载动态调整每个模型的并行度（例如，低负载时让两个模型共享一张卡；高负载时通过资源限制确保每个模型独占一卡）。
4. 执行与反馈：通过K8s API或云服务商API执行伸缩操作，并监控操作后的实际效果，形成闭环反馈，优化预测模型和策略参数。

这个方案将静态的混合部署升级为动态的、智能的资源调度系统，能够在保证服务质量的前提下，最大化资源利用率，降低成本。

---

优化大模型推理效率是一场贯穿算法、系统和工程的综合挑战。从模型分片、动态批处理到弹性伸缩，每一步优化都离不开对计算、内存和通信三大资源的深刻理解与精细权衡。希望这篇从实战角度出发的笔记，能为你部署自己的ChatGLM、ChatGPT或其他大模型提供一条清晰的路径。

如果你对从零开始构建一个完整的、可交互的AI应用更感兴趣，想体验将语音识别、大模型对话和语音合成串联起来的乐趣，我强烈推荐你试试这个 从0打造个人豆包实时通话AI 动手实验。它带你走完一个实时语音AI应用的全链路，从API调用到Web集成，步骤清晰，环境都准备好了，非常适合作为大模型应用落地的第一个实践项目。我跟着做了一遍，把几个关键服务调通的过程，对理解整个AI应用的架构帮助特别大。