一、LangGraph 构建图 RAG 的五步曲在 LangGraph 中实现 RAG标准流程分为以下五步读取加载原始文件。切分将长文档切分成小块。向量化与存储将文本块转化为向量并存入向量数据库。构建检索器创建一个可以根据问题去向量库搜答案的工具。定义图节点用 LangGraph 编排“检索”和“回答”的先后顺序。划重点前 4 步都是 LangChain 的原生能力只有第 5 步是 LangGraph 的专属魔法。它赋予了我们对 RAG 流程的绝对控制力二、核心实战代码下面的代码完整展示了从加载 TXT 到使用 LangGraph 生成答案的全过程。from typing import TypedDict from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_core.prompts import ChatPromptTemplate import os from dotenv import load_dotenv load_dotenv() # 步骤 1加载文档 # 实际项目中这里可以换成 PDF/Word/HTML 等 loader TextLoader(test.txt, encodingutf-8) documents loader.load() # 步骤 2分割文档 text_splitter RecursiveCharacterTextSplitter( chunk_size100, chunk_overlap20 ) texts text_splitter.split_documents(documents) # 步骤 3向量化并存储 embeddings OpenAIEmbeddings( modelos.getenv(ZHIPU_EMBEDDING_MODEL), api_keyos.getenv(ZHIPU_API_KEY), base_urlos.getenv(ZHIPU_EMBEDDING_BASE_URL) ) # 持久化存储到本地目录下次启动无需重新计算 db Chroma.from_documents(texts, embeddings, persist_directory./chroma_db) # 步骤 4构建检索器 # search_kwargs{k: 2} 表示每次返回最匹配的 2 个文本块 retriever db.as_retriever(search_kwargs{k: 2}) # 初始化 LLM llm ChatOpenAI( modeldeepseek-chat, api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL), temperature0 ) # 步骤 5定义图状态与节点 class RAGState(TypedDict): question: str # 用户提问 context: str # 检索到的背景资料 answer: str # 最终答案 def retrieve_node(state: RAGState) - dict: 检索节点只负责查资料不负责回答 print(f 正在检索关于: {state[question]} 的资料...) retrieved_docs retriever.invoke(state[question]) # 把检索到的多个文档拼成一个字符串 context_text \n\n.join([doc.page_content for doc in retrieved_docs]) return {context: context_text} def answer_node(state: RAGState) - dict: 回答节点拿着原始问题和检索到的资料让 LLM 总结答案 print( LLM 正在根据资料生成回答...) prompt ChatPromptTemplate.from_template( 使用以下上下文回答用户问题仅基于上下文内容不要编造。 如果上下文没有相关信息回答“不知道”。 上下文 {context} 问题 {question} ) response llm.invoke(prompt.format(contextstate[context], questionstate[question])) return {answer: response.content} # 构建并运行图 workflow StateGraph(RAGState) workflow.add_node(retrieve, retrieve_node) workflow.add_node(answer, answer_node) workflow.set_entry_point(retrieve) # 强行规定必须先检索再回答这就是 Graph 的绝对控制力 workflow.add_edge(retrieve, answer) workflow.add_edge(answer, END) rag_app workflow.compile() # 测试运行 if __name__ __main__: print(--- 提问 1 ---) result1 rag_app.invoke({question: 我想退款大概几天能收到钱}) print(f 回答: {result1[answer]}\n) print(--- 提问 2 (超出知识库范围) ---) result2 rag_app.invoke({question: 你们公司CEO是谁}) print(f 回答: {result2[answer]}) 为什么要把 RAG 拆成两个节点示例中我们定义了retrieve_node和answer_node。虽然逻辑上合并成一个节点也能跑但拆开有巨大的好处极度清晰一眼就能看出数据流向是问题 - 检索器 - 上下文 - LLM。方便扩展以后如果想在检索前加个“敏感词拦截节点”或者在回答后加个“答案合规性检查节点”只需要在图中插入新节点加一条边即可完全不用改原有代码三、RAG 基础设施数据加载与处理全指南上面代码的前 4 步是 RAG 的基础设施。针对这一块这里做一份详细的避坑与扩展指南。1. 文件读取你不知道的 Loader 宝库LangChain 支持几百种数据源的读取除了上面的.txt以下是常用的加载器汇总使用时需安装对应的包如pip install unstructured等本地文件类纯文本/代码TextLoader(支持 .txt, .md, .py 等)PDFPyPDFLoader(基础提取推荐) /UnstructuredPDFLoader(高级解析)OfficeUnstructuredWordLoader(.docx) /UnstructuredExcelLoader(.xlsx)HTML/XMLBSHTMLLoader/XMLLoaderCSV/JSONCSVLoader(按行转为文档) /JSONLoader(支持 jq 表达式提取)网络数据源网页抓取WebBaseLoader(支持传入 URL 列表批量抓取)维基百科WikipediaLoader(直接搜索并返回词条内容)数据库与云存储云端S3FileLoader(AWS) /GCSFileLoader(谷歌云)关系型库通过DataFrameLoader结合pandas的 SQL 查询直接将数据库内容转为文档。批量加载技巧如果想加载一整个文件夹下的所有 PDF不要用 for 循环使用DirectoryLoaderloader DirectoryLoader(./docs/, glob*.pdf, loader_clsPyPDFLoader)2. 文本切分Chunk 的艺术大模型有 Token 限制文档必须切碎。from langchain_text_splitters import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( separator\n\n, # 分隔符优先按双换行切 chunk_size100, # 每块最大字符数 chunk_overlap20, # 块间重叠字符数防止上下文被生硬截断 length_functionlen, # 长度计算函数 )推荐使用RecursiveCharacterTextSplitter它是CharacterTextSplitter的升级版。它会先尝试按段落\n\n切如果块还是太大就尝试按句子\n切再大就按空格切。这种基于语义层级的切分方式效果远好于暴力截断。3. 向量化与向量库记忆的物理载体关于 Embedding 模型的选择除了商业 API如智谱、OpenAI强烈推荐本地部署开源模型省钱且隐私好from langchain_community.embeddings import OllamaEmbeddings # 使用 Ollama 本地运行 nomic-embed-text 模型 embeddings OllamaEmbeddings( modelnomic-embed-text, base_urlhttp://192.168.1.100:11434 # 局域网地址本机可省略 )关于 Chroma 向量库的避坑指南首次创建与持久化使用Chroma.from_documents(..., persist_directory./chroma_db)时会将生成的向量存入本地磁盘。二次读取非常重要如果代码已经跑过一次向量库已经建立。下次启动时千万不要再调用from_documents否则会覆盖原有库应该这样读取# 正确的二次读取方式 db Chroma( persist_directory./chroma_db, embedding_functionembeddings # ⚠️ 传入的模型必须和创建时一模一样 )*(注除了 Chroma生产环境还常用 FAISS、Milvus、Qdrant 等用法大同小异)*4. 构建检索器精准召回检索器不仅可以控制返回数量还可以控制相似度门槛。retriever db.as_retriever( search_kwargs{ k: 5, # 返回最相似的前 5 个结果 score_threshold: 0.5 # ⚠️ 仅返回相似度得分大于 0.5 的结果 } )加入score_threshold是一个极好的习惯可以有效过滤掉那些八竿子打不着的垃圾文本块防止它们干扰大模型的判断。