夜雨聆风
AI 客服 Day3:丢掉 FAQ 文档,用 RAG 让 AI 自己去知识库找答案
这一章解决”AI 一本正经地胡说八道”的问题。通过 RAG（检索增强生成），让机器人在回答之前先查自己公司的知识库，基于真实文档给出准确答案。
本章目录
-
RAG 是什么，为什么要用它 -
文档加载与切分 -
向量存储与检索 -
RAG 问答系统 -
实战：完整知识库客服
1. RAG 是什么，为什么要用它
大家可能都有过这种体验：问 AI 一个很具体的问题，比如”你们家 StarPods Pro 支不支持 Android”，AI 给了一个看起来很靠谱的答案，但实际上完全是编的。这种现象有个专业名词叫”幻觉”。
为什么会产生幻觉？
LLM 的”知识”来自训练数据，训练完就固化了。它不知道你们公司最新的产品参数，不知道你们的退款政策，更不知道你们内部的 FAQ。让它直接回答这类问题，只能靠编。
RAG 怎么解决这个问题？
普通 LLM 回答：用户问题 → LLM 直接凭印象回答（可能编造）RAG 回答：用户问题 → 先去知识库搜相关内容 → 把搜到的内容 + 用户问题一起发给 LLM → 基于真实资料回答换个通俗的说法：RAG 相当于给 AI 开卷考试。普通模式是闭卷，考啥得全靠记忆；RAG 是开卷，能翻书查。
2. 文档加载与切分
文件：01_document_loader.py
知识库里的文档，大部分是 Markdown、PDF、Word，或者纯文本。第一步是把这些文件读进来，然后切成小块。
为什么要切分？
LLM 的输入有长度限制（Context Window）。一份 100 页的产品手册不可能整个塞进去，得切成几百字的小块，检索时只取相关的几块。
怎么切？
常见策略是按字数切，同时保留一定重叠量，避免一句话被切断导致语义丢失：
文档内容： [ Block 1 ][ Block 2 ][ Block 3 ]带重叠后： [ Block 1 ↔ Block 2 ↔ Block 3 ]这一章用的是公司产品知识库（knowledge_base/product_knowledge.md），大家可以直接换成自己公司的文档。
核心代码
加载文档
from langchain_community.document_loaders import TextLoader# 加载单个文档loader = TextLoader("knowledge_base/product_knowledge.md", encoding="utf-8")documents = loader.load()print(f"文档字符数: {len(documents[0].page_content)}")print(f"元数据: {documents[0].metadata}") # 包含 source 文件路径文本切分（关键步骤）
from langchain.text_splitter import RecursiveCharacterTextSplitter# RecursiveCharacterTextSplitter：递归切分，尽量在语义边界处断开text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个块最大 500 字符 chunk_overlap=50, # 相邻块重叠 50 字符（保持连贯） length_function=len, separators=["\n## ", # 优先在二级标题处切"\n### ", # 其次在三级标题处切"\n\n", # 再次在段落处切"\n", # 再在换行处切"。", # 中文句号"，", # 中文逗号" ", # 空格"", # 最后才在任意字符处切 ],)chunks = text_splitter.split_documents(documents)print(f"切分为 {len(chunks)} 个文本块")💡 中文优化：默认 Separators 对英文友好，中文场景需要加上
。，，等标点，否则可能把一个完整句子拦腰切断。
不同切分参数对比
|
|
|
|
|
|---|---|---|---|
|
|
|
|
|
|
|
|
|
通用推荐 |
|
|
|
|
|
3. 向量存储与检索
文件：02_vector_store.py
文档切分好了，怎么让计算机知道哪一块跟用户的问题”最相关”？
原理：把每一块文本都通过 Embedding 模型转成一个向量（高维空间里的一个点）。语义相近的内容，向量之间距离就近。用户问问题时，也把问题转成向量，然后找距离最近的几个文档块。
"StarPods Pro 多少钱" → 向量 [0.23, -0.11, 0.87, ...]"StarPods Pro 售价 699" → 向量 [0.25, -0.09, 0.85, ...] ← 距离近，命中！"退款政策是什么" → 向量 [0.61, 0.33, -0.12, ...] ← 距离远，不命中这章用的是 Milvus 做向量存储。
Milvus 有一个很方便的 Milvus Lite 模式——不需要装任何服务，直接一个 Python 文件就能跑，跟用 SQLite 一样简单。代码里默认用的就是 Lite 模式。等数据量上来了，把 URI 从本地文件改成 Milvus Server 的地址就行，一行配置的事。
核心代码
创建 Milvus 向量存储
from langchain_milvus import Milvusfrom config import get_embeddings, get_milvus_configembeddings = get_embeddings()milvus_config = get_milvus_config()# Milvus.from_documents：自动完成向量化 + 入库vectorstore = Milvus.from_documents( documents=chunks, # 切分后的文档块 embedding=embeddings, # Embedding 模型 connection_args={"uri": milvus_config["uri"]}, # Lite 模式 = 本地文件路径 collection_name="customer_service_kb", drop_old=True, # 每次重新创建（生产环境改为 False）)print(f"向量库已创建，共 {len(chunks)} 个向量")相似度检索
# 检索最相关的 top_k 个文档块results = vectorstore.similarity_search("耳机连接不稳定怎么办", k=3)for i, doc in enumerate(results, 1):print(f"[{i}] {doc.page_content[:80]}...")# 带相似度分数（Milvus 默认用内积 IP，分数越大越相似）results_with_scores = vectorstore.similarity_search_with_score("降噪效果不好", k=5)for i, (doc, score) in enumerate(results_with_scores, 1):print(f"[{i}] 分数={score:.4f} | {doc.page_content[:60]}...")从已有向量库加载（不需要重新计算）
# 只要 uri 和 collection_name 一致，数据就还在loaded_vectorstore = Milvus( embedding_function=embeddings, connection_args={"uri": milvus_config["uri"]}, collection_name="customer_service_kb",)# 直接用它检索，不需要重新算向量results = loaded_vectorstore.similarity_search("StarPods Pro 的价格", k=2)💡 Milvus 三种部署模式：
uri="milvus.db"→ Milvus Lite（开发用，零部署）uri="http://localhost:19530"→ Milvus Server（自建集群）uri="https://xxx.zilliz.cloud"→ Zilliz Cloud（托管服务，免运维）
4. RAG 问答系统
文件：03_rag_qa.py
把文档加载、向量检索、LLM 问答三步串起来，形成一个完整的 RAG 链：
用户问题 │ ▼向量检索（找最相关的 3 个文档块） │ ▼拼装 Prompt（相关文档 + 用户问题 → 发给 LLM） │ ▼LLM 基于文档内容给出答案代码用的是 LangChain 的 LCEL（表达式语言）写法，一行代码就能串起一条完整的处理链。
核心代码
构建 RAG 链（LCEL 写法）
from langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.runnables import RunnablePassthrough# 1. 创建检索器（从向量库取 top_k 个相关文档）retriever = vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 3},)# 2. RAG 提示词模板（关键：用 {context} 注入检索结果）RAG_PROMPT_TEMPLATE = """你是「星辰科技」的客服助手。请根据参考资料回答用户问题。## 回答规则1. 只基于参考资料回答，不要编造信息2. 资料中没有的信息，诚实回答"抱歉，我没有找到相关信息"3. 回答简洁，3 句以内## 参考资料{context}## 用户问题{question}## 回答"""rag_prompt = ChatPromptTemplate.from_template(RAG_PROMPT_TEMPLATE)# 3. 用 LCEL 组装完整 RAG 链def format_docs(docs):return "\n\n---\n\n".join(doc.page_content for doc in docs)rag_chain = ( {"context": retriever | format_docs, # 先检索并格式化文档"question": RunnablePassthrough()} # 原样传递用户问题 | rag_prompt # 填充模板 | llm # 调用 LLM | StrOutputParser() # 解析输出为字符串)# 使用：直接调用链answer = rag_chain.invoke("StarPods Pro 的降噪深度是多少？")print(answer)带来源引用的回复
# 在提示词中要求 LLM 标注信息来源RAG_PROMPT_WITH_SOURCES = """请根据参考资料回答，并在回答末尾用「参考：xxx」标注来源。## 参考资料{context}## 用户问题{question}## 回答（含来源标注）"""rag_chain_sources = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | ChatPromptTemplate.from_template(RAG_PROMPT_WITH_SOURCES) | llm | StrOutputParser())answer = rag_chain_sources.invoke("手表的心率监测准确吗？")# 输出示例：# "StarWatch X 采用医疗级心率传感器，误差范围±3bpm。参考：产品知识库-智能手表章节"✅ RAG 链数据流：
用户问题→retriever 检索→format_docs 格式化→prompt 填充→LLM 生成→StrOutputParser 解析
5. 实战：完整知识库客服
文件：04_full_rag_bot.py
第 4 个文件解决一个在实际项目中很容易踩坑的问题：追问上下文丢失。
举个例子：
用户：StarPods Pro 的保修期是多久？AI： 一年质保。用户：那如果坏了怎么申请？ ← "那"指代什么？如果直接把「那如果坏了怎么申请」拿去向量检索，搜不到任何东西——因为这句话语义不完整。
解决方案：Standalone Question Generation
在检索之前，先让 LLM 把追问改写成独立完整的问题：
「那如果坏了怎么申请？」 ↓「StarPods Pro 在质保期内损坏，如何申请售后维修？」 ↓拿这个完整的问题去向量检索，就能命中了这个细节在实际项目里非常重要，大家落地时别忘了加上。
核心代码
独立问题生成链
STANDALONE_QUESTION_PROMPT = """根据对话历史和最新问题，生成一个独立的、完整的问题。生成的问题应该包含对话中提到的具体产品名称和上下文信息。对话历史：{chat_history}最新问题：{question}独立问题："""# 用 LCEL 构建独立问题生成链question_chain = ( ChatPromptTemplate.from_template(STANDALONE_QUESTION_PROMPT) | llm | StrOutputParser())# 使用：先把追问转成独立问题，再去检索history_text = "用户: StarPods Pro 的保修期是多久？\n客服: 一年质保。"user_input = "那如果坏了怎么申请？"standalone_question = question_chain.invoke({"chat_history": history_text,"question": user_input,})# 输出：「StarPods Pro 在质保期内损坏，如何申请售后维修？」完整 RAG 客服机器人类
class RAGCustomerServiceBot:def __init__(self, window_size: int = 5):self.llm = get_llm(temperature=0.3)self.vectorstore = get_or_build_vectorstore(get_embeddings())self.retriever = self.vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 3} )self.memory = ConversationBufferWindowMemory( k=window_size, return_messages=True, memory_key="chat_history" )self._build_chains()def _build_chains(self):# 链 1：生成独立问题self.question_chain = ( ChatPromptTemplate.from_template(STANDALONE_QUESTION_PROMPT) | self.llm | StrOutputParser() )# 链 2：基于文档回答问题self.answer_chain = ( ChatPromptTemplate.from_template(RAG_ANSWER_PROMPT) | self.llm | StrOutputParser() )def chat(self, user_input: str) -> str:# Step 1: 加载历史 history = self.memory.load_memory_variables({}).get("chat_history", []) history_text = format_chat_history(history)# Step 2: 生成独立问题（解决追问上下文丢失） standalone_question = self.question_chain.invoke({"chat_history": history_text,"question": user_input, })# Step 3: 用独立问题去检索 docs = self.retriever.invoke(standalone_question) context = format_docs(docs)# Step 4: 基于文档生成回复 answer = self.answer_chain.invoke({"chat_history": history_text,"context": context,"question": standalone_question, })# Step 5: 保存对话self.memory.save_context( {"input": user_input}, {"output": answer} )return answer, standalone_question, docs实测效果对比
|
|
|
|
|---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
✅ 本章完结：现在机器人能基于真实文档回答问题，不会胡编乱造；追问也能正确关联到之前提到的产品。
代码文件
|
|
|
|---|---|
01_document_loader.py |
|
02_vector_store.py |
|
03_rag_qa.py |
|
04_full_rag_bot.py |
|
下一章
第四章：情绪识别与智能路由
知识库搭起来之后，机器人能回答产品问题了。但有些用户不只是要答案，他们可能在发火，可能要退款，可能需要人工介入。第四章就来处理这些情况。
往期推荐
AI 客服 Day2：为什么你的 AI 聊着聊着就”失忆”了？
👉 关注我回复“源码”｜获取本章完整代码