摘要
检索到的文档块不能直接扔给用户，得让模型读了资料再回答。本文讲怎么把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, sourcesmain.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、高质量的检索、测试集验证和必要的人工审核。