基于基础模型的AI工程实践：构建高可用在线阅读应用架构

1次阅读

没有评论

共计 2288 个字符，预计需要花费 6 分钟才能阅读完成。

在构建在线阅读应用时，我们遇到了几个关键技术挑战：

长文本 embedding 的延迟问题：传统模型处理整本电子书章节时，推理时间可能达到秒级，严重影响用户体验。
用户个性化推荐的计算开销：为每个用户实时计算相似书籍推荐，对 GPU 资源消耗极大。
突发流量的自动扩缩容：热门书籍发布时，流量可能瞬间增长 10 倍以上。

我们对比了两种主流模型：

BERT-base：擅长长文本理解，但 embedding 速度较慢（约 300ms/ 千字）
DistilBERT：体积小 40%，速度提升 2 倍，精度下降约 3%
MiniLM：专为 embedding 优化的蒸馏模型，速度最快但需要微调

最终选择 MiniLM-v2 作为基础模型，其性能如下：

# 性能对比示例
models = {'bert-base': {'speed': '300ms', 'size': '420MB'},
    'distilbert': {'speed': '150ms', 'size': '250MB'},
    'minilm-v2': {'speed': '80ms', 'size': '180MB'}
}

使用 FastAPI 构建服务，关键代码如下：

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/embedding")
async def get_embedding(text: str, token: str = Depends(oauth2_scheme)):
    """
    生成文本 embedding 向量
    :param text: 输入文本（最大支持 10 万字）:param token: JWT 认证令牌
    :return: 512 维 embedding 向量
    """
    if len(text) > 100000:
        raise HTTPException(status_code=400, detail="Text too long")
    # 实际推理代码省略
    return {"embedding": [...]}

Redis 缓存实现要点：

使用 HSET 存储 embedding 结果，key 为文本 MD5 值
设置 TTL 为 24 小时应对数据更新
采用双重检查锁防止缓存击穿

import redis
import hashlib

r = redis.Redis(host='redis', port=6379)

def cached_embedding(text):
    text_hash = hashlib.md5(text.encode()).hexdigest()
    # 先查缓存
    if emb := r.hget("embeddings", text_hash):
        return emb

    # 缓存未命中，获取锁
    with r.lock(f"lock:{text_hash}", timeout=5):
        # 再次检查防止并发时重复计算
        if emb := r.hget("embeddings", text_hash):
            return emb
        # 实际计算并写入缓存
        emb = model.inference(text)
        r.hset("embeddings", text_hash, emb)
        r.expire(text_hash, 86400)
        return emb

使用 Locust 模拟的测试数据：

单节点 QPS：从 120 提升到 310（优化后）
P99 延迟：从 450ms 降至 190ms
资源消耗：GPU 利用率稳定在 70-80%

精度	显存占用	推理速度	准确率
FP32	4.2GB	80ms	92.3%
FP16	2.1GB	65ms	92.1%
INT8	1.2GB	45ms	90.8%

最终选择 FP16 作为折中方案。

Prometheus 关键指标：

metrics:
  - name: "model_latency_seconds"
    help: "Model inference latency in seconds"
    type: histogram
    labels: ["model_version"]
  - name: "cache_hit_rate"
    help: "Embedding cache hit rate"
    type: gauge

CI/CD 流程包含：

自动化测试：对比新老模型的准确率差异
灰度发布：先对 5% 流量启用新模型
回滚机制：出现异常时 30 秒内自动切换

当 GPU 内存不足时：

自动切换到 CPU 模式
对长文本进行分段处理
返回低维 embedding（从 512 降到 256 维）

在推理前增加 hook：

def safety_check(text):
    """检查敏感内容"""
    if detect_sensitive_words(text):
        raise ContentSafetyError

@app.post("/embedding")
async def get_embedding(text: str):
    safety_check(text)  # 前置检查
    # ... 原有逻辑

实践中我们仍在探索：