基于LangChain与向量数据库构建RAG智能问答系统实践

📅 2026/7/9 14:31:00 👁️ 阅读次数 📝 编程学习
基于LangChain与向量数据库构建RAG智能问答系统实践

1. 项目概述与核心价值

最近在折腾一个内部知识库的智能问答功能,核心需求是让大语言模型(LLM)能准确回答关于公司内部文档、产品手册等非公开信息的问题。直接让模型“凭空”回答,结果往往不尽人意,要么是“根据我的知识截止到...”,要么就是一本正经地胡说八道,也就是所谓的“幻觉”。为了解决这个问题,检索增强生成(RAG)成了我的首选方案。简单来说,RAG就是让模型在回答问题前,先去一个专属的“资料库”里查查相关资料,然后基于查到的资料来组织答案。这个“资料库”,通常就是向量数据库。整个过程,我用LangChain这个开发框架来串联,它把文档处理、向量化、检索、提示工程这些繁琐的步骤都封装好了,让开发者能更专注于业务逻辑。这篇文章,我就把自己从零搭建一个简单RAG系统的完整过程、踩过的坑以及一些实用技巧记录下来,希望能给同样想入手的朋友一个清晰的参考。

2. 技术选型与整体架构设计

2.1 为什么选择LangChain + 向量数据库的组合

在开始敲代码之前,明确技术选型的理由很重要。市面上框架不少,为什么是LangChain?首先,它不是一个“黑盒”产品,而是一个高度模块化的开源框架。这意味着它提供了构建基于LLM应用所需的各种“乐高积木”,比如文档加载器、文本分割器、向量存储接口、各种链(Chain)的抽象。你可以自由组合这些模块,定制性极强。其次,它的生态非常活跃,集成了海量的第三方工具和模型,从OpenAI、Anthropic到本地部署的Ollama,从Chroma、Weaviate到Pinecone,几乎你能想到的组件都能找到对应的集成。这对于快速原型验证和未来技术栈迁移非常友好。

至于向量数据库,它是RAG系统的“记忆中枢”。传统数据库按行、列存储,擅长精确匹配(比如WHERE name='张三'),但对于“帮我找和这个概念相似的文档”这种模糊语义查询就无能为力了。向量数据库将文本、图像等数据通过嵌入模型转化为高维向量(一组数字),然后存储这些向量。查询时,也将问题转化为向量,并通过计算向量间的“距离”(如余弦相似度)来找到最相似的向量,从而定位到最相关的文本片段。这个过程完美契合了语义检索的需求。我选择ChromaDB作为起步,主要是因为它轻量、易用,可以纯内存运行也可以持久化,非常适合学习和中小型项目初期。

2.2 RAG系统的工作流程拆解

一个最小化的RAG系统,其工作流可以清晰地分为两个阶段:索引(Indexing)检索与生成(Retrieval & Generation)

索引阶段(线下进行)

  1. 文档加载:从各种来源(PDF、Word、网页、数据库)把原始文档读进来。
  2. 文本分割:把大文档切成小块。这一步是关键,切得太大会包含无关信息,影响检索精度;切得太碎会丢失上下文,影响生成质量。
  3. 向量化:使用嵌入模型将每个文本块转化为一个向量。
  4. 存储:将文本块和对应的向量一起存入向量数据库。

检索与生成阶段(线上响应)

  1. 用户提问:用户输入一个问题。
  2. 问题向量化:使用同样的嵌入模型将问题转化为向量。
  3. 语义检索:在向量数据库中查找与问题向量最相似的几个文本块(即上下文)。
  4. 提示构建:将问题、检索到的上下文以及系统指令,按照一定格式组装成一个完整的提示。
  5. LLM生成:将组装好的提示发送给LLM,由LLM生成最终答案。

LangChain的价值就在于,它用清晰的抽象(如DocumentLoader,TextSplitter,VectorStore,RetrievalQA链)把上述每一步都封装好了,我们只需要进行配置和组装。

3. 环境准备与核心工具部署

3.1 Python环境与依赖库安装

我习惯使用conda管理Python环境,能有效避免包冲突。首先创建一个新的环境。

conda create -n rag-demo python=3.10 conda activate rag-demo

接下来安装核心依赖。除了LangChain,我们还需要嵌入模型和向量数据库的客户端。这里我选择OpenAItext-embedding-ada-002模型作为嵌入模型,因为它效果稳定且API易用。LLM也先用OpenAI的gpt-3.5-turbo来演示。当然,后续完全可以替换为开源模型。

pip install langchain langchain-community langchain-openai pip install chromadb pip install tiktoken # 用于OpenAI模型的文本分词计数 pip install pypdf # 用于读取PDF文档

注意:使用OpenAI API会产生费用。请务必在OpenAI平台创建账户并获取API密钥,同时设置好用量监控。对于本地测试或对成本敏感的场景,完全可以替换为免费的本地嵌入模型(如sentence-transformers)和本地LLM(通过Ollama等工具部署)。

3.2 嵌入模型与LLM的配置

安装好后,我们需要在代码中配置API密钥和要使用的模型。永远不要将API密钥硬编码在代码中或上传到GitHub。最佳实践是使用环境变量。

# 在终端中设置环境变量(Linux/macOS) export OPENAI_API_KEY="你的-api-key-here"

在Python脚本中,我们可以这样加载和使用:

import os from langchain_openai import OpenAIEmbeddings, ChatOpenAI # 从环境变量读取API Key openai_api_key = os.getenv("OPENAI_API_KEY") if not openai_api_key: raise ValueError("请设置 OPENAI_API_KEY 环境变量") # 初始化嵌入模型 embeddings = OpenAIEmbeddings( model="text-embedding-ada-002", openai_api_key=openai_api_key ) # 初始化LLM llm = ChatOpenAI( model="gpt-3.5-turbo", temperature=0, # 温度设为0,使输出更确定、更少创造性,适合事实问答 openai_api_key=openai_api_key )

这里将temperature设为0,是为了让模型在回答基于事实的问题时更加严谨和一致,减少“编造”的可能性。对于创意写作类任务,可以适当调高。

4. 文档处理与向量知识库构建

4.1 文档加载与文本分割策略

假设我们有一个名为company_handbook.pdf的产品手册。LangChain提供了多种文档加载器,位于langchain_community.document_loaders模块下。

from langchain_community.document_loaders import PyPDFLoader # 加载PDF文档 loader = PyPDFLoader("./docs/company_handbook.pdf") documents = loader.load() print(f"加载了 {len(documents)} 页文档")

加载后的documents是一个列表,每个元素是一页的内容。但一页内容可能还是太长,我们需要进行分割。这里我选择RecursiveCharacterTextSplitter,它是一个递归字符分割器,会优先尝试用双换行符\n\n分割,不行再用单换行符\n,再不行用句号.,以此类推。这种方式比简单的按固定字符数切割更能保持语义完整性。

from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个文本块的最大字符数 chunk_overlap=50, # 相邻块之间的重叠字符数 length_function=len, # 计算长度的函数 separators=["\n\n", "\n", "。", "!", "?", ";", ",", " ", ""] # 分割符优先级 ) # 执行分割 split_docs = text_splitter.split_documents(documents) print(f"原始文档被分割成了 {len(split_docs)} 个文本块")

参数选择的经验

  • chunk_size:需要权衡。太小(如100)会丢失上下文,太大(如2000)会引入噪声。对于通用文档,500-1000是个不错的起点。你可以根据你的文档类型(技术文档、小说、法律条文)和后续检索效果进行调整。
  • chunk_overlap:设置重叠是为了避免一个完整的句子或概念被硬生生切到两个块里,导致检索时只拿到一半信息。50-100个字符的重叠通常足够。

4.2 向量化与存储到ChromaDB

有了文本块,接下来就是用嵌入模型把它们变成向量,并存入ChromaDB。

from langchain.vectorstores import Chroma # 指定持久化目录 persist_directory = './chroma_db' # 创建向量存储。这将自动完成:1.为每个split_doc生成嵌入向量 2.存入ChromaDB vectordb = Chroma.from_documents( documents=split_docs, embedding=embeddings, persist_directory=persist_directory ) # 持久化到磁盘 vectordb.persist() print(f"向量知识库已创建并保存至 {persist_directory}")

这个过程可能会花费一些时间,取决于文档数量和嵌入模型的API速度。Chroma.from_documents方法封装了向量化和存储的整个过程。persist_directory参数指定了数据库存储的本地路径,下次启动时可以直接加载,无需重新计算嵌入向量,这能节省大量时间和API费用。

5. 检索链的组装与优化

5.1 基础检索问答链的实现

知识库建好了,现在来实现问答部分。LangChain提供了RetrievalQA链,它把检索器、LLM和提示模板打包在一起,使用起来非常简单。

from langchain.chains import RetrievalQA # 首先,从持久化目录加载已有的向量数据库 vectordb = Chroma( persist_directory=persist_directory, embedding_function=embeddings ) # 将向量数据库转换为检索器 retriever = vectordb.as_retriever() # 创建RetrievalQA链 qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", # 最常用的类型,将所有检索到的上下文“塞”进提示 retriever=retriever, return_source_documents=True, # 返回检索到的源文档,便于调试 verbose=True # 打印链的详细执行过程,开发时很有用 ) # 进行提问 question = "我们公司产品的核心优势是什么?" result = qa_chain.invoke({"query": question}) print("答案:", result["result"]) print("\n--- 参考来源 ---") for i, doc in enumerate(result["source_documents"]): print(f"[来源{i+1}] {doc.page_content[:200]}...") # 打印前200字符

chain_type="stuff"是最直接的方式,它把检索到的所有相关文档片段(上下文)和问题一起拼接成一个大的提示,送给LLM。这种方式简单有效,但受限于LLM的上下文窗口长度(例如,GPT-3.5-turbo是16K),如果检索到的文档总长度超过限制就会报错。

5.2 检索策略的深度优化

基础的检索可能不够精准。LangChain的检索器提供了多种参数来优化:

# 优化后的检索器配置 retriever = vectordb.as_retriever( search_type="similarity", # 默认,相似度搜索。还可选 "mmr" (最大边际相关性) search_kwargs={ "k": 4 # 检索返回的文档数量。不是越多越好,需要实验。 } ) # 使用MMR搜索,在保证相关性的同时增加多样性,避免返回内容雷同 retriever_mmr = vectordb.as_retriever( search_type="mmr", search_kwargs={ "k": 4, "fetch_k": 10 # MMR算法会先取fetch_k个文档,再从中精选k个 } )
  • k值调优:这是最重要的参数之一。k=3意味着每次检索只取最相似的3个片段。如果答案分散在多个片段中,k太小可能找不全;k太大则会引入不相关噪声,还可能触发上下文长度限制。需要通过测试集来调整。
  • MMR搜索:适用于用户问题可能对应多个子主题的情况。比如问“介绍一下公司的市场和产品”,similarity可能返回4段都是讲产品的,而mmr可能会返回2段讲产品,1段讲市场,1段讲公司愿景,答案会更全面。

5.3 提示工程与上下文管理

默认的提示模板可能不够理想。我们可以自定义提示,明确指示模型如何利用上下文。

from langchain.prompts import PromptTemplate # 自定义提示模板 prompt_template = """请根据以下上下文信息来回答问题。如果你不知道答案,就老实说不知道,不要编造。 上下文: {context} 问题:{question} 请给出准确、基于上下文的答案:""" PROMPT = PromptTemplate( template=prompt_template, input_variables=["context", "question"] ) # 使用自定义提示创建QA链 qa_chain_custom = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=retriever, chain_type_kwargs={"prompt": PROMPT}, # 传入自定义提示 return_source_documents=True )

这个模板比默认的更强调“基于上下文”和“不要编造”,对于减轻幻觉有一定帮助。更复杂的场景还可以使用chain_type="map_reduce""refine",它们能处理更长的文档,但延迟和API调用次数也会增加。

6. 系统测试、评估与效果调优

6.1 构建测试集与效果评估

系统搭起来了,但效果到底怎么样?不能只靠手动问几个问题。建立一个简单的测试集是必要的。

test_questions = [ "公司的成立时间是哪一年?", "产品A的主要功能有哪些?", "售后服务政策是怎样的?", "请总结一下公司的企业文化。" ] for q in test_questions: print(f"\n问题:{q}") result = qa_chain_custom.invoke({"query": q}) print(f"答案:{result['result'][:300]}...") # 限制输出长度 # 这里可以人工判断答案是否正确,或与标准答案对比 # 更高级的做法可以计算BLEU、ROUGE分数或使用LLM-as-a-Judge来评估

评估RAG系统是个复杂课题。初期可以人工评估,关注几个点:答案相关性(是否答非所问)、答案正确性(是否基于上下文且事实正确)、上下文引用率(是否有效利用了检索到的内容)以及幻觉程度(是否编造了不存在的信息)。

6.2 常见问题排查与调优技巧

在实际操作中,我遇到了不少典型问题,这里分享排查思路:

  1. 检索不到相关内容

    • 检查分割:文本块是否太小或太大?用print(split_docs[0].page_content)看看前几个块的内容是否完整。
    • 检查嵌入:确保索引和查询使用的是同一个嵌入模型。不同模型生成的向量空间不同,无法直接比较。
    • 检查检索器k:尝试增大k值。
    • 检查问题表述:用户的问题是否太模糊或用了知识库里没有的术语?可以尝试对用户问题进行查询重写扩展。LangChain提供了MultiQueryRetriever等工具,能从不同角度生成多个查询去检索。
  2. 检索到内容但答案质量差

    • 上下文过长或噪声大:减小k值或尝试mmr搜索。或者,在将上下文送入LLM前,增加一个重排序步骤,用更精细的模型对检索结果再次排序,把最相关的放在前面。
    • 提示模板不佳:优化你的提示词,明确指令。比如加入“如果上下文里没有明确提到,请回答‘根据提供的信息,无法确定...’”。
    • LLM能力或参数问题:尝试换一个更强的模型(如gpt-4),或调整temperature等参数。
  3. 响应速度慢

    • 嵌入模型延迟:如果使用云端API,网络是主要瓶颈。考虑使用更快的模型或本地嵌入模型。
    • LLM生成延迟:同上,或考虑对答案进行缓存。
    • 向量检索延迟:对于百万级以上的数据量,纯内存的Chroma可能变慢,需要考虑专业的向量数据库(如Weaviate, Qdrant)及其索引优化。

一个实用的调优流程:先固定LLM和提示词,集中精力优化检索部分(分割策略、检索参数)。待检索到的上下文质量稳定后,再去优化提示工程和LLM调用。

7. 进阶考量与生产化部署

7.1 多文档管理与元数据过滤

真实的知识库往往包含多种类型、多个来源的文档。我们可以在存储时附加元数据,检索时进行过滤。

# 在分割文档后,为每个块添加元数据 for i, doc in enumerate(split_docs): doc.metadata = { "source": "company_handbook.pdf", "page": doc.metadata.get("page", 0), # 保留原始页号 "doc_type": "manual" } # 创建向量存储时,元数据会自动保存 vectordb = Chroma.from_documents(split_docs, embeddings, persist_directory=persist_directory) # 检索时,可以基于元数据过滤 retriever = vectordb.as_retriever( search_kwargs={ "k": 3, "filter": {"doc_type": "manual"} # 只检索手册类文档 } )

这样,当用户问“某个合同里的条款”,你可以过滤只搜索doc_type"contract"的文档,避免其他无关文档干扰。

7.2 对话历史与多轮问答

基础的RetrievalQA是无状态的。要实现带历史的多轮对话,需要使用ConversationalRetrievalChain

from langchain.chains import ConversationalRetrievalChain from langchain.memory import ConversationBufferMemory memory = ConversationBufferMemory( memory_key="chat_history", return_messages=True, output_key='answer' # 与链的输出键匹配 ) conversational_chain = ConversationalRetrievalChain.from_llm( llm=llm, retriever=retriever, memory=memory, chain_type="stuff", verbose=True ) # 第一轮 result1 = conversational_chain.invoke({"question": "我们产品的优势是什么?"}) print(result1['answer']) # 第二轮,模型能记住之前的对话 result2 = conversational_chain.invoke({"question": "能再详细说说其中的一点吗?"}) print(result2['answer'])

这个链会自动将之前的问答历史作为上下文的一部分,让LLM能理解指代关系(如“其中的一点”)。

7.3 向生产环境过渡的思考

当Demo跑通,考虑上线时,有几个关键点:

  • 数据安全与隐私:如果使用云端API,确保传输和存储的数据符合合规要求。对于高度敏感数据,必须使用本地部署的嵌入模型和LLM。
  • 性能与扩展性:ChromaDB在数据量极大时可能成为瓶颈。评估是否需要迁移到分布式向量数据库(如Milvus, Weaviate Cluster)。
  • 监控与日志:记录每一次问答的查询、检索到的源文档、生成的答案、耗时和Token使用量。这对于排查问题、优化成本和评估效果至关重要。
  • 流水线自动化:建立自动化的文档更新流水线。当源文档更新时,能自动触发重新索引,确保知识库的时效性。

构建RAG系统是一个迭代的过程,没有一劳永逸的“最佳配置”。核心在于理解每个环节(文档加载、分割、嵌入、检索、提示、生成)的作用和可调参数,然后基于你的具体数据、问题和业务目标,进行持续的测试、评估和调优。从这个小Demo出发,你已经掌握了最核心的链路,接下来可以根据实际需求,在各个节点上替换更强大的组件,比如尝试不同的文本分割器、使用ColBERT进行更精准的检索、或者集成GraphRAG来处理复杂逻辑关系,让这个系统变得越来越智能。