背景痛点：客服系统最怕的三件事
去年“618”大促，我们内部客服网关直接被打到 502，总结下来就是三句话：
- 突发流量一来，模型推理排队，P99 响应从 300 ms 飙到 3 s，用户直接开骂。
- 多轮对话里，用户中途改口“不对，我要退款”，状态机却还在上一个“开发票”分支，答非所问。
- 意图识别模型对“我要退”和“我要退货”置信度只差 0.02，结果把退货申请开成了退款申请，人工客服连夜擦屁股。
痛定思痛，我们决定把问答系统拆成独立 API，用 AI 辅助开发的方式把这三颗雷一次性排掉。
技术选型：BERT 还是 LLM？——让数据说话
| 维度 | BERT+Fine-tune | LLM Few-shot |
|---|---|---|
| 训练数据 | 5 k 条人工标注 | 0 条，纯 prompt |
| 平均意图准确率 | 94.7 % | 91.2 % |
| 推理耗时（T4） | 38 ms | 520 ms |
| 线上显存占用 | 1.1 GB | 12 GB |
| 版本漂移风险 | 低，参数固定 | 高，随 prompt 变化 |
结论：在“高频、封闭域”的智能客服场景，Transformer Encoder + 轻量 Fine-tuning性价比最高；LLM 留给“创意写作”更合适。
核心实现：FastAPI 异步骨架
1. 项目结构
bot_api/ ├── main.py ├── auth.py ├── model.py ├── cache.py └── logs/2. JWT 鉴权工具（auth.py）
from datetime import datetime, timedelta from typing import Optional import jwt from fastapi import HTTPException, Security from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer SECRET = "ReplaceMeInProd" ALG = "HS256" EXP_MIN = 30 def create_token(sub: str) -> str: expire = datetime.utcnow() + timedelta(minutes=EXP_MIN) return jwt.encode({"exp": expire, "sub": sub}, SECRET, algorithm=ALG) def decode_token(token: str) -> str: try: payload = jwt.decode(token, SECRET, algorithms=[ALG]) return payload["sub"] except jwt.ExpiredSignatureError: raise HTTPException(status_code=401, detail="Token expired")3. 意图分类模型封装（model.py）
from typing import Tuple from torch import cuda from transformers import AutoTokenizer, AutoModelForSequenceClassification, pipeline device = 0 if cuda.is_available() else -1 MODEL_ID = "bert-base-chinese" tokenizer = AutoTokenizer.from_pretrained(MODEL_ID) model = AutoModelForSequenceClassification.from_pretrained("./finetuned") cls = pipeline("text-classification", model=model, tokenizer=tokenizer, device=device) def predict_intent(text: str) -> Tuple[str, float]: """返回意图标签及置信度""" res = cls(text, truncation=True, max_length=128, top_k=1)[0] return res["label"], res["score"]4. 带缓存与日志的完整路由（main.py）
import time import logging from typing import Dict from fastapi import FastAPI, Depends, Request from pydantic import BaseModel, Field from auth import decode_token, HTTPBearer from model import predict_intent from cache import cache_get, cache_set app = FastAPIAPI(title="SmartQA-API", version="1.2.0") security = HTTPBearer() logger = logging.getLogger("api") class QARequest(BaseModel): uid: str = Field(..., description="用户唯一标识") query: str = Field(..., description="用户问题") history: Dict[str, str] = Field(default_factory=dict, description="多轮上下文") class QAResponse(BaseModel): intent: str confidence: float answer: str cached: bool @app.post("/qa", response_model=QAResponse) async def qa(req: QARequest, _: str = Depends(decode_token)): start = time.time() key = f"cache:{req.query}" hit = cache_get(key) if hit: logger.info("cache_hit key=%s", key) return QAResponse(**hit, cached=True) intent, score = predict_intent(req.query) answer = select_answer(intent, req.history) # 业务规则函数，略 body = {"intent": intent, "confidence": score, "answer": answer} cache_set(key body, ex=300) logger.info("predict cost=%.2fms query=%s intent=%s", (time.time()-start)*1000, req.query, intent) return QAResponse(**body, cached=False)说明：
- 所有 I/O 均用
async/await，保证 FastAPI 的协程池不被阻塞select_answer内部可继续访问知识库或图谱，这里不铺开展示
性能优化：缓存让 QPS 翻 4 倍
1. 压测脚本（locustfile.py 片段）
from locust import HttpUser, task class QAUser(HttpUser): @task def ask(self): self.client.post("/qa", json={"uid":"u1", "query":"如何退货"}, headers={"Authorization":"Bearer xxx"})2. 结果对比（4 核 8 G，T4 GPU）
| 场景 | 平均 QPS | 95% 延迟 | GPU 利用率 |
|---|---|---|---|
| 无缓存 | 42 | 880 ms | 98 % |
| Redis 缓存 | 178 | 120 ms | 23 % |
缓存命中率 72 %，直接把 GPU 从火葬场救回养老院。
3. 缓存设计模式
- Key：
cache:{query} - Value：JSON 序列化后的
QAResponse（不含cached字段，避免循环） - 过期策略：TTL 5 min + LRU 驱逐
- 一致性：模型热更新时统一发
FLUSHDB，防止旧答案赖着不走
避坑指南：三个深夜踩过的坑
对话上下文存储的序列化陷阱
最早用pickle存history，结果模型迭代新增字段后反序列化失败。改为pydantic.BaseModel+json后，向前兼容只需加默认值。异步 IO 协程泄漏
压测时发现内存随 QPS 线性上涨，用asyncio.all_tasks()打印发现大量pending任务，定位到是httpx.AsyncClient没复用。加单例后内存平稳。模型热更新版本兼容
采用“双目录 + 软链”方案：models/v1.0.0/models/v1.1.0/
发布时把models/current指向新目录，重启 Worker；回滚只需改链。配合__version__打到日志，灰度可追踪。
代码规范小结
- 统一
black格式化，行宽 88 - 函数必须写
Google Styledocstring，参数类型用typing标注 - 单元测试覆盖 ≥ 80 %，CI 用
pytest-cov强制卡口 - 日志用
structlog，方便后续接入 Loki
部署 Tips：Docker-Compose 一键起
version: "3.9" services: api: build: . env_file: .env depends_on: - redis redis: image: redis:7-alpine command: redis-server --maxmemory 1gb --maxmemory-policy allkeys-lru把uvicorn main:app --workers 4塞进ENTRYPOINT，单卡 4 进程 GPU 利用率刚好 90 %，留 10 % 给突发。
互动时间
如何设计跨语言问答的语料对齐方案？
当同一商品在中文、英文站点分别维护 FAQ，人工翻译往往滞后，导致用户问“return policy”命中英文、问“退货”命中中文，答案不同步。你有批量对齐的好办法吗？欢迎留言交流！
延伸阅读
- 《Speech-to-Intent: 端到端口语意图识别》
- HuggingFace 官方博客：Accelerating BERT Inference with TensorRT
- Redis 官方文档：Eviction Policies
