当前位置：首页 > news >正文

扣子的知识库智能客服：从架构设计到生产环境部署的实战指南

news 2026/5/12 12:33:42

1. 传统客服系统的三大顽疾

过去两年做 ToB SaaS 时，我接触过不少“祖传”客服后台：

全文检索靠 MySQL LIKE，平均响应 800 ms，并发一高就雪崩
FAQ 更新靠运营手动贴 Excel，上线流程 2 天，客户早把电话打爆
多轮对话用 if-else 写死 3 000 行，需求一改，开发得重新读“文言文”

这些痛点归结起来就是：延迟高、更新慢、逻辑乱。扣子团队决定用“知识库智能客服”重新打地基，目标很明确——P99 响应 <200 ms、知识 5 分钟内生效、意图识别准确率 ≥90%。

2. 技术选型：规则、ML 与深度学习的三角权衡

方案	优点	缺点	适用场景
规则引擎（ES、正则）	可控、可解释	泛化差、维护地狱	冷启动、灰度兜底
传统 ML（TF-IDF+SVM）	训练快、CPU 友好	语义鸿沟、特征工程重	数据量 <10 万
深度学习（Sentence-BERT+向量检索）	泛化强、支持多语言	需要 GPU、调参复杂	数据量 >10 万、长期演进

扣子最终采用“规则兜底 + 深度语义”的混合架构：

新意图先用规则快速上线
累计 >200 条语料后自动触发 BERT 微调
向量索引每日增量更新，旧规则逐步下线

3. 系统总览：一张图看懂微服务拓扑

（文字描述，方便手绘）

┌──────┐ �URL ┌──────────┐ 用户──►│ 网关 │──►/chat──►│ 对话服务 │ FastAPI └──┬───┘ └──┬──────┘ │ │ │gRPC │HTTP ▼ ▼ ┌──────────┐ ┌──────────┐ │意图模型 │ │知识检索 │ Milvus │ TF-Serving │ │ 服务 │ └──────────┘ └──────────┘ ▲ ▲ │Kafka │Kafka └───────┐ ┌───────┘ ▼ ▼ ┌──────────────┐ │ 知识运营后台 │ MySQL+MinIO └──────────────┘

4. 核心实现拆解

4.1 向量化知识库：把 FAQ 变成向量

采用sentence-transformers/all-MiniLM-L6-v2，维度 384，平衡精度与速度
文本先过段落拆分（按 512 token 滑动窗口），再批量编码
写入 Milvus 2.3，索引类型 IVF_AUTO_INDEX，召回 Top-10，阈值 0.75

4.2 多轮对话状态机

用有限状态机（FSM）记录每轮槽位：

状态 = 意图 + 已填充槽位
转移条件 = 用户新意图或缺失槽位
状态持久化在 Redis Hash，TTL 600 s，支持并发扩容

4.3 微服务骨架：FastAPI + Uvicorn + Gunicorn

单容器 1 核 2 G，4 Worker，异步接口平均 QPS 提升 3 倍
通过pydantic.BaseSettings管理配置，12-Factor 合规
自研ai-common包统一日志、链路追踪（OpenTelemetry）

5. 关键代码片段

以下示例均基于 Python 3.11，已脱敏。

5.1 知识库向量化脚本（offline_job.py）

# -*- coding: utf-8 -*- """ 离线作业：把 MySQL FAQ 表全量编码并写入 Milvus """ import os import json from typing import List import pymysql from sentence_transformers import SentenceTransformer from pymilvus import Collection, FieldSchema, CollectionSchema, DataType, connections DB_CFG = dict(host=os.getenv("DB_HOST", "127.0.0.1"), user=os.getenv("DB_USER", "root"), password=os.getenv("DB_PWD"), database="knowledge", charset="utf8mb4") MODEL = SentenceTransformer("all-MiniLM-L6-v2") BATCH_SIZE = 256 # 根据 GPU 显存调整 COLLECTION_NAME = "faq_v1" def fetch_faq() -> List[dict]: """拉取全量 FAQ""" sql = "SELECT id, question, answer FROM faq WHERE status=1" with pymysql.connect(**DB_CFG).cursor(pymysql.cursors.DictCursor) as cur: cur.execute(sql) return cur.fetchall() def create_collection(): """首次运行时建立 Collection""" fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True/True), Field(name="question", dtype=DataType.VARCHAR, max_length=1024), Field(name="answer", dtype=DataType.VARCHAR, max_length=2048), Field(name="vector", dtype=DataType.FLOAT_VECTOR, dim=384) ] schema = CollectionSchema(fields, description="FAQ 向量库") collection = Collection(name=COLLECTION_NAME, schema=schema) # IVF 索引，加速内积搜索 index_params = {"metric_type": "IP", "index_type": "IVF_FLAT", "params": {"nlist": 128}} collection.create_index("vector", index_params) return collection def main(): connections.connect(alias="default", host=os.getenv("MILVUS_HOST", "127.0.0.1"), port="19530") collection = create_collection() faq_rows = fetch_faq() for i in range(0, len(faq_rows), BART_SIZE): batch = faq_rows[i:i+BART_SIZE] questions = [b["question"] for b in batch] vectors = MODEL.encode(questions, normalize_embeddings=True).tolist() ids = [b["id"] for b in batch] questions_ans = [(b["question"], b["answer"]) for b in batch] data = [ids, [q for q, _ in questions_ans], [a for _, a in questions_ans], vectors] collection.insert(data) collection.flush() collection.load() print(f"Inserted {len(faq_rows)} records into Milvus.") if __name__ == "__main__": main()

5.2 语义检索服务（search_service.py）

# 核心逻辑：向量召回 + 精排 from typing import List, Tuple from pymilvus import Collection, connections from sentence_transformers import SentenceTransformer MODEL = SentenceTransformer("all-MiniLM-L6-v2") TOPK = 10 THRESHOLD = 0.75 class FaqSearchService: def __init__(self, collection_name: str = "faq_v1"): connections.connect(alias="search", host="milvus-svc", port="19530") self.collection = Collection(name=collection_name) self.collection.load() def search(self, query: str) -> List[Tuple[str, str, float]]: vec = MODEL.encode(query, normalize_embeddings=True).tolist() results = self.collection.search( data=[vec], anns_field="vector", param={"metric_type": "IP", "params": {"nprobe": 16}}, limit=TOPK, output_fields=["question", "answer"] ) hits = [] for hits_per_query in results: for h in hits_per_query: if h.score < THRESHOLD: continue hits.append((h.entity.question, h.entity.answer, h.score)) return hits