Python + LangChain/LangGraph 构建 RAG 检索增强生成系统

📝 本章学习目标：从零搭建生产级 RAG 系统，掌握 LangChain 核心组件 + LangGraph 流程编排，实现高精度、可观测、可调试的检索增强生成，可直接落地企业知识库、文档问答场景。

一、引言：RAG 是什么？为什么必须学

1.1 背景与意义

大模型（LLM）存在知识过时、幻觉、隐私泄露三大痛点，RAG（Retrieval Augmented Generation，检索增强生成）是目前最成熟、最安全的解决方案。

• 不微调模型，低成本接入私有数据
• 实时引用最新文档，杜绝知识过时
• 给出可溯源答案，大幅降低幻觉
• 数据不离开本地 / 私有云，满足合规要求

RAG 已成为 AI 应用开发必备核心技能，90% 企业级 AI 助手均基于 RAG 构建。

1.2 技术栈概览

plaintext

Python + LangChain（组件封装）+ LangGraph（流程编排）+ 向量数据库 + LLM

• LangChain：文档加载、切分、向量化、检索、生成全流程工具
• LangGraph：构建循环、条件判断、可观测的 RAG 工作流
• RAG 核心流程：加载文档 → 文档切分 → 向量化入库 → 用户查询 → 检索 → 生成答案

二、核心概念解析

2.1 RAG 基础架构

标准 RAG 五步法：

1. Indexing（索引）：文档加载 → 切分 → 向量化 → 存入向量库
2. Retrieval（检索）：用户问题向量化 → 匹配向量库 → 召回相关文档
3. Generation（生成）：问题 + 召回文档 → LLM → 精准答案

2.2 核心组件说明

表格

2.3 技术架构图

plaintext

┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 文档源 │────▶│ 构建索引库 │────▶│ 向量数据库 │ └─────────────┘ └─────────────┘ └─────────────┘ │ ┌─────────────┐ ┌─────────────┐ │ │ 用户问题 │────▶│ 检索相关文档 │◀──────────────┘ └─────────────┘ └─────────────┘ │ ▼ ┌─────────────────┐ │ LLM 生成答案 │ └─────────────────┘ │ ▼ ┌─────────────────┐ │ 返回带溯源答案 │ └─────────────────┘

三、环境搭建与依赖安装

3.1 环境要求

• Python 3.10+
• 网络可访问 LLM API（或本地 Ollama）

3.2 安装依赖

bash

运行

# 核心库 pip install langchain langchain-community langgraph # 向量数据库（轻量级本地方案） pip install faiss-cpu chromadb # 文档加载（根据需求选择） pip install pypdf python-docx # PDF/Word pip install beautifulsoup4 # 网页 # LLM 接入（以通义千问/DeepSeek/Ollama 为例） pip install dashscope langchain-openai

3.3 API 配置（以阿里云通义千问为例）

python

运行

import os from langchain_community.embeddings import DashScopeEmbeddings from langchain_community.llms import Tongyi # 配置 API Key os.environ["DASHSCOPE_API_KEY"] = "你的通义千问 API Key" # 初始化 Embedding 模型 embeddings = DashScopeEmbeddings(model="text-embedding-v1") # 初始化 LLM 模型 llm = Tongyi(model_name="qwen-turbo", temperature=0.01)

本地方案：可使用Ollama运行本地 LLM + Embedding，无需 API Key，完全离线。

四、基础版 RAG（LangChain 实现）

4.1 步骤 1：准备测试文档

创建test.txt：

plaintext

LangChain 是一个用于开发大语言模型应用的框架。 LangGraph 是 LangChain 生态中的流程编排工具，用于构建状态化、多步骤的 LLM 应用。 RAG 是检索增强生成，通过外部知识库提升大模型回答准确性。 向量数据库用于存储文本向量，支持相似度检索。

4.2 步骤 2：构建索引库

python

运行

from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS # 1. 加载文档 loader = TextLoader("test.txt", encoding="utf-8") documents = loader.load() # 2. 文档切分（核心：避免文本过长/过短） text_splitter = RecursiveCharacterTextSplitter( chunk_size=200, # 每个块大小 chunk_overlap=50, # 块重叠，保证语义完整 ) texts = text_splitter.split_documents(documents) # 3. 构建向量库 vectorstore = FAISS.from_documents(texts, embeddings) # 4. 创建检索器（召回最相关的 2 条文档） retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

4.3 步骤 3：构建检索 + 生成链

python

运行

from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate # 自定义 Prompt（关键：减少幻觉） prompt_template = """ 你是一个专业的AI助手，请根据以下参考资料回答问题。 如果参考资料中没有答案，请直接回答："我没有找到相关信息"，不要编造内容。 参考资料：{context} 问题：{question} """ PROMPT = PromptTemplate( template=prompt_template, input_variables=["context", "question"] ) # 构建 RAG 链 qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=retriever, chain_type_kwargs={"prompt": PROMPT} )

4.4 步骤 4：测试问答

python

运行

# 测试问题 query = "LangGraph 是什么？" result = qa_chain.run(query) print("答案：", result)

输出结果：

plaintext

答案：LangGraph 是 LangChain 生态中的流程编排工具，用于构建状态化、多步骤的 LLM 应用。

五、进阶版 RAG（LangGraph 实现）

5.1 为什么用 LangGraph

基础 RAG 是线性流程，无法处理：

• 检索结果差 → 重新检索
• 问题复杂 → 多轮检索
• 需要判断 → 条件分支
• 需要调试 → 状态观测

LangGraph 支持状态管理、条件跳转、循环重试，构建工业级 RAG。

5.2 核心：定义图状态

python

运行

from typing import List, TypedDict from langchain.schema import Document # 定义 RAG 状态（全局共享数据） class GraphState(TypedDict): question: str # 用户问题 documents: List[Document] # 检索到的文档 answer: str # 生成的答案

5.3 定义 RAG 节点函数

python

运行

# 节点 1：检索文档 def retrieve(state: GraphState): question = state["question"] documents = retriever.invoke(question) return {"documents": documents} # 节点 2：生成答案 def generate(state: GraphState): question = state["question"] documents = state["documents"] # 拼接文档内容 context = "\n".join([doc.page_content for doc in documents]) prompt = PROMPT.format(context=context, question=question) # LLM 生成 answer = llm.invoke(prompt) return {"answer": answer}

5.4 构建 LangGraph 工作流

python

运行

from langgraph.graph import StateGraph, START, END # 1. 初始化流程图 workflow = StateGraph(GraphState) # 2. 添加节点 workflow.add_node("retrieve", retrieve) workflow.add_node("generate", generate) # 3. 构建流程 workflow.add_edge(START, "retrieve") workflow.add_edge("retrieve", "generate") workflow.add_edge("generate", END) # 4. 编译图 app = workflow.compile()

5.5 运行图 RAG

python

运行

# 执行 inputs = {"question": "RAG 是什么？"} output = app.invoke(inputs) # 输出结果 print("答案：", output["answer"]) print("\n检索到的文档：") for doc in output["documents"]: print("-", doc.page_content)

六、高级特性：带判断 / 重试的 RAG（LangGraph 最强功能）

6.1 增加答案校验节点

python

运行

# 节点 3：校验答案是否有效 def check_answer(state: GraphState): answer = state["answer"] # 如果答案是无信息，返回重新检索 if "没有找到相关信息" in answer: return "retrieve" else: return END

6.2 构建带条件判断的流程

python

运行

workflow = StateGraph(GraphState) workflow.add_node("retrieve", retrieve) workflow.add_node("generate", generate) workflow.add_edge(START, "retrieve") workflow.add_edge("retrieve", "generate") # 条件跳转：有效→结束，无效→重试检索 workflow.add_conditional_edges( "generate", check_answer, { "retrieve": "retrieve", END: END } ) app = workflow.compile()

6.3 测试无答案场景

python

运行

inputs = {"question": "Python 是什么时候发明的？"} output = app.invoke(inputs) print("答案：", output["answer"])

输出：我没有找到相关信息

七、生产级优化方案

7.1 检索优化（提升准确率核心）

1. 切分优化：调整 chunk_size + overlap，保证语义完整
2. 检索策略：
   • 相似度检索 + 关键词检索（混合检索）
   • 召回后重排序（Rerank 模型）
3. 向量库：大数据量使用 Pinecone、Milvus

7.2 生成优化

• 低temperature（0.01~0.1），减少幻觉
• 严格 Prompt 约束：禁止编造、强制引用
• 答案添加引用来源，支持溯源

7.3 部署优化

• 向量库持久化存储
• 异步接口提升并发
• 缓存高频问题答案
• 日志监控检索准确率

八、完整项目代码（可直接运行）

python

运行

# -*- coding: utf-8 -*- # RAG + LangChain + LangGraph 完整代码 import os from typing import List, TypedDict from langchain.schema import Document from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS from langchain_community.embeddings import DashScopeEmbeddings from langchain_community.llms import Tongyi from langchain.prompts import PromptTemplate from langgraph.graph import StateGraph, START, END # ===================== 1. 配置 ===================== os.environ["DASHSCOPE_API_KEY"] = "你的API Key" embeddings = DashScopeEmbeddings(model="text-embedding-v1") llm = Tongyi(model_name="qwen-turbo", temperature=0.01) # ===================== 2. 构建索引 ===================== loader = TextLoader("test.txt", encoding="utf-8") documents = loader.load() text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=50) texts = text_splitter.split_documents(documents) vectorstore = FAISS.from_documents(texts, embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 2}) # ===================== 3. Prompt ===================== prompt_template = """ 根据参考资料回答问题，无答案则回复：我没有找到相关信息。 参考资料：{context} 问题：{question} """ PROMPT = PromptTemplate(template=prompt_template, input_variables=["context", "question"]) # ===================== 4. LangGraph 流程 ===================== class GraphState(TypedDict): question: str documents: List[Document] answer: str def retrieve(state): question = state["question"] documents = retriever.invoke(question) return {"documents": documents} def generate(state): question = state["question"] documents = state["documents"] context = "\n".join([d.page_content for d in documents]) answer = llm.invoke(PROMPT.format(context=context, question=question)) return {"answer": answer} # 构建图 workflow = StateGraph(GraphState) workflow.add_node("retrieve", retrieve) workflow.add_node("generate", generate) workflow.add_edge(START, "retrieve") workflow.add_edge("retrieve", "generate") workflow.add_edge("generate", END) app = workflow.compile() # ===================== 5. 测试 ===================== if __name__ == "__main__": query = "LangGraph 用来做什么？" result = app.invoke({"question": query}) print("问题：", query) print("答案：", result["answer"])

九、常见问题与解决方案

9.1 常见问题

表格

9.2 关键注意事项

1. 文档质量决定 RAG 上限
2. Prompt 是减少幻觉的关键
3. LangGraph 适合复杂业务流程
4. 优先轻量级方案，再逐步升级

十、本章小结

10.1 核心要点回顾

1. RAG 流程：加载 → 切分 → 向量化 → 检索 → 生成
2. LangChain：提供全流程组件，快速搭建基础 RAG
3. LangGraph：状态管理 + 流程编排，构建工业级 RAG
4. 优化核心：检索精度 + Prompt 约束 + 模型选择

10.2 学习路径

1. 入门：运行基础 RAG 示例
2. 进阶：掌握 LangGraph 条件流程
3. 实战：接入 PDF / 多文档 / 企业知识库
4. 生产：优化检索、部署、监控