LangChain RAG检索增强生成

摘要

检索到的文档块不能直接扔给用户，得让模型读了资料再回答。本文讲怎么把Milvus转成Retriever、把文档块整理成上下文、写RAG专用的Prompt，以及最终拼成一条完整的RAG Chain。顺带处理一个问题：资料不足时让模型老实说"不知道"。

一、什么是RAG

RAG全称Retrieval-Augmented Generation，检索增强生成。

核心思路就三步：

用户提问 → 从知识库检索相关资料 → 把资料和问题一起交给模型生成答案

RAG不训练模型，不改模型参数，只是在调用时往Prompt里塞外部资料。

二、为什么需要RAG

模型不知道企业内部资料。DeepSeek不知道你们公司的考勤制度，你得把制度文档喂给它。

模型知识不是最新的。企业制度、产品价格随时在变，RAG用最新资料就能答最新问题。

模型可能编造答案。不知道的时候会硬编，RAG可以要求它"不知道就说不知道"。

回答需要展示来源。企业场景不仅要答案，还得告诉用户"这是来自员工手册第3页"。

三、RAG的两个阶段

索引阶段：加载文档→切分→向量化→存Milvus。跑一次就行，资料变了再重跑。

问答阶段：用户提问→检索相关块→组装Prompt→模型生成答案。每次问答都要走一遍。

第七章做了索引阶段，本章做问答阶段。

四、Retriever检索器

Retriever是LangChain里统一的检索接口。输入字符串问题，返回Document列表。

python

from langchain_milvus import Milvus from utils.model_factory import get_embeddings embeddings = get_embeddings() vector_store = Milvus( embedding_function=embeddings, connection_args={"host": "localhost", "port": "19530"}, collection_name="knowledge_base" ) retriever = vector_store.as_retriever(search_kwargs={"k": 3}) docs = retriever.invoke("员工迟到怎么处理") for doc in docs: print(doc.page_content) print(doc.metadata)

VectorStore负责存向量和相似度搜索，Retriever负责"根据问题返回文档"这个动作。RAG Chain通常依赖Retriever。

五、案例一：使用Retriever检索文档

python

from langchain_milvus import Milvus from utils.model_factory import get_embeddings embeddings = get_embeddings() vector_store = Milvus( embedding_function=embeddings, connection_args={"host": "localhost", "port": "19530"}, collection_name="knowledge_base" ) retriever = vector_store.as_retriever(search_kwargs={"k": 3}) question = "公司对迟到早退有什么规定" docs = retriever.invoke(question) for i, doc in enumerate(docs, 1): print(f"结果{i}") print(f"内容：{doc.page_content[:150]}...") print(f"来源：{doc.metadata.get('file_name')}") if doc.metadata.get("page") is not None: print(f"页码：{doc.metadata.get('page') + 1}") print("-" * 40)

六、整理文档上下文

Retriever返回的是Document列表，不能直接塞进Prompt。需要把多个文档块拼成一段上下文文本：

python

def format_documents(docs): if not docs: return "未找到相关资料。" parts = [] for i, doc in enumerate(docs, 1): source = doc.metadata.get("file_name", "未知来源") page = doc.metadata.get("page") if page is not None: source_info = f"【资料{i}】来自《{source}》第{page + 1}页" else: source_info = f"【资料{i}】来自《{source}》" parts.append(f"{source_info}\n{doc.page_content}") return "\n\n".join(parts)

七、编写RAG专用Prompt

RAG Prompt要告诉模型四件事：你是谁、用户问了什么、可以参考什么资料、资料不够怎么办。

python

from langchain.prompts import ChatPromptTemplate template = ChatPromptTemplate.from_messages([ ("system", """你是企业知识库助手。请严格根据以下参考资料回答用户问题。 参考资料： {context} 规则： 1. 只根据参考资料回答，不要使用自己的知识 2. 如果参考资料中没有相关信息，明确说"资料中没有提到" 3. 不要编造答案 4. 引用资料时注明来源文件名称"""), ("human", "{question}") ])

"资料不足时说不知道"这条规则很关键，能有效降低模型瞎编的概率。

八、案例二：手动完成一次RAG问答

先拆开步骤写，理解之后再上LCEL：

python

from langchain_milvus import Milvus from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from utils.model_factory import get_deepseek_model, get_embeddings embeddings = get_embeddings() model = get_deepseek_model(temperature=0.1) parser = StrOutputParser() vector_store = Milvus( embedding_function=embeddings, connection_args={"host": "localhost", "port": "19530"}, collection_name="knowledge_base" ) retriever = vector_store.as_retriever(search_kwargs={"k": 3}) def format_documents(docs): parts = [] for i, doc in enumerate(docs, 1): source = doc.metadata.get("file_name", "未知") parts.append(f"【资料{i}】来自《{source}》\n{doc.page_content}") return "\n\n".join(parts) template = ChatPromptTemplate.from_messages([ ("system", "根据参考资料回答问题。参考资料：\n{context}\n如果资料中没有相关信息，就说不知道。"), ("human", "{question}") ]) # 手动走三步 question = "员工每年有几天年假" docs = retriever.invoke(question) context = format_documents(docs) prompt = template.invoke({"context": context, "question": question}) response = model.invoke(prompt) answer = parser.invoke(response) print(answer)

三步走完：检索→格式化→生成。

九、使用LCEL组合RAG Chain

Retriever也是Runnable，可以接入LCEL管道：

python

from langchain_core.runnables import RunnablePassthrough from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from utils.model_factory import get_deepseek_model model = get_deepseek_model(temperature=0.1) parser = StrOutputParser() def format_documents(docs): parts = [] for i, doc in enumerate(docs, 1): source = doc.metadata.get("file_name", "未知") parts.append(f"【资料{i}】来自《{source}》\n{doc.page_content}") return "\n\n".join(parts) template = ChatPromptTemplate.from_messages([ ("system", "根据参考资料回答问题。参考资料：\n{context}\n如果资料中没有相关信息，就说不知道。"), ("human", "{question}") ]) chain = ( { "context": retriever | format_documents, "question": RunnablePassthrough() } | template | model | parser ) result = chain.invoke("员工每年有几天年假") print(result)

数据流动：输入问题→context分支走检索+格式化，question分支原样透传→合并成字典→进Prompt→进模型→进Parser。

十、LCEL写法的一个限制

上面这条链直接返回字符串答案，拿不到来源信息。如果还要展示来源，得再查一次，但两次检索结果可能不一样。

所以完整案例改用服务函数封装，答案和来源用同一次检索结果：

python

def rag_answer(question): docs = retriever.invoke(question) context = format_documents(docs) prompt = template.invoke({"context": context, "question": question}) response = model.invoke(prompt) answer = parser.invoke(response) sources = extract_sources(docs) return answer, sources

答案和来源用同一批docs，不会对不上。

十一、在回答中展示来源

从metadata里读来源，不要靠模型自己猜：

python

def extract_sources(docs): seen = set() sources = [] for doc in docs: file_name = doc.metadata.get("file_name", "未知文件") page = doc.metadata.get("page") key = f"{file_name}_{page}" if key not in seen: seen.add(key) if page is not None: sources.append(f"《{file_name}》第{page + 1}页") else: sources.append(f"《{file_name}》") return sources

用set去重，避免多个文档块来自同一文件同一页时重复展示。

十二、资料不足时拒绝编造

资料里没有的信息，让模型明确说不知道。Prompt里写清楚规则，然后测试：

python

question = "公司年终奖怎么发" # 知识库里没有 answer, sources = rag_answer(question) print(answer) # 预期输出类似"资料中没有提到年终奖的相关信息"

需要说明：大模型不是程序，不能保证100%遵守规则。正式项目还要结合相似度阈值、测试集验证等手段。

十三、企业案例：内部知识库问答系统

项目结构：

text

knowledge_base/ # 原始文档 build_index.py # 第七章，建索引 rag_service.py # RAG服务封装 main.py # 终端交互程序

rag_service.py：

python

from langchain_milvus import Milvus from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from utils.model_factory import get_deepseek_model, get_embeddings class RAGService: def __init__(self, collection_name="knowledge_base"): self.embeddings = get_embeddings() self.model = get_deepseek_model(temperature=0.1) self.parser = StrOutputParser() self.vector_store = Milvus( embedding_function=self.embeddings, connection_args={"host": "localhost", "port": "19530"}, collection_name=collection_name ) self.retriever = self.vector_store.as_retriever(search_kwargs={"k": 3}) self.template = ChatPromptTemplate.from_messages([ ("system", """你是企业知识库助手。严格根据参考资料回答用户问题。 参考资料： {context} 规则： 1. 只根据参考资料回答 2. 资料中没有相关信息时，明确说"资料中没有提到" 3. 不编造答案 4. 引用资料时注明来源文件"""), ("human", "{question}") ]) def _format_docs(self, docs): parts = [] for i, doc in enumerate(docs, 1): source = doc.metadata.get("file_name", "未知文件") parts.append(f"【资料{i}】来自《{source}》\n{doc.page_content}") return "\n\n".join(parts) def _extract_sources(self, docs): seen = set() sources = [] for doc in docs: file_name = doc.metadata.get("file_name", "未知文件") page = doc.metadata.get("page") key = f"{file_name}_{page}" if key not in seen: seen.add(key) if page is not None: sources.append(f"《{file_name}》第{page + 1}页") else: sources.append(f"《{file_name}》") return sources def ask(self, question): docs = self.retriever.invoke(question) context = self._format_docs(docs) prompt = self.template.invoke({"context": context, "question": question}) response = self.model.invoke(prompt) answer = self.parser.invoke(response) sources = self._extract_sources(docs) return answer, sources

main.py：

python

from rag_service import RAGService def main(): rag = RAGService() print("企业知识库问答系统已启动，输入exit退出") while True: question = input("\n问题：").strip() if question.lower() == "exit": break if not question: continue answer, sources = rag.ask(question) print(f"\n答案：{answer}") if sources: print(f"\n来源：{', '.join(sources)}") if __name__ == "__main__": main()

运行前确保已执行build_index.py建好索引。

十四、search_type调整检索方式

similarity：默认，按相似度从高到低取前k个。简单直接，但文档块内容重叠时容易返回重复信息。

MMR（Maximum Marginal Relevance）：先多取一些候选（fetch_k），再从中挑选既相关又多样化的k个结果，避免重复。

python

retriever = vector_store.as_retriever( search_type="mmr", search_kwargs={"k": 3, "fetch_k": 10, "lambda_mult": 0.5} )

lambda_mult越靠近1越偏向相似度优先，越靠近0越偏向多样性优先。文档切片重叠多时用MMR，切片内容独立时用similarity。

十五、k值怎么设

k太小可能遗漏资料，k太大可能混入无关内容、增加token消耗。

从k=3或k=5开始试，根据真实问题效果调整。

十六、RAG调试顺序

回答不对时按顺序查：

打印检索结果，确认文档对不对：

python

docs = retriever.invoke(question) for doc in docs: print(doc.page_content)

打印格式化后的上下文，确认模型实际收到了什么：

python

context = format_documents(docs) print(context)

确认检索正确后再调Prompt。顺序是：数据→切分→检索→Prompt→模型，从前往后查。

常见问题

检索正确但回答错误

查Prompt是否明确要求根据资料回答、上下文是否完整传入、多个文档间是否有冲突。打印最终Prompt能看到模型实际收到了什么。

检索结果就是错的

问题在检索阶段：切分不合理、Embedding效果差、k值不对、文档缺少用户常用表达。改Prompt没用，得改索引。

文档更新后回答还是旧的

向量库里存的还是旧数据，重新跑build_index.py。

RAG能完全避免幻觉吗

不能。只能明显降低概率，还需要清晰的Prompt、高质量的检索、测试集验证和必要的人工审核。