随着大模型应用逐渐进入生产环境，仅仅完成模型调用已经无法满足业务需求。一个完整的大模型服务通常还需要具备以下能力：

• 通过 API 对外提供统一服务
• 使用 Docker 快速部署
• 支持多个模型或多个模型版本
• 根据权重进行请求分流
• 支持模型懒加载，降低启动时间
• 返回类似 OpenAI API 的标准响应格式
• 便于后续扩展鉴权、限流、监控和日志

本文使用 FastAPI、Transformers 和 Docker，构建一个简易的大模型服务平台。

一、整体架构

客户端
  |
  | HTTP 请求
  v
FastAPI 网关
  |
  | 模型名称 / 版本 / 权重
  v
模型路由器
  |
  +--> chat-small@1.0.0
  |
  +--> chat-small@1.1.0
  |
  +--> embedding@1.0.0

请求模型时：

model = chat-small

表示按照权重自动选择可用版本。

model = chat-small@1.0.0

表示固定调用指定版本。

二、项目结构

llm-service/
├── app/
│   └── main.py
├── models.json
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

三、安装依赖

创建 requirements.txt：

fastapi>=0.115,<1.0
uvicorn[standard]>=0.32,<1.0
transformers>=4.45,<5.0
torch>=2.2,<2.6
pydantic>=2.7,<3.0

安装依赖：

pip install -r requirements.txt

四、模型配置

创建 models.json：

{
  "models": [
    {
      "name": "chat-small",
      "version": "1.0.0",
      "model_path": "Qwen/Qwen2.5-0.5B-Instruct",
      "enabled": true,
      "weight": 3,
      "aliases": ["small", "stable"]
    },
    {
      "name": "chat-small",
      "version": "1.1.0",
      "model_path": "Qwen/Qwen2.5-1.5B-Instruct",
      "enabled": true,
      "weight": 1,
      "aliases": ["canary"]
    }
  ]
}

其中：

• name：模型逻辑名称
• version：模型版本
• model_path：HuggingFace 模型名称或本地路径
• enabled：是否参与调度
• weight：流量权重
• aliases：模型别名

当前配置中，1.0.0 和 1.1.0 的流量比例约为 3:1，可以用于灰度发布。

五、FastAPI 服务实现

创建 app/main.py：

import asyncio
import json
import os
import threading
import time
import uuid
from typing import Literal

import torch
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from transformers import AutoModelForCausalLM, AutoTokenizer


class ModelConfig(BaseModel):
    name: str
    version: str
    model_path: str
    enabled: bool = True
    weight: int = Field(default=1, ge=1)
    aliases: list[str] = Field(default_factory=list)

    @property
    def route_id(self) -> str:
        return f"{self.name}@{self.version}"


class ChatMessage(BaseModel):
    role: Literal["system", "user", "assistant"]
    content: str


class ChatRequest(BaseModel):
    model: str | None = None
    messages: list[ChatMessage] = Field(min_length=1)
    temperature: float = Field(default=0.7, ge=0, le=2)
    top_p: float = Field(default=0.9, gt=0, le=1)
    max_tokens: int = Field(default=256, ge=1, le=4096)


class ModelRuntime:
    """
    一个模型实例的运行时封装。

    模型采用懒加载策略：
    第一次收到请求时才加载模型，避免服务启动时阻塞。
    """

    def __init__(self, config: ModelConfig):
        self.config = config
        self.device = torch.device(
            "cuda" if torch.cuda.is_available() else "cpu"
        )

        self.tokenizer = None
        self.model = None

        self._load_lock = threading.Lock()
        self._generation_lock = threading.Lock()

    @property
    def loaded(self) -> bool:
        return self.model is not None

    def load(self):
        if self.model is not None:
            return

        with self._load_lock:
            if self.model is not None:
                return

            print(
                f"Loading model: {self.config.route_id}, "
                f"path={self.config.model_path}"
            )

            self.tokenizer = AutoTokenizer.from_pretrained(
                self.config.model_path,
                trust_remote_code=True
            )

            if self.tokenizer.pad_token is None:
                self.tokenizer.pad_token = self.tokenizer.eos_token

            dtype = (
                torch.float16
                if self.device.type == "cuda"
                else torch.float32
            )

            self.model = AutoModelForCausalLM.from_pretrained(
                self.config.model_path,
                torch_dtype=dtype,
                trust_remote_code=True
            )

            self.model.to(self.device)
            self.model.eval()

            print(
                f"Model loaded: {self.config.route_id}, "
                f"device={self.device}"
            )

    def build_prompt(self, messages: list[dict[str, str]]) -> str:
        if getattr(self.tokenizer, "chat_template", None):
            return self.tokenizer.apply_chat_template(
                messages,
                tokenize=False,
                add_generation_prompt=True
            )

        # 兼容没有 chat_template 的模型
        parts = []

        for message in messages:
            parts.append(
                f"{message['role']}: {message['content']}"
            )

        parts.append("assistant:")
        return "\n".join(parts)

    def generate(
        self,
        messages: list[dict[str, str]],
        temperature: float,
        top_p: float,
        max_tokens: int
    ) -> tuple[str, int, int]:
        self.load()

        # 同一个模型实例只允许一个生成任务进入，避免显存竞争
        with self._generation_lock:
            prompt = self.build_prompt(messages)

            inputs = self.tokenizer(
                prompt,
                return_tensors="pt"
            )

            inputs = {
                key: value.to(self.device)
                for key, value in inputs.items()
            }

            generation_args = {
                "max_new_tokens": max_tokens,
                "pad_token_id": self.tokenizer.pad_token_id,
                "eos_token_id": self.tokenizer.eos_token_id,
                "do_sample": temperature > 0
            }

            if temperature > 0:
                generation_args.update({
                    "temperature": temperature,
                    "top_p": top_p
                })

            start_time = time.time()

            with torch.inference_mode():
                outputs = self.model.generate(
                    **inputs,
                    **generation_args
                )

            input_token_count = inputs["input_ids"].shape[1]
            new_tokens = outputs[0][input_token_count:]

            answer = self.tokenizer.decode(
                new_tokens,
                skip_special_tokens=True
            ).strip()

            elapsed = time.time() - start_time

            print(
                f"model={self.config.route_id}, "
                f"input_tokens={input_token_count}, "
                f"output_tokens={len(new_tokens)}, "
                f"elapsed={elapsed:.2f}s"
            )

            return answer, input_token_count, len(new_tokens)


class ModelRegistry:
    """
    模型注册中心和简单的加权轮询调度器。
    """

    def __init__(self, configs: list[ModelConfig]):
        route_ids = [config.route_id for config in configs]

        if len(route_ids) != len(set(route_ids)):
            raise ValueError("model name and version must be unique")

        self.configs = configs
        self.runtimes = {
            config.route_id: ModelRuntime(config)
            for config in configs
        }

        self._cursor = 0
        self._route_lock = threading.Lock()

    @classmethod
    def from_file(cls, path: str) -> "ModelRegistry":
        with open(path, "r", encoding="utf-8") as file:
            data = json.load(file)

        configs = [
            ModelConfig.model_validate(item)
            for item in data["models"]
        ]

        return cls(configs)

    def choose(self, requested_model: str | None) -> ModelConfig:
        enabled_models = [
            config for config in self.configs
            if config.enabled
        ]

        if not enabled_models:
            raise RuntimeError("no enabled model is available")

        if requested_model:
            exact_model = next(
                (
                    config for config in enabled_models
                    if config.route_id == requested_model
                    or requested_model in config.aliases
                ),
                None
            )

            if exact_model:
                return exact_model

            same_name_models = [
                config for config in enabled_models
                if config.name == requested_model
            ]

            if not same_name_models:
                raise ValueError(
                    f"model not found: {requested_model}"
                )

            candidates = same_name_models
        else:
            candidates = enabled_models

        # 加权轮询：weight=3 的模型大约获得 3 倍请求量
        weighted_candidates = []

        for config in candidates:
            weighted_candidates.extend(
                [config] * config.weight
            )

        with self._route_lock:
            selected = weighted_candidates[
                self._cursor % len(weighted_candidates)
            ]
            self._cursor += 1

        return selected

    def list_models(self) -> list[dict]:
        result = []

        for config in self.configs:
            result.append({
                "id": config.route_id,
                "name": config.name,
                "version": config.version,
                "enabled": config.enabled,
                "weight": config.weight,
                "loaded": self.runtimes[
                    config.route_id
                ].loaded
            })

        return result


CONFIG_PATH = os.getenv(
    "MODEL_CONFIG_PATH",
    "models.json"
)

registry = ModelRegistry.from_file(CONFIG_PATH)

app = FastAPI(
    title="LLM Service",
    version="1.0.0"
)


@app.get("/healthz")
def healthz():
    return {
        "status": "ok",
        "device": str(
            "cuda" if torch.cuda.is_available() else "cpu"
        )
    }


@app.get("/v1/models")
def list_models():
    return {
        "object": "list",
        "data": registry.list_models()
    }


@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest):
    try:
        selected_config = registry.choose(request.model)
    except ValueError as exc:
        raise HTTPException(
            status_code=404,
            detail=str(exc)
        ) from exc
    except RuntimeError as exc:
        raise HTTPException(
            status_code=503,
            detail=str(exc)
        ) from exc

    runtime = registry.runtimes[
        selected_config.route_id
    ]

    messages = [
        message.model_dump()
        for message in request.messages
    ]

    try:
        answer, prompt_tokens, completion_tokens = (
            await asyncio.to_thread(
                runtime.generate,
                messages,
                request.temperature,
                request.top_p,
                request.max_tokens
            )
        )
    except Exception as exc:
        raise HTTPException(
            status_code=500,
            detail=f"inference failed: {exc}"
        ) from exc

    return {
        "id": f"chatcmpl-{uuid.uuid4().hex}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": selected_config.route_id,
        "choices": [
            {
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": answer
                },
                "finish_reason": (
                    "length"
                    if completion_tokens >= request.max_tokens
                    else "stop"
                )
            }
        ],
        "usage": {
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": (
                prompt_tokens + completion_tokens
            )
        }
    }

六、启动服务

直接启动：

uvicorn app.main:app --host 0.0.0.0 --port 8000

启动成功后，可以访问接口文档：

http://127.0.0.1:8000/docs

检查服务状态：

curl http://127.0.0.1:8000/healthz

返回结果：

{
  "status": "ok",
  "device": "cpu"
}

七、查看模型列表

curl http://127.0.0.1:8000/v1/models

返回结果类似：

{
  "object": "list",
  "data": [
    {
      "id": "chat-small@1.0.0",
      "name": "chat-small",
      "version": "1.0.0",
      "enabled": true,
      "weight": 3,
      "loaded": false
    },
    {
      "id": "chat-small@1.1.0",
      "name": "chat-small",
      "version": "1.1.0",
      "enabled": true,
      "weight": 1,
      "loaded": false
    }
  ]
}

loaded=false 表示模型尚未加载。第一次调用该模型时，服务才会从 HuggingFace 下载并加载模型。

八、调用聊天接口

1. 自动进行多模型调度

curl -X POST http://127.0.0.1:8000/v1/chat/completions ^
  -H "Content-Type: application/json" ^
  -d "{\"model\":\"chat-small\",\"messages\":[{\"role\":\"user\",\"content\":\"请解释什么是 RAG\"}],\"temperature\":0.7,\"max_tokens\":128}"

当 model 设置为 chat-small 时，系统会在该名称下的多个版本之间进行加权轮询。

2. 固定调用指定版本

curl -X POST http://127.0.0.1:8000/v1/chat/completions ^
  -H "Content-Type: application/json" ^
  -d "{\"model\":\"chat-small@1.0.0\",\"messages\":[{\"role\":\"user\",\"content\":\"什么是 LoRA 微调？\"}],\"max_tokens\":128}"

也可以使用别名：

curl -X POST http://127.0.0.1:8000/v1/chat/completions ^
  -H "Content-Type: application/json" ^
  -d "{\"model\":\"stable\",\"messages\":[{\"role\":\"user\",\"content\":\"解释 Transformer 的注意力机制\"}]}"

九、Docker 镜像构建

创建 Dockerfile：

FROM python:3.11-slim

ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
ENV HF_HOME=/models/cache

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY app ./app
COPY models.json ./models.json

EXPOSE 8000

CMD [
  "uvicorn",
  "app.main:app",
  "--host",
  "0.0.0.0",
  "--port",
  "8000",
  "--workers",
  "1"
]

构建镜像：

docker build -t llm-service:1.0.0 .

启动容器：

docker run --rm -p 8000:8000 ^
  -v llm-hf-cache:/models/cache ^
  llm-service:1.0.0

这里通过 Docker Volume 持久化模型缓存，避免容器每次启动时重复下载模型。

十、使用 Docker Compose 部署

创建 docker-compose.yml：

services:
  llm-api:
    build: .
    image: llm-service:1.0.0
    container_name: llm-api
    ports:
      - "8000:8000"
    environment:
      MODEL_CONFIG_PATH: /app/models.json
      HF_HOME: /models/cache
    volumes:
      - hf-cache:/models/cache
      - ./models.json:/app/models.json:ro
    restart: unless-stopped

volumes:
  hf-cache:

启动：

docker compose up --build

后台启动：

docker compose up -d --build

查看日志：

docker compose logs -f llm-api

十一、多模型调度原理

本文使用加权轮询算法：

weighted_candidates = []

for config in candidates:
    weighted_candidates.extend(
        [config] * config.weight
    )

假设模型配置如下：

chat-small@1.0.0  weight=3
chat-small@1.1.0  weight=1

调度序列大致为：

1.0.0
1.0.0
1.0.0
1.1.0
1.0.0
1.0.0
1.0.0
1.1.0

这种方式适合：

• 新版本灰度发布
• 多个模型版本分流
• A/B 测试
• 不同业务流量分配

如果需要更复杂的调度，可以增加以下指标：

模型平均响应时间
模型当前并发数
GPU 显存占用
模型错误率
业务优先级
用户租户信息

例如，可以优先把请求分配给当前并发较少的模型：

def choose_by_load(model_runtimes):
    return min(
        model_runtimes,
        key=lambda runtime: runtime.active_requests
    )

十二、模型版本管理

模型版本的核心是为每一个模型建立唯一标识：

模型名称@版本号

例如：

chat-small@1.0.0
chat-small@1.1.0
chat-large@2.0.0

推荐使用语义化版本号：

主版本号.次版本号.修订号

版本变更建议：

变更类型	版本变化
训练数据修正	1.0.0 -> 1.0.1
Prompt 或推理参数变化	1.0.0 -> 1.1.0
模型结构或能力明显变化	1.0.0 -> 2.0.0

灰度发布时，可以修改权重：

{
  "name": "chat-small",
  "version": "1.0.0",
  "weight": 9,
  "enabled": true
}

{
  "name": "chat-small",
  "version": "1.1.0",
  "weight": 1,
  "enabled": true
}

此时新版本只接收约 10% 的请求。观察一段时间后，可以逐步提高权重。

十三、生产环境注意事项

1. 不建议在一个进程加载过多模型

多个模型同时加载会消耗大量内存和显存。生产环境更推荐：

网关服务
  |
  +--> 模型服务 A
  |
  +--> 模型服务 B
  |
  +--> 模型服务 C

每个模型独立运行在一个容器中，由网关负责调度。

2. 使用 workers 时要注意显存复制

不建议直接配置：

uvicorn app.main:app --workers 4

因为每个 worker 进程都可能加载一份模型，导致显存快速耗尽。大模型服务通常采用：

单进程 + 异步请求

或者使用 vLLM、TGI 等专门的推理框架。

3. 增加 API 鉴权

可以通过请求头增加简单的 API Key：

from fastapi import Header


@app.post("/v1/chat/completions")
async def chat_completions(
    request: ChatRequest,
    authorization: str | None = Header(default=None)
):
    expected_key = os.getenv("API_KEY")

    if expected_key:
        if authorization != f"Bearer {expected_key}":
            raise HTTPException(
                status_code=401,
                detail="invalid api key"
            )

    # 后续执行模型推理

启动时设置：

docker run --rm -p 8000:8000 ^
  -e API_KEY=change-me ^
  llm-service:1.0.0

4. 增加请求超时和限流

大模型推理属于长耗时任务，应当设置：

单请求最大输入长度
单请求最大输出长度
单用户并发数
整体并发上限
请求超时时间

5. GPU 环境建议使用专业推理框架

本文使用 Transformers 主要用于演示服务化流程。生产环境可以替换为：

• vLLM：高吞吐、支持连续批处理
• Text Generation Inference：适合 HuggingFace 生态
• Ollama：适合本地快速部署
• Triton Inference Server：适合统一推理服务管理

十四、总结

本文使用 FastAPI 封装了一个基础的大模型推理服务，并实现了：

• 统一聊天接口
• OpenAI 风格响应格式
• Docker 容器化部署
• 模型懒加载
• 多模型加权调度
• 模型版本固定调用
• 模型别名管理
• HuggingFace 模型缓存持久化
• 基础健康检查和模型查询

在实际项目中，还可以继续扩展流式输出、Redis 限流、Prometheus 监控、日志追踪、请求队列和 GPU 负载均衡等能力，逐步演进为完整的大模型服务平台。