模型服务Skill model-serving

---

名称: 模型服务 描述: LLM 和 ML 模型部署用于推理。在生产中服务模型、构建AI API 或优化推理时使用。覆盖 vLLM (LLM 服务)、TensorRT-LLM (GPU 优化)、Ollama (本地)、BentoML (ML 部署)、Triton (多模型)、LangChain (编排)、LlamaIndex (RAG) 和流式模式。

模型服务

目的

部署 LLM 和 ML 模型进行生产推理，使用优化的服务引擎、流式响应模式和编排框架。专注于自托管模型服务、GPU 优化和与前端应用程序的集成。

何时使用

• 部署 LLM 进行生产（自托管 Llama、Mistral、Qwen）
• 构建带有流式响应的 AI API
• 服务传统 ML 模型（scikit-learn、XGBoost、PyTorch）
• 实现带向量数据库的 RAG 管道
• 优化推理吞吐量和延迟
• 集成 LLM 服务与前端聊天界面

模型服务选择

LLM 服务引擎

vLLM（推荐主要）

• PagedAttention 内存管理（吞吐量提升 20-30 倍）
• 连续批处理用于动态请求处理
• OpenAI 兼容的 API 端点
• 用途：大多数自托管 LLM 部署

TensorRT-LLM

• 最大 GPU 效率（比 vLLM 快 2-8 倍）
• 需要模型转换和优化
• 用途：需要绝对最大吞吐量的生产工作负载

Ollama

• 无需 GPU 的本地开发
• 简单 CLI 接口
• 用途：原型设计、笔记本电脑开发、教育目的

决策框架：

需要自托管 LLM 部署吗？
├─ 是，需要最大吞吐量 → vLLM
├─ 是，需要绝对最大 GPU 效率 → TensorRT-LLM
├─ 是，仅本地开发 → Ollama
└─ 否，使用托管 API (OpenAI、Anthropic) → 无需服务层

ML 模型服务（非 LLM）

BentoML（推荐）

• Python 原生，易于部署
• 自适应批处理以提高吞吐量
• 多框架支持（scikit-learn、PyTorch、XGBoost）
• 用途：大多数传统 ML 模型部署

Triton 推理服务器

• 在同一 GPU 上服务多模型
• 模型集成（链式多个模型）
• 用途：NVIDIA GPU 优化、服务 10+ 模型

LLM 编排

LangChain

• 通用工作流、代理、RAG
• 100+ 集成（LLM、向量数据库、工具）
• 用途：大多数 RAG 和代理应用

LlamaIndex

• 专注于 RAG，带有高级检索策略
• 100+ 数据连接器（PDF、Notion、Web）
• 用途：RAG 是主要用例

快速开始示例

vLLM 服务器设置

# 安装
pip install vllm

# 服务模型（OpenAI 兼容 API）
vllm serve meta-llama/Llama-3.1-8B-Instruct \
  --dtype auto \
  --max-model-len 4096 \
  --gpu-memory-utilization 0.9 \
  --port 8000

关键参数：

• --dtype: 模型精度（auto、float16、bfloat16）
• --max-model-len: 上下文窗口大小
• --gpu-memory-utilization: GPU 内存分数（0.8-0.95）
• --tensor-parallel-size: 模型并行 GPU 数量

流式响应（SSE 模式）

后端（FastAPI）：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
import json

app = FastAPI()
client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")

@app.post("/chat/stream")
async def chat_stream(message: str):
    async def generate():
        stream = client.chat.completions.create(
            model="meta-llama/Llama-3.1-8B-Instruct",
            messages=[{"role": "user", "content": message}],
            stream=True,
            max_tokens=512
        )

        for chunk in stream:
            if chunk.choices[0].delta.content:
                token = chunk.choices[0].delta.content
                yield f"data: {json.dumps({'token': token})}

"

        yield f"data: {json.dumps({'done': True})}

"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"}
    )

前端（React）：

// 与 ai-chat 技能集成
const sendMessage = async (message: string) => {
  const response = await fetch('/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ message })
  })

  const reader = response.body!.getReader()
  const decoder = new TextDecoder()

  while (true) {
    const { done, value } = await reader.read()
    if (done) break

    const chunk = decoder.decode(value)
    const lines = chunk.split('

')

    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const data = JSON.parse(line.slice(6))
        if (data.token) {
          setResponse(prev => prev + data.token)
        }
      }
    }
  }
}

BentoML 服务

import bentoml
from bentoml.io import JSON
import numpy as np

@bentoml.service(
    resources={"cpu": "2", "memory": "4Gi"},
    traffic={"timeout": 10}
)
class IrisClassifier:
    model_ref = bentoml.models.get("iris_classifier:latest")

    def __init__(self):
        self.model = bentoml.sklearn.load_model(self.model_ref)

    @bentoml.api(batchable=True, max_batch_size=32)
    def classify(self, features: list[dict]) -> list[str]:
        X = np.array([[f['sepal_length'], f['sepal_width'],
                       f['petal_length'], f['petal_width']] for f in features])
        predictions = self.model.predict(X)
        return ['setosa', 'versicolor', 'virginica'][predictions]

LangChain RAG 管道

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Qdrant
from langchain.chains import RetrievalQA
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 加载和分块文档
text_splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)
chunks = text_splitter.split_documents(documents)

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Qdrant.from_documents(
    chunks,
    embeddings,
    url="http://localhost:6333",
    collection_name="docs"
)

# 创建检索链
llm = ChatOpenAI(model="gpt-4o")
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
    return_source_documents=True
)

# 查询
result = qa_chain({"query": "What is PagedAttention?"})

性能优化

GPU 内存估计

LLM 经验法则：

GPU 内存 (GB) = 模型参数 (B) × 精度 (字节) × 1.2

示例：

• Llama-3.1-8B (FP16): 8B × 2 字节 × 1.2 = 19.2 GB
• Llama-3.1-70B (FP16): 70B × 2 字节 × 1.2 = 168 GB（需要 2-4 个 A100）

量化减少内存：

• FP16: 每个参数 2 字节
• INT8: 每个参数 1 字节（内存减少 2 倍）
• INT4: 每个参数 0.5 字节（内存减少 4 倍）

vLLM 优化

# 启用量化（AWQ 4 位）
vllm serve TheBloke/Llama-3.1-8B-AWQ \
  --quantization awq \
  --gpu-memory-utilization 0.9

# 多 GPU 部署（张量并行）
vllm serve meta-llama/Llama-3.1-70B-Instruct \
  --tensor-parallel-size 4 \
  --gpu-memory-utilization 0.9

批处理策略

连续批处理（vLLM 默认）：

• 动态添加/移除请求到批次
• 比静态批处理吞吐量更高
• 无需配置

自适应批处理（BentoML）：

@bentoml.api(
    batchable=True,
    max_batch_size=32,
    max_latency_ms=1000  # 最多等待 1 秒填充批次
)
def predict(self, inputs: list[np.ndarray]) -> list[float]:
    # BentoML 自动批处理请求
    return self.model.predict(np.array(inputs))

生产部署

Kubernetes 部署

查看 examples/k8s-vllm-deployment/ 获取完整 YAML 清单。

关键考虑：

• GPU 资源请求：nvidia.com/gpu: 1
• 健康检查：/health 端点
• 基于队列深度的水平 Pod 自动伸缩
• 持久卷用于模型缓存

API 网关模式

对于生产，添加速率限制、身份验证和监控：

Kong 配置：

services:
  - name: vllm-service
    url: http://vllm-llama-8b:8000
    plugins:
      - name: rate-limiting
        config:
          minute: 60  # 每分钟每个 API 密钥 60 个请求
      - name: key-auth
      - name: prometheus

监控指标

基本 LLM 指标：

• 每秒令牌数（吞吐量）
• 首次令牌时间（TTFT）
• 令牌间延迟
• GPU 利用率和内存
• 队列深度

Prometheus 仪表化：

from prometheus_client import Counter, Histogram

requests_total = Counter('llm_requests_total', '总请求数')
tokens_generated = Counter('llm_tokens_generated', '总令牌数')
request_duration = Histogram('llm_request_duration_seconds', '请求持续时间')

@app.post("/chat")
async def chat(request):
    requests_total.inc()
    start = time.time()
    response = await generate(request)
    tokens_generated.inc(len(response.tokens))
    request_duration.observe(time.time() - start)
    return response

集成模式

前端（ai-chat）集成

这个技能为 ai-chat 技能提供后端服务层。

流程：

前端 (React) → API 网关 → vLLM 服务器 → GPU 推理
     ↑                                                  ↓
     └─────────── SSE 流 (令牌) ─────────────────┘

查看 references/streaming-sse.md 获取完整实现模式。

带向量数据库的 RAG

架构：

用户查询 → LangChain
              ├─> 向量数据库 (Qdrant) 用于检索
              ├─> 组合上下文 + 查询
              └─> LLM (vLLM) 用于生成

查看 references/langchain-orchestration.md 和 examples/langchain-rag-qdrant/ 获取完整模式。

异步推理队列

对于批处理或非实时推理：

客户端 → API → 消息队列 (Celery) → 工作者 (vLLM) → 结果数据库

有用用于：

• 批处理文档处理
• 后台摘要
• 非交互工作流

基准测试

使用 scripts/benchmark_inference.py 测量部署：

python scripts/benchmark_inference.py \
  --endpoint http://localhost:8000/v1/chat/completions \
  --model meta-llama/Llama-3.1-8B-Instruct \
  --concurrency 32 \
  --requests 1000

输出：

• 每秒请求数
• P50/P95/P99 延迟
• 每秒令牌数
• GPU 内存使用

捆绑资源

详细指南：

• references/vllm.md - vLLM 设置、PagedAttention、优化
• references/tgi.md - 文本生成推理模式
• references/bentoml.md - BentoML 部署模式
• references/langchain-orchestration.md - LangChain RAG 和代理
• references/inference-optimization.md - 量化、批处理、GPU 调优

工作示例：

• examples/vllm-serving/ - 完整 vLLM + FastAPI 流式设置
• examples/ollama-local/ - 使用 Ollama 的本地开发
• examples/langchain-agents/ - LangChain 代理模式

实用脚本：

• scripts/benchmark_inference.py - 吞吐量和延迟基准测试
• scripts/validate_model_config.py - 验证部署配置

常见模式

从 OpenAI API 迁移

vLLM 提供 OpenAI 兼容端点，便于迁移：

# 之前 (OpenAI)
from openai import OpenAI
client = OpenAI(api_key="sk-...")

# 之后 (vLLM)
from openai import OpenAI
client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="not-needed"
)

# 相同的 API 调用有效！
response = client.chat.completions.create(
    model="meta-llama/Llama-3.1-8B-Instruct",
    messages=[{"role": "user", "content": "Hello"}]
)

多模型服务

基于任务路由请求到不同模型：

MODEL_ROUTING = {
    "small": "meta-llama/Llama-3.1-8B-Instruct",  # 快速、便宜
    "large": "meta-llama/Llama-3.1-70B-Instruct", # 准确、昂贵
    "code": "codellama/CodeLlama-34b-Instruct"    # 代码特定
}

@app.post("/chat")
async def chat(message: str, task: str = "small"):
    model = MODEL_ROUTING[task]
    # 路由到适当的 vLLM 实例

成本优化

跟踪令牌使用：

import tiktoken

def estimate_cost(text: str, model: str, price_per_1k: float):
    encoding = tiktoken.encoding_for_model(model)
    tokens = len(encoding.encode(text))
    return (tokens / 1000) * price_per_1k

# 比较成本
openai_cost = estimate_cost(text, "gpt-4o", 0.005)  # 每 100 万令牌 $5
self_hosted_cost = 0  # 固定 GPU 成本，无限令牌

故障排除

GPU 内存不足：

• 减少 --max-model-len
• 降低 --gpu-memory-utilization（尝试 0.8）
• 启用量化 (--quantization awq)
• 使用更小模型变体

吞吐量低：

• 增加 --gpu-memory-utilization（尝试 0.95）
• 启用连续批处理（vLLM 默认）
• 检查 GPU 利用率（应 >80%）
• 考虑多 GPU 张量并行

高延迟：

• 如果使用静态批处理，减少批次大小
• 检查到 GPU 服务器的网络延迟
• 使用 scripts/benchmark_inference.py 进行性能分析

后续步骤

1. 本地开发：从 examples/ollama-local/ 开始，进行无需 GPU 的测试
2. 生产设置：使用 examples/vllm-serving/ 部署 vLLM
3. RAG 集成：使用 examples/langchain-rag-qdrant/ 添加向量数据库
4. Kubernetes：使用 examples/k8s-vllm-deployment/ 扩展
5. 监控：使用 Prometheus 和 Grafana 添加指标