LangChain智能客服实战：如何基于直属库构建高效问答系统
背景痛点：传统客服的“慢”与“乱”
去年我在一家 SaaS 公司做客服中台，高峰期每天 3 万条工单。老系统用 MySQL 全文索引 + 正则匹配，平均响应 4.8 秒，命中率只有 42%。痛点集中在这几点：
- 关键词检索只能做字面匹配，用户口语化提问直接翻车
- 知识库 18 万条 FAQ，每次 like '%xxx%' 把 CPU 打满
- 大模型直接回答，幻觉严重，还慢——一次 GPT-4 调用 8 秒，用户早走了
老板一句话：能不能“秒回”且“不说错”？于是我们把目光投向 LangChain + 直属库（公司内部的 PostgreSQL 集群，延迟 < 5 ms）的 RAG 方案。
技术选型：直接 LLM vs RAG
| 维度 | 直接调用大模型 | RAG + 直属库 |
|---|---|---|
| 延迟 | 2–8 s | 200–600 ms |
| 幻觉 | 高 | 低（可控） |
| 数据新鲜度 | 训练截止日 | 实时同步 |
| 成本 | 按 1k tokens 计费 | 自建向量索引，几乎 0 增量 |
| 可解释 | 黑盒 | 返回出处，可审计 |
结论：客服场景要“快、准、省”，RAG 完胜。
核心实现：三步搞定“直属库 + LangChain”
1. 数据预处理：把 18 万条 FAQ 变成“向量块”
先写个并行清洗脚本，把数据库里 title、content 两列捞出来，清洗掉 HTML、连续换行，再按 512 token 滑动窗口切分，overlap 10% 防止截断语义。
# etl/chunk_worker.py import re, os, json from sqlalchemy import create_engine from langchain.text_splitter import RecursiveCharacterTextSplitter DSN = "postgresql+psycopg2://user:pwd@直属库:5432/crm" engine = create_engine(D，SN, pool_pre_ping=True) def clean(txt): txt = re.sub(r'<.*?>', '', txt) # 去 HTML txt = re.sub(r'\s+', ' ', txt) return txt.strip() def yield_rows(): sql = "SELECT id, title, content FROM faq WHERE status='ONLINE'" for chunk in pd.read_sql(sql, engine, chunksize=5000): for _, row in chunk.iterrows(): text = f"{row['title']}\n{row['content']}" yield row['id'], clean(text) def build_chunks(): splitter = RecursiveCharacterTextSplitter( chunk_size=512, chunk_overlap=51, length_function=len, separators=["\n", "。", "；"] ) for pk, text in yield_rows(): for idx, sub in enumerate(splitter.split_text(text)): yield {"faq_id": pk, "chunk_id": idx, "text": sub} if __name__ == "__main__": with open("chunks.jsonl", "w", encoding="utf8") as f: for c in build_chunks(): f.write(json.dumps(c, ensure_ascii=False) + "\n")用 GNU Parallel 把 18 万条拆 20 进程跑，10 分钟搞定。
2. 向量化与索引：直属库 pgvector 一步到位
直属库已装 pgvector 插件，直接建表：
CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE faq_emb ( id SERIAL PRIMARY KEY, faq_id INT, chunk_id INT, text TEXT, emb VECTOR(1536) );Python 灌库：
# emb/insert.py import openai, json, psycopg2 from pgvector.psycopg2 import register_vector openai.api_key = os.getenv("OPENAI_KEY") conn = psycopg2.connect(dbname="crm", user="user", password="pwd", host="直属库") register_vector(conn) def get_embedding(text): resp = openai.Embedding.create(input=text, model="text-embedding-ada-002") return resp['data'][0]['embedding'] cur = conn.cursor() with open("chunks.jsonl", encoding="utf8") as f: for line in f: c = json.loads(line) emb = get_embedding(c["text"]) cur.execute( "INSERT INTO faq_emb(faq_id,chunk_id,text,emb) VALUES %s", [(c["faq_id"], c["chunk_id"], c["text"], emb)] ) conn.commit()建 IVFFlat 索引加速：
CREATE INDEX ON faq_emb USING ivfflat (emb vector_cosine_ops) WITH (lists = 100);3. LangChain RetrievalQA：200 毫秒完成“检索 + 生成”
# bot/chain.py from langchain.vectorstores.pgvector import PGVector from langchain.chat_models import ChatOpenAI from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate CONNECTION_STRING = "postgresql+psycopg2://user:pwd@直属库:5432/crm" COLLECTION = "faq_emb" vectorstore = PGVector( connection_string=CONNECTION_STRING, collection_name=COLLECTION, embedding_function=OpenAIEmbeddings(model="text-embedding-ada-002") ) prompt = PromptTemplate( input_variables=["context", "question"], template=""" 你是公司客服机器人，只能使用以下上下文回答问题，禁止编造： {context} 问题：{question} """ ) qa = RetrievalQA.from_chain_type( llm=ChatOpenAI(model="gpt-3.5-turbo", temperature=0), chain_type="stuff", retriever=vectorstore.as_retriever(search_kwargs={"k": 3}), chain_type_kwargs={"prompt": prompt}, return_source_documents=True )FastAPI 包一层：
# bot/api.py from fastapi import FastAPI app = FastAPI() @app.post("/ask") def ask(q: str): ans = qa({"query": q}) return {"answer": ans["result"], "source": ans["source_documents"]}压测结果：P99 580 ms，命中率 87%，老板终于笑了。
性能优化：让“秒回”更稳
1. 并行建索引
上面 ETL 脚本开 20 进程，IO 打满但 CPU 还有余量，就把 embedding 请求改成异步：
import asyncio, aiohttp, openai async def embed_many(texts): tasks = [openai.Embedding.acreate(input=t, model="ada-002") for t in texts] return await asyncio.gather(*tasks)一次批量 100 条，QPS 提升 6 倍。
2. 缓存热问
客服 80% 问题集中在 200 个高频 FAQ。用 Redis 把“向量哈希 → 答案”缓存 5 分钟，命中后 20 ms 返回。哈希取法：
import hashlib def qhash(q): return hashlib.blake2b(q.encode('utf8'), digest_size=8).hexdigest()3. 超时重试
OpenAI 接口偶发 429，用 tenacity 装饰器：
from tenacity import retry, stop_after_attempt, wait_random_exponential @retry(wait=wait_random_exponential(min=1, max=20), stop=stop_after_attempt(5)) def get_embedding(text): ...避坑指南：中文场景的小地雷
分词器选错，句子被拦腰斩断
用RecursiveCharacterTextSplitter时把 separators 放中文标点：“\n。；，” 实测召回 +5%。向量相似度阈值
cosine < 0.78 基本胡言乱语，> 0.82 又太保守。用验证集画 PR 曲线，选 0.8 最平衡。对话状态管理
多轮场景要保留上文。把历史问答拼成“伪文档”再检索，否则用户追问“那怎么办”会断片。LangChain 的ConversationalRetrievalQA可接盘。
生产建议：监控 + 容灾
| 指标 | 采集方式 | 告警阈值 |
|---|---|---|
| P99 延迟 | FastAPI middleware | > 1 s |
| 检索命中率 | 日志统计 | < 80% |
| 幻觉率 | 随机 100 条人工抽检 | > 5% |
| 向量索引延迟 | pg_stat_statements | > 100 ms |
容灾：直属库做主从 + 延迟从，向量表每日逻辑备份到对象存储；LLM 侧配置 fallback 到 Azure OpenAI，DNS 秒级切换。
进阶思考
- 查询改写：先用 LLM 把口语问句改写成关键词组合，再向量检索，能否再提 10% 命中率？
- 分级召回：先走倒排索引粗排 1k 候选，再走向量精排 10，延迟能否压到 200 ms 以内？
- 私有化 embedding：用中文 BGE-large-en-v1.5 替代 OpenAI，成本归零，效果会不会掉？
把这三点跑通，也许就能从“秒回”进化到“毫秒回”了。祝你玩得开心，有问题评论区一起踩坑。
