智能客服问答系统API开发实战：从架构设计到性能优化

---

背景痛点：客服系统最怕的三件事

去年“618”大促，我们内部客服网关直接被打到 502，总结下来就是三句话：

1. 突发流量一来，模型推理排队，P99 响应从 300 ms 飙到 3 s，用户直接开骂。
2. 多轮对话里，用户中途改口“不对，我要退款”，状态机却还在上一个“开发票”分支，答非所问。
3. 意图识别模型对“我要退”和“我要退货”置信度只差 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，防止旧答案赖着不走

避坑指南：三个深夜踩过的坑

1. 对话上下文存储的序列化陷阱
   最早用pickle存history，结果模型迭代新增字段后反序列化失败。改为pydantic.BaseModel+json后，向前兼容只需加默认值。

2. 异步 IO 协程泄漏
   压测时发现内存随 QPS 线性上涨，用asyncio.all_tasks()打印发现大量pending任务，定位到是httpx.AsyncClient没复用。加单例后内存平稳。

3. 模型热更新版本兼容
   采用“双目录 + 软链”方案：

   • 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

---