引言
随着自然语言处理（NLP）和大模型技术的发展，构建基于大模型的知识库检索增强生成（RAG, Retrieval-Augmented Generation）系统成为许多项目的重要组成部分。本文将详细介绍我如何结合LangChain实现一个完整的RAG流程，涵盖从数据读取到效果评估的具体步骤，并分享一些遇到的问题及解决方案。
1. LangChain与RAG简介
为什么使用外挂知识库？
尽管大模型在自然语言处理任务中表现出色，但它们仍然存在一定的局限性，如对最新信息的获取不足、特定领域知识的欠缺等。为了弥补这些不足，我们采用了RAG技术，通过给大模型外挂知识库，提升其回答问题的准确性和时效性。
LangChain的作用
LangChain是一个强大的工具集，提供了丰富的函数和模块，帮助我们轻松实现RAG流程中的各个环节。它不仅简化了开发过程，还提高了系统的灵活性和可扩展性。本文将详细介绍如何利用LangChain实现一个完整的RAG系统。
上图展示了一个复杂的语言处理系统，其中包含模型、提示、链、代理和嵌入与向量存储。
- Models 负责理解和生成语言，提示用于引导模型输出；
- Chains 代表将多个步骤串联起来完成复杂任务的过程；
- Agents 则用于让模型与外部环境互动，比如执行API调用。
- Embedding 嵌入与向量存储 VectorStore 是数据表示和检索的手段，为模型提供必要的语言理解基础。
2. RAG开发流程
第一步：数据读取
首先，我们需要读取本地的数据文件。LangChain提供了TextLoader函数，可以方便地读取txt文件。除此之外，我还实现了对PDF和Word文件的支持：
PDF文件读取：使用
PyPDF2时，会将普通文字和表格中的文字一起读出来，无法区分开。为了解决这个问题，我采用了fitz库。fitz库能够分块读取PDF内容，每一块都是一个元组，其中第5个位置如果是0，则代表普通文本；如果不是0，则代表表格中的数据。利用这个特性，我通过if-else判断提取出所需的文本，剔除表格数据。Word文件读取：使用
python-docx库，可以方便地提取Word文档中的文本和表格。
from langchain.document_loaders import TextLoader import fitz from docx import Document def load_pdf(file_path): doc = fitz.open(file_path) text = "" for page_num in range(len(doc)): page = doc.load_page(page_num) blocks = page.get_text("blocks") for block in blocks: if block[4] == 0: # 普通文本 text += block[4] return text def load_word(file_path): doc = Document(file_path) text = "\n".join([para.text for para in doc.paragraphs]) return text背景信息补充：不同文件格式的读取方法各有优劣。例如，PDF文件通常包含丰富的排版信息，而Word文件则可能包含复杂的表格和图表。选择合适的库和方法，可以确保数据的完整性和准确性。
第二步：数据清洗和预处理
数据清洗主要包括四个部分的工作：
指代消解：将原文中的代词替换成真实值。例如，将“他”替换为具体的对象名称。这一步可以通过大模型提示词来实现：
def resolve_coreferences(text, llm): prompt = f"请将文本块中的代词替换成真实值：{text}" response = llm(prompt) return response名称替换：将知识库中的不同名称用统一的名称进行替换。例如，“大模型”、“llm”、“大语言模型”统一替换为“大模型”。
剔除杂乱数据：去除特殊符号和没有意义的信息。
标题匹配：对于层级较多的文本，给每个切分后的段落匹配上对应的一级和二级标题。
def clean_and_preprocess(text, llm): text = resolve_coreferences(text, llm) text = replace_names(text) text = remove_noise(text) text = match_titles(text) return text
背景信息补充：数据清洗是确保后续处理质量的关键步骤。不干净的数据可能导致错误的分析结果或低效的模型训练。通过细致的数据清洗，我们可以显著提高系统的性能和可靠性。
第三步：文本切分
处理好的数据需要转换成Document类型，然后送入TextSplitter进行文本切分。我采取了两种切分方式：
固定字数切分：设置一定的重叠字数，缓解切分带来的语义破坏。
递归切分：先按段落切，如果段落太大，再按句子切，最后按标点符号或字数切。
from langchain.text_splitter import TextSplitter, ChineseRecursiveTextSplitter def split_text(text, method="fixed"): if method == "fixed": splitter = TextSplitter(chunk_size=100, chunk_overlap=20) elif method == "recursive": splitter = ChineseRecursiveTextSplitter() else: raise ValueError("Invalid splitting method") documents = splitter.split_documents(text) return documents
背景信息补充：文本切分的方式直接影响到后续向量化的质量和检索效率。合理选择切分策略，可以在保持语义完整性的同时，最大化检索的准确性。
第四步：文本转向量与存储
文本切分完成后，需要将文本块转成向量并存储。我选择了m3e模型进行文本转向量，并使用faiss作为向量库。具体做法如下：
下载并部署m3e模型：根据GitHub上的README安装依赖并测试模型。
向量存储与检索优化：检测存储结果，避免重复导入数据，提升问答速度。设置聚类数量以加速检索。调用
faiss时需加入allow_dangerous_deserialization=True参数。from sentence_transformers import SentenceTransformer import faiss def vectorize_and_store(text_chunks): model = SentenceTransformer('m3e-base') vectors = model.encode(text_chunks) index = faiss.IndexFlatL2(vectors.shape[1]) index.add(vectors) return index
背景信息补充：向量化和存储是RAG系统的核心环节之一。高效的向量库可以显著提升检索速度和精度，从而改善用户体验。
第五步：多路召回策略
我采用了三种召回策略：
正常分割 + m3e转向量：每段添加主题词和上下文摘要，选出10块。
正常分割 + tf-idf转向量：同样添加主题词和上下文摘要，选出10块。
子块召回父块：先将文档分成大块，再细分成小块，标记清楚每个小块属于哪个大块。用户提问时，先匹配小块，返回对应的大块以包含更多上下文信息。
def multi_recall(query, chunks): results = [] results.extend(normal_recall(query, chunks, 'm3e')) results.extend(normal_recall(query, chunks, 'tf-idf')) results.extend(parent_chunk_recall(query, chunks)) return results
背景信息补充：多路召回策略可以提高检索结果的多样性和准确性。不同的召回方法各有优势，综合使用可以更好地覆盖用户的查询需求。
第六步：精排策略
多路召回后，对几十个文本块进行进一步筛选，选出最相关的5个文本块。具体方法是采用余弦相似性和BM25模型进行重排。
from sklearn.metrics.pairwise import cosine_similarity from rank_bm25 import BM25Okapi def rerank(results, query): cos_sim = cosine_similarity(results, [query]) bm25_scores = BM25Okapi(results).get_scores(query) combined_scores = (cos_sim + bm25_scores) / 2 top_5 = sorted(combined_scores, reverse=True)[:5] return top_5背景信息补充：精排策略确保最终呈现给用户的结果是最相关和最有用的。合理的排序算法可以显著提升用户的满意度。
第七步：Query转写
针对用户提问不明确或信息不全的情况，提出了多种Query转写方案：
换表达方式：保持语义不变的情况下，换一种表达方式。
关键词匹配：提取关键实体词，与召回结果进行匹配。
结合上下文改写：例如，将“Langchain 如何工作？”转写为“Langchain 的工作流程有几个步骤？”。
任务拆分：将复合问题拆分为多个简单问题分别检索。
宽泛问题：将具体问题改为宽泛问题，如“Langchain到底是什么？”。
大模型自答：让大模型先回答问题，再将答案连同原问题一起送入RAG检索。
def rewrite_query(query, llm): rewritten_queries = [ llm(f"请换一种表达方式：{query}"), extract_keywords(query), combine_context(query), split_tasks(query), broaden_question(query), llm_answer(query) ] return rewritten_queries背景信息补充：Query转写可以帮助系统更好地理解用户的意图，从而提高检索的准确性和覆盖面。灵活的转写策略可以应对各种复杂情况。
第八步：提示词模板设计
提示词模板的设计思想是告诉大模型严格按照检索文本块中的内容进行回答，不允许大模型自己编造答案。这样可以保证问答结果的可控性和准确性。
def create_prompt(context, query): return f"根据以下内容回答问题：\n{context}\n\n问题：{query}\n"背景信息补充：良好的提示词设计可以引导大模型生成更符合预期的回答，减少不必要的误差和误解。
第九步：Web界面展示
最后，我使用Gradio创建了一个简单的Web页面，方便展示和测试。
import gradio as gr def demo_interface(): interface = gr.Interface(fn=run_rag_pipeline, inputs="text", outputs="text") interface.launch()背景信息补充：用户友好的界面可以极大地提升用户体验，使系统更易于操作和展示。
第十步：效果评估
主要用两种方法进行评估：
技术指标评估：使用ROUGE和BLEU指标。
人工盲测：设置一百个问题，由人生成相关答案，再让机器生成答案。找第三方评估两个答案哪个更好，统计机器比人好、人比机器好、两者相等的比例。
最终结果显示，76%的结果机器和人一样好，14%人比机器好，10%机器比人好。86%的结果表明机器至少接近人类水平，满足业务需求。
背景信息补充：评估是确保系统性能和效果的关键步骤。通过多种评估方法，可以全面了解系统的优点和不足，为进一步优化提供依据。
结语
通过以上步骤，我成功构建了一个基于LangChain的RAG系统，不仅提升了大模型的回答准确性，还增强了系统的灵活性和可控性。希望这篇文章能为你提供一些有价值的参考和启发。
参考资料
- LangChain Documentation
- M3E Model on GitHub
- FAISS Documentation
- Sentence Transformers
- Gradio Documentation
