最近在做一个智能客服项目客户要求系统能实时回答产品更新、政策变动这类动态问题。传统的基于关键词匹配的规则引擎每次更新知识库都得改规则维护起来太痛苦了。而直接用大语言模型LLM吧它又容易“一本正经地胡说八道”生成一些不在知识库里的内容或者信息过时。为了解决这些问题我们最终选择了RAG检索增强生成架构效果很不错今天就来分享一下从原理到上线的完整过程。1. 为什么是RAG聊聊传统方案的痛点在动手之前我们得先搞清楚为什么要用RAG。传统的客服系统主要有两种路子但各有各的“坑”。基于规则/关键词的引擎这是最老派的做法。比如用户问“怎么退货”系统匹配到“退货”这个关键词就回复预设好的退货流程文本。它的优点是速度快、结果确定。但缺点太明显了知识库一更新规则就得跟着改维护成本极高而且无法理解语义用户换个说法问“商品不想要了怎么办”可能就匹配不上了。纯LLM方案直接把整个对话和问题扔给像GPT-3.5这样的模型。这倒是能理解语义了对话也流畅。但问题在于LLM的知识是“冻结”在训练时的对于产品最新价格、刚发布的公告一无所知容易产生“幻觉”编造信息。另外把公司内部文档全部作为提示词输入也存在成本高、响应慢、可能泄露敏感信息的问题。RAG正好取两者之长避两者之短。它的核心思想是先检索后生成。当用户提问时系统先从最新的、经过审核的知识库比如产品手册、FAQ文档中精准地找到相关的信息片段然后把这些信息片段和用户问题一起作为“参考资料”交给LLM让它基于这些可靠的资料来组织语言生成回答。这样回答既准确源于知识库又自然LLM润色还能随着知识库更新而更新。2. 技术选型搭建积木哪块最合适确定了RAG这条路接下来就是挑选搭建它的“积木”。主要涉及两块向量数据库和模型。向量数据库选型它的任务是存储知识库文本转换成的向量Embedding并快速进行相似度检索。FAISS (Facebook AI Similarity Search)Meta开源的库单机性能强悍特别适合对延迟要求高、数据量在千万级以内的场景。它就像一个本地的高速搜索引擎部署简单但需要自己管理持久化和高可用。我们初期数据量不大且追求极致响应速度所以选择了它。Pinecone / Weaviate / Qdrant这些都是云原生的向量数据库服务。它们提供了开箱即用的持久化、可扩展性和运维管理。如果你的数据量非常大亿级以上或者团队不想操心数据库运维这类托管服务是更好的选择。我们考虑到未来扩展性在架构设计上也预留了切换到这类服务的接口。模型选型这里分两个模型一个是把文本变成向量的嵌入模型一个是生成答案的大语言模型。嵌入模型 (Embedding Model)负责“理解”文本将其转化为数学向量。我们选择了text-embedding-ada-002OpenAI和开源的BGEBAAI/bge-base-zh做对比测试。对于中文场景BGE系列模型表现非常出色且本地部署没有网络延迟和调用成本。最终我们采用了BGE它让我们的检索质量很高。大语言模型 (LLM)负责“组织语言”生成最终答案。我们评估了GPT-3.5-Turbo、GPT-4和开源模型如ChatGLM3、Qwen。考虑到生产环境的响应速度、成本可控性和数据隐私我们选择了在本地部署的Qwen-7B-Chat模型。它在中文理解和生成上效果很好通过量化技术后对GPU资源的要求也在可接受范围内。3. 核心实现三步走打造智能核心整个系统的核心流程可以拆解为三个关键步骤知识处理、检索、生成。知识库构建分块与嵌入你不能把一整本产品手册直接扔进去。需要把它切成有意义的“块”。分块策略我们采用了“递归字符分割”结合“语义分割”的策略。先用固定大小如500字符重叠滑动窗口确保完整性再尝试按段落、标题等自然边界进行二次分割。关键是要保证每个“块”有独立的语义同时避免把一句话或一个关键信息拆散。生成嵌入并存储将分块后的文本通过我们选定的BGE嵌入模型转化为768维的向量。然后将这些向量和对应的原始文本、元数据如来源文档、章节一起存入FAISS索引中。这个过程通常在系统初始化或知识库批量更新时完成。检索与生成的协同机制这是RAG的“大脑”。当用户提问“Q请问旗舰手机X的续航时间是多少”时检索器工作将用户问题也转化为向量然后在FAISS索引中搜索最相似的K个比如top-3知识块。相似度计算通常使用余弦相似度。生成器工作系统会构造一个这样的提示词Prompt给LLM你是一个专业的客服助手。请严格根据以下提供的参考信息来回答问题。如果参考信息中没有答案请直接说“根据现有资料我暂时无法回答这个问题”。 参考信息 1. [知识块1的内容旗舰手机X配备5000mAh电池在典型使用场景下续航可达1.5天...] 2. [知识块2的内容手机X支持65W超级快充30分钟可充电至80%...] 用户问题请问旗舰手机X的续航时间是多少 请生成回答LLM会基于我们提供的“参考信息”来生成答案比如“根据产品信息旗舰手机X配备5000mAh大容量电池在典型使用场景下续航时间可达1.5天。” 这样就确保了答案的准确性和有据可依。流式响应实现为了提升用户体验我们让答案像打字一样一个个词蹦出来而不是等全部生成完再显示。这里用到了FastAPI的StreamingResponse。from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse import asyncio from typing import AsyncGenerator import json app FastAPI() # 假设这是你的RAG生成函数改造成了异步生成器 async def rag_stream_generator(user_query: str, context_chunks: list) - AsyncGenerator[str, None]: 模拟LLM流式生成基于检索到的上下文(chunks)生成回答。 # 1. 构建Prompt (简化示例) prompt f基于以下信息回答问题{context_chunks}\n\n问题{user_query}\n回答 # 2. 这里模拟调用一个支持流式输出的LLM例如通过OpenAI API或本地vLLM # 假设 call_llm_stream 是一个能返回异步生成器的函数 async for chunk in call_llm_stream(prompt): # 每次yield一个生成的文本片段 yield fdata: {json.dumps({text: chunk}, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n app.post(/chat/stream) async def chat_stream(query: str): # 1. 检索相关上下文 (这里简化实际应从向量库检索) retrieved_chunks [知识块1内容..., 知识块2内容...] # 替换为实际检索逻辑 if not retrieved_chunks: raise HTTPException(status_code404, detail未找到相关信息) # 2. 返回流式响应媒体类型为 text/event-stream (Server-Sent Events) return StreamingResponse( rag_stream_generator(query, retrieved_chunks), media_typetext/event-stream ) # 前端的JavaScript就可以通过EventSource来接收并实时显示这些数据块了。4. 性能优化让系统又快又稳上线后随着用户量增加我们做了一系列优化。缓存策略设计很多用户问的是高频问题比如“客服电话多少”。每次都对相同问题进行检索和生成太浪费。我们引入了两级缓存内存缓存Redis存储“问题向量 - top-K知识块ID”的映射。因为问题向量是固定的相同问题检索结果相同缓存命中后直接跳过向量搜索。回答缓存Redis对于完全相同的用户问题直接缓存最终生成的回答文本。我们给缓存设置了合理的TTL例如10分钟平衡响应速度和知识更新需求。并发查询处理FastAPI本身支持异步但我们的一些组件如某些本地模型推理是同步的。为了避免这些阻塞操作拖慢整个系统我们使用asyncio.to_thread将同步的检索、模型调用函数放到线程池中执行防止它们阻塞事件循环。余弦相似度阈值调优检索时不能把所有相似度大于0的结果都返回。我们通过测试集设定了一个相似度阈值例如0.75。只有当最相关片段的相似度超过这个阈值才会将其用作生成答案的上下文否则系统直接回复“未找到相关信息”避免LLM根据不相关的资料胡编乱造。这个阈值需要根据实际数据分布反复调整。5. 避坑指南前人踩坑后人绕行在开发过程中我们遇到了几个典型问题这里分享出来希望大家能避开。知识库冷启动问题系统刚上线时知识库内容少很多问题检索不到相关内容。我们的解决方案是构建一个高质量的种子FAQ覆盖最常见、最核心的问题。引入“回退策略”当检索结果为空或相似度太低时不是直接报错而是回退到一个通用的、基于LLM自身知识的应答模式可以明确告知用户“以下是通用建议”或者引导用户转接人工。建立反馈闭环将“未找到答案”的问题记录下来定期由运营人员补充进知识库。敏感信息过滤方案知识库里难免有内部联系方式、价格策略等敏感信息。我们做了两层过滤入库前过滤在知识库构建阶段通过关键词匹配或正则表达式识别并剔除或脱敏明确标为机密的段落。输出前过滤在LLM生成答案后、返回给用户前再用一个轻量级的分类模型或规则对最终答案做一次扫描确保没有泄露敏感词。对话状态管理真实的客服对话是有上下文的。我们实现了一个简单的多轮对话记忆窗口。将当前问题和之前几轮比如最近3轮的对话历史拼接起来作为一个“增强查询”去检索。这样当用户问“它续航怎么样”时系统能知道“它”指的是上一轮提到的“手机X”。在Prompt中明确告诉LLM之前的对话历史使其能进行连贯的上下文应答。6. 代码示例一个完整的FastAPI服务端点下面是一个简化但相对完整的后端接口示例展示了检索与生成的核心逻辑。from fastapi import FastAPI, Depends, HTTPException from pydantic import BaseModel import numpy as np import faiss from sentence_transformers import SentenceTransformer from typing import List import hashlib import redis import json # --- 初始化组件 --- app FastAPI(titleRAG智能客服API) # 1. 加载嵌入模型 embed_model SentenceTransformer(BAAI/bge-base-zh) # 2. 加载FAISS索引 (假设已提前构建好) index faiss.read_index(knowledge_base.index) with open(chunk_texts.json, r) as f: chunk_texts json.load(f) # 存储文本块的列表与index顺序对应 # 3. 连接Redis缓存 redis_client redis.Redis(hostlocalhost, port6379, db0) # 4. 模拟LLM调用 (实际应替换为真实的LLM API或本地调用) def call_llm(prompt: str) - str: # 这里简化处理实际应接入Qwen等模型 return f模拟生成基于提示词的答案。提示词{prompt[:50]}... # --- 数据模型 --- class ChatRequest(BaseModel): question: str session_id: str None # 用于多轮对话 class ChatResponse(BaseModel): answer: str sources: List[str] # 引用的知识片段来源 # --- 核心检索函数 --- def retrieve_chunks(query: str, top_k: int 3, threshold: float 0.75) - List[str]: 检索与查询最相关的文本块。 # 将查询转换为向量 query_vector embed_model.encode([query], normalize_embeddingsTrue) # 在FAISS中搜索 distances, indices index.search(query_vector, top_k) # 根据阈值过滤结果 retrieved_chunks [] for dist, idx in zip(distances[0], indices[0]): if 1 - dist threshold: # FAISS返回的是L2距离我们近似处理。更严谨应使用内积或余弦距离。 retrieved_chunks.append(chunk_texts[idx]) return retrieved_chunks # --- 主要聊天端点 --- app.post(/chat, response_modelChatResponse) async def chat(request: ChatRequest): user_question request.question # 1. 检查缓存 cache_key hashlib.md5(user_question.encode()).hexdigest() cached_answer redis_client.get(cache_key) if cached_answer: return ChatResponse(answercached_answer.decode(), sources[来自缓存]) # 2. 检索相关上下文 context_chunks retrieve_chunks(user_question) if not context_chunks: # 可以在这里实现回退策略 return ChatResponse( answer抱歉在现有知识库中未找到相关问题的明确答案。, sources[] ) # 3. 构建Prompt context_str \n\n.join([f[{i1}] {chunk} for i, chunk in enumerate(context_chunks)]) prompt f你是一个专业的客服助手。请严格根据以下提供的参考信息来回答问题。如果参考信息中没有答案请直接说“根据现有资料我暂时无法回答这个问题”。 参考信息 {context_str} 用户问题{user_question} 请生成回答 # 4. 调用LLM生成答案 answer call_llm(prompt) # 5. 缓存答案 (设置10分钟过期) redis_client.setex(cache_key, 600, answer) # 6. 返回结果 return ChatResponse(answeranswer, sourcescontext_chunks) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)7. 结尾思考如何评估RAG系统的回答质量系统上线后我们面临一个持续性的问题怎么知道它的回答好不好准确率有多少这比评估一个分类模型要复杂得多。我们目前采用“人工评估自动指标”相结合的方式人工评估定期抽样一批用户问题及其回答由专业人员从“相关性”、“准确性”、“有用性”、“流畅性”几个维度打分。这是黄金标准但成本高。自动指标检索相关度评估检索到的知识块与问题的匹配程度。答案忠实度生成的答案在多大程度上源自检索到的上下文而不是LLM的“幻觉”。可以通过计算答案句子与上下文句子的重叠度或使用NLI自然语言推理模型来判断。答案相关性答案是否直接回答了用户的问题。但这还不够。一个开放性的问题是是否存在一套标准化的、可量化的、全面的评估体系能够像评估精确率/召回率那样自动化地评估一个RAG智能客服系统的回答准确性特别是在多轮对话、上下文依赖、以及需要综合多个知识片段进行推理的复杂场景下如何设计这样的评估基准这是我们和业界都在探索的方向。总的来说通过RAG搭建智能客服是一个迭代和优化的过程。从技术选型到核心实现再到性能调优和问题规避每一步都需要结合业务实际进行考量。希望这篇笔记能为你提供一条清晰的实践路径。