最近在尝试将大语言模型应用到实际业务中时很多开发者都会遇到一个核心难题如何让模型理解并处理我们私有的、最新的数据单纯依赖模型的预训练知识往往无法满足企业级应用对精准性和时效性的要求。这正是 RAG检索增强生成技术大放异彩的场景而 LangChain 作为构建此类应用的顶级框架其学习曲线却让不少人望而却步。本文旨在为你提供一份从零到一的 LangChain 实战指南不仅涵盖构建 RAG 应用的核心流程还会深入 LangGraph 构建多智能体系统的进阶玩法。内容基于最新的 LangChain 1.3 版本整合了官方文档精华与一线开发经验力求让你避开那些文档中未明说、但实践中必踩的“坑”。无论你是想快速搭建一个智能问答助手还是设计复杂的多智能体协作流程这里都有完整的代码示例和清晰的配置思路。1. 背景与核心概念为什么是 LangChain 和 RAG在深入代码之前我们先厘清几个关键概念这能帮助你理解后续每一步设计的初衷。1.1 什么是 RAGRAG 的全称是 Retrieval-Augmented Generation即检索增强生成。你可以把它想象成一个“超级研究员”检索当用户提出一个问题时系统不会直接让大模型“凭空想象”而是先从你的知识库如文档、数据库、网页中查找最相关的信息片段。增强将这些查找到的准确信息连同用户的问题一起作为“上下文”提交给大模型。生成大模型基于这些可靠的上下文信息生成最终的回答。这样做的好处显而易见回答更准确、可追溯知道答案来自哪份文档、并且能有效缓解大模型的“幻觉”问题即编造不存在的信息。1.2 LangChain 扮演什么角色LangChain 不是一个具体的 AI 模型而是一个框架。它就像一套功能强大的“乐高积木”提供了标准化、模块化的组件让你能够轻松地将大语言模型如 OpenAI GPT、通义千问、向量数据库如 Chroma、Milvus、文档加载器、记忆模块等连接起来构建端到端的应用。它的核心价值在于组件化提供了Document Loaders、Text Splitters、Vector Stores、Chains、Agents等可复用的模块。编排能力通过LCELLangChain 表达式语言或Chain对象可以像搭积木一样将多个步骤串联成复杂的工作流。生态丰富集成了数百种工具、数据源和模型提供商避免了重复造轮子。1.3 LangGraph 又是什么LangGraph 是 LangChain 框架中用于构建有状态、多智能体应用的库。如果说普通的 Chain 是一个线性流水线那么 LangGraph 允许你构建带有循环、条件分支的图结构。智能体Agent可以在图中根据当前状态决定下一步行动并与其他智能体或工具进行交互。这非常适合需要长期规划、工具调用和多角色协作的复杂任务比如自动化的数据分析流水线或游戏 NPC 对话系统。2. 环境准备与版本说明在开始实战前请确保你的开发环境已就绪。本文示例以 Python 为主要语言重点演示配置思路版本需要根据你的项目实际情况调整。2.1 基础环境操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文命令以 Linux/macOS 为例Windows 用户可在 PowerShell 或 WSL 中运行。Python 版本建议使用 Python 3.10 或 3.11。LangChain 新版本对 Python 3.12 的支持可能仍在完善中。包管理工具使用pip或conda。推荐在虚拟环境中操作。2.2 核心依赖安装创建一个新的项目目录并初始化虚拟环境mkdir langchain-rag-demo cd langchain-rag-demo python -m venv venv # 激活虚拟环境 # Linux/macOS: source venv/bin/activate # Windows: # venv\Scripts\activate安装 LangChain 及其相关组件。注意我们这里安装的是包含常用社区集成包的langchain-community和用于构建图的langgraph。pip install -U langchain langchain-community langgraph2.3 模型与向量数据库选择大语言模型你可以选择云端 API如 OpenAI或本地模型。为了演示的通用性我们将同时展示两种方式。使用 OpenAI 需要 API Key。pip install openai向量数据库我们选择轻量级、易于上手的Chroma内存模式。pip install chromadb嵌入模型用于将文本转换为向量。使用 OpenAI 的text-embedding-3-small或开源的BAAI/bge-small-zh-v1.5。这里我们安装sentence-transformers来使用开源模型。pip install sentence-transformers2.4 项目结构预览一个典型的 RAG 项目结构如下我们将逐步填充langchain-rag-demo/ ├── docs/ # 存放你的原始文档PDF, TXT, MD等 ├── data/ # 处理后的数据向量存储等 ├── src/ │ ├── ingest.py # 文档加载、切分、向量化入库脚本 │ ├── rag_chain.py # RAG 核心链 │ └── langgraph_agent.py # LangGraph 多智能体示例 ├── .env # 环境变量存放API Key等 └── requirements.txt # 项目依赖3. 核心流程拆解构建 RAG 的四大步骤一个完整的 RAG 应用可以拆解为四个核心步骤理解每一步是灵活运用 LangChain 的关键。3.1 文档加载与处理这是“知识库”的源头。LangChain 提供了大量的DocumentLoader支持从 PDF、Word、HTML、数据库甚至 YouTube 字幕加载文本。# src/ingest.py 片段 - 文档加载 from langchain_community.document_loaders import PyPDFLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter # 1. 加载文档 loader PyPDFLoader(./docs/your_document.pdf) # 加载PDF # loader TextLoader(./docs/note.txt, encodingutf-8) # 加载文本 documents loader.load() # 2. 分割文本 # 大模型有上下文长度限制必须将长文档切分成小块chunks text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个块的最大字符数 chunk_overlap50, # 块之间的重叠字符避免语义被割裂 separators[\n\n, \n, 。, , , , , , ] # 中文友好分隔符 ) chunks text_splitter.split_documents(documents) print(f原始文档被切分为 {len(chunks)} 个文本块。)关键点chunk_size和chunk_overlap需要根据模型上下文窗口和文档特性调整。重叠太少可能导致信息断裂太多则增加冗余和成本。3.2 向量化与存储将文本块转换为数值向量嵌入并存入向量数据库以便快速检索。# src/ingest.py 片段 - 向量化与存储 from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings from langchain.vectorstores import Chroma import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 # 选择嵌入模型 # 方式一使用 OpenAI API (需要付费但效果好且稳定) # embeddings OpenAIEmbeddings(openai_api_keyos.getenv(OPENAI_API_KEY)) # 方式二使用本地开源模型 (免费首次下载需要时间) embeddings HuggingFaceEmbeddings(model_nameBAAI/bge-small-zh-v1.5) # 创建向量存储 vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./data/chroma_db # 指定持久化目录 ) vectorstore.persist() # 将向量数据库保存到磁盘 print(向量数据库已创建并持久化到 ./data/chroma_db)3.3 检索根据用户问题从向量库中找出最相关的文本块。# 这是一个在链中使用的检索器示例 retriever vectorstore.as_retriever( search_typesimilarity, # 相似度搜索还有 mmr (最大边际相关性) 可兼顾相关性与多样性 search_kwargs{k: 4} # 返回最相关的 4 个块 )关键点search_typemmr在回答需要多角度信息的问题时可能更有优势。3.4 生成将“用户问题”和“检索到的上下文”组合成提示词Prompt发送给大模型生成最终答案。from langchain.prompts import ChatPromptTemplate from langchain.chat_models import ChatOpenAI # 如果是本地模型例如使用 Ollama # from langchain_community.llms import Ollama # 1. 定义提示词模板 template 请根据以下上下文来回答问题。如果你不知道答案就说你不知道不要编造信息。 上下文{context} 问题{question} 请用中文给出有帮助的答案 prompt ChatPromptTemplate.from_template(template) # 2. 初始化大模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) # 本地模型示例llm Ollama(modelqwen2.5:7b) # 3. 组合成链 (这是一个简单的链后续我们会用更强大的 LCEL) from langchain.schema.runnable import RunnablePassthrough from langchain.schema.output_parser import StrOutputParser rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() )4. 完整实战案例构建一个本地知识库问答系统现在我们将上述步骤整合构建一个完整的、可运行的问答系统。4.1 项目初始化与文档准备在项目根目录下创建.env文件存放你的 OpenAI API Key如果使用。OPENAI_API_KEYsk-你的真实key在docs/目录下放入你的知识文档例如product_manual.pdf。4.2 编写文档处理脚本创建src/ingest.py完成从加载到存储的全流程。# src/ingest.py import os from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import Chroma def ingest_documents(): 加载、分割文档并存入向量数据库 # 1. 加载 docs 目录下所有 pdf 文件 loader DirectoryLoader(./docs, glob**/*.pdf, loader_clsPyPDFLoader) raw_documents loader.load() print(f已加载 {len(raw_documents)} 个文档页面。) # 2. 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_size800, chunk_overlap80, separators[\n\n, \n, 。, , , , , , ] ) documents text_splitter.split_documents(raw_documents) print(f文档被切分为 {len(documents)} 个文本块。) # 3. 使用开源嵌入模型 embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-small-zh-v1.5, model_kwargs{device: cpu}, # 如果有 GPU 可改为 cuda encode_kwargs{normalize_embeddings: True} ) # 4. 创建并持久化向量存储 vectorstore Chroma.from_documents( documentsdocuments, embeddingembeddings, persist_directory./data/chroma_db, collection_nameproduct_manual ) print(向量数据库已成功创建并保存。) if __name__ __main__: ingest_documents()运行此脚本python src/ingest.py。首次运行会下载嵌入模型请耐心等待。4.3 编写 RAG 问答链创建src/rag_chain.py实现问答功能。# src/rag_chain.py import os from dotenv import load_dotenv from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import Chroma from langchain.prompts import ChatPromptTemplate from langchain.chat_models import ChatOpenAI from langchain.schema.runnable import RunnablePassthrough from langchain.schema.output_parser import StrOutputParser load_dotenv() class RAGQASystem: def __init__(self, use_local_llmFalse): # 1. 加载相同的嵌入模型和向量库 self.embeddings HuggingFaceEmbeddings(model_nameBAAI/bge-small-zh-v1.5) self.vectorstore Chroma( persist_directory./data/chroma_db, embedding_functionself.embeddings, collection_nameproduct_manual ) self.retriever self.vectorstore.as_retriever(search_kwargs{k: 3}) # 2. 定义提示词模板加入了更详细的指令 self.prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个专业的客服助手请严格根据提供的上下文信息回答问题。 上下文可能包含多个相关片段。请综合这些信息组织成连贯、准确的答案。 如果上下文信息不足以回答问题请明确告知用户“根据现有资料我无法回答这个问题”并建议用户提供更多信息。 请始终使用中文回答。), (human, 上下文\n{context}\n\n问题{question}) ]) # 3. 初始化大语言模型 if use_local_llm: # 假设使用本地部署的 Ollama需要先安装并启动 Ollama 服务 from langchain_community.llms import Ollama self.llm Ollama(modelqwen2.5:7b) else: # 使用 OpenAI API self.llm ChatOpenAI( modelgpt-3.5-turbo, temperature0.1, # 较低的温度使输出更确定 openai_api_keyos.getenv(OPENAI_API_KEY) ) # 4. 使用 LCEL 构建链 self.chain ( {context: self.retriever, question: RunnablePassthrough()} | self.prompt_template | self.llm | StrOutputParser() ) def ask(self, question: str): 提问并获取答案 try: answer self.chain.invoke(question) return answer except Exception as e: return f系统在处理您的问题时出现错误{str(e)} if __name__ __main__: # 初始化系统传入 True 则使用本地模型需提前部署 qa_system RAGQASystem(use_local_llmFalse) # 进行问答测试 test_questions [ 这个产品的主要功能是什么, 如何启动这个设备, 保修期是多久 ] for q in test_questions: print(f\n[用户问题]: {q}) answer qa_system.ask(q) print(f[系统回答]: {answer}) print(- * 50)4.4 运行与验证确保已运行ingest.py生成了向量数据库。运行python src/rag_chain.py。观察输出系统会基于你的docs/目录下的文档内容进行回答。4.5 结果说明如果一切顺利你将看到系统针对每个测试问题从向量库中检索出相关上下文并生成相应的中文答案。答案的质量取决于文档切分的合理性。嵌入模型对中文语义的理解能力。检索到的上下文片段是否相关且充足。大语言模型的总结和生成能力。5. 进阶实战使用 LangGraph 构建一个多智能体协作系统当任务变得复杂需要多个步骤、决策或不同专长的“智能体”协作时LangGraph就派上用场了。我们设计一个场景一个研究助手系统包含一个“研究员”智能体和一个“校对员”智能体。5.1 场景定义用户提出一个研究主题如“LangChain 的最新特性”。研究员Researcher负责搜索网络或本地知识库获取信息并起草一份初步报告。校对员Proofreader负责审查研究员的报告检查事实准确性、语言流畅性并提出修改意见。两个智能体可以循环对话直到校对员满意为止最终输出一份精炼的报告。5.2 定义智能体状态与节点首先我们需要定义整个图的状态State和各个节点Node的行为。# src/langgraph_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI import os from dotenv import load_dotenv load_dotenv() # 1. 定义状态结构 class AgentState(TypedDict): 图的状态所有智能体共享和修改的信息 topic: str # 用户输入的研究主题 research_materials: str # 研究员收集的资料 draft_report: str # 研究员起草的报告 critique: str # 校对员提出的批评意见 final_report: str # 最终报告 revision_count: int # 修订轮次用于控制循环 # 2. 初始化大模型两个智能体可以共享一个模型也可以使用不同模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7, api_keyos.getenv(OPENAI_API_KEY)) # 3. 定义“研究员”节点函数 def research_node(state: AgentState) - AgentState: 研究员根据主题和校对意见收集资料并起草报告 topic state[topic] critique state.get(critique, ) # 构建给研究员的提示 if critique: # 如果不是第一轮需要根据校对意见修改 research_prompt f你是一位研究员。用户的研究主题是{topic}。 校对员对你的上一版报告提出了以下意见 {critique} 请根据这些意见重新搜索和整理资料并起草一份新的报告。报告应结构清晰、内容详实。 else: # 第一轮直接研究 research_prompt f你是一位研究员。请针对以下主题进行研究并起草一份详细的报告。 主题{topic} 请先列出你收集到的关键资料要点然后基于这些要点撰写报告。 # 模拟“研究”过程实际项目中这里可以调用搜索工具或检索本地知识库 research_response llm.invoke([HumanMessage(contentresearch_prompt)]) research_content research_response.content # 简单地将响应内容分为“资料”和“报告草案” # 这里为了简化我们假设前半部分是资料后半部分是报告 lines research_content.split(\n) split_point len(lines) // 2 materials \n.join(lines[:split_point]) draft \n.join(lines[split_point:]) # 更新状态 return { research_materials: materials, draft_report: draft, revision_count: state.get(revision_count, 0) 1 } # 4. 定义“校对员”节点函数 def proofread_node(state: AgentState) - AgentState: 校对员审查报告草案提出修改意见或批准 draft state[draft_report] topic state[topic] revision_count state.get(revision_count, 0) proofread_prompt f你是一位严格的校对员。请审查以下关于“{topic}”的研究报告草案 ---报告草案开始--- {draft} ---报告草案结束--- 请从以下方面进行审查 1. **事实准确性**报告中的信息是否准确有无明显错误或过时信息 2. **逻辑与结构**报告结构是否清晰论点是否有力 3. **语言表达**语句是否通顺用词是否专业、准确 如果报告质量合格无需修改请直接说“APPROVED”。 如果报告需要修改请明确指出具体问题并提供修改建议。 proofread_response llm.invoke([HumanMessage(contentproofread_prompt)]) critique proofread_response.content # 判断是否批准 if APPROVED in critique.upper() or revision_count 3: # 设置最大修订轮次为3 # 批准生成最终报告 final_prompt f请基于以下资料和草案整理出一份最终版的研究报告。 主题{topic} 研究资料{state.get(research_materials, )} 报告草案{draft} 请输出一份格式优美、内容完整的最终报告。 final_response llm.invoke([HumanMessage(contentfinal_prompt)]) return {critique: critique, final_report: final_response.content} else: # 不批准返回修改意见让流程继续循环 return {critique: critique} # 5. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(researcher, research_node) workflow.add_node(proofreader, proofread_node) # 设置入口点 workflow.set_entry_point(researcher) # 定义边决定流程走向 workflow.add_edge(researcher, proofreader) # 定义条件边校对员节点后根据内容决定是循环还是结束 def decide_after_proofread(state: AgentState) - str: 判断校对后的下一步继续修订还是结束 if state.get(final_report): return end # 已有最终报告结束 elif state.get(revision_count, 0) 3: return end # 达到最大修订次数强制结束 else: return continue_revise # 需要继续修订 workflow.add_conditional_edges( proofreader, decide_after_proofread, { continue_revise: researcher, # 跳回研究员节点 end: END # 结束图 } ) # 编译图 app workflow.compile() # 6. 运行图 if __name__ __main__: # 初始化状态 initial_state: AgentState { topic: LangChain 1.3 版本的主要新特性有哪些, research_materials: , draft_report: , critique: , final_report: , revision_count: 0 } print(开始多智能体协作研究流程...) print(f研究主题{initial_state[topic]}\n) # 执行图 final_state None for step, output in app.stream(initial_state, stream_modevalues): node_name list(step.keys())[0] print(f[步骤] 执行节点{node_name}) if node_name proofreader: critique output[proofreader].get(critique, 无) print(f 校对意见{critique[:200]}...) # 打印前200字符 if output[node_name].get(final_report): final_state output[node_name] print(\n[完成] 最终报告已生成) break if final_state: print(\n *50) print(最终研究报告) print(*50) print(final_state[final_report])5.3 运行与理解运行python src/langgraph_agent.py。你会看到控制台输出两个智能体交替工作的过程researcher节点先运行生成初步报告。proofreader节点运行提出修改意见。根据意见图会循环回researcher节点进行修改。直到报告被批准或达到最大修订次数生成最终报告。这个例子展示了 LangGraph 如何管理有状态的、循环的、多角色的工作流这是构建复杂 AI 应用如自动客服、游戏 NPC、工作流自动化的强大工具。6. 常见问题与排查思路在开发 LangChain RAG 应用时你可能会遇到以下典型问题问题现象常见原因解决思路ModuleNotFoundError: No module named ‘langchain_community’安装的 LangChain 版本不包含社区包或环境未安装正确。使用pip install langchain-community单独安装。确保在正确的虚拟环境中操作。向量检索结果不相关1. 文本切分不合理chunk_size 过大或过小。2. 嵌入模型不适合中文或特定领域。3. 检索器参数如k值设置不当。1. 调整chunk_size和chunk_overlap尝试不同的分隔符。2. 尝试不同的嵌入模型如BAAI/bge-large-zh-v1.5。3. 尝试search_type“mmr”并调整fetch_k参数。大模型回答“根据上下文无法回答”1. 检索到的上下文确实不包含答案。2. 提示词Prompt设计不佳未强制模型使用上下文。3. 上下文长度超过模型限制。1. 检查源文档是否包含相关信息优化检索。2. 强化 Prompt 中的指令如“你必须且只能使用以下上下文”。3. 减少retriever返回的 chunk 数量 (k) 或减小chunk_size。使用本地模型如 Ollama速度慢或无响应1. 模型未正确下载或加载。2. 硬件资源RAM/GPU不足。3. Ollama 服务未启动。1. 运行ollama pull model-name确保模型存在。2. 尝试更小的模型如 7B 参数。3. 检查 Ollama 服务是否运行 (ollama serve)。LangGraph 图编译或运行出错1. 状态State结构定义与节点返回值不匹配。2. 条件边conditional_edge的判断函数返回值未在映射中定义。1. 确保每个节点返回的字典键名与TypedDict定义的字段一致。2. 检查add_conditional_edges中path_map字典的键是否覆盖了判断函数所有可能的返回值。处理中文 PDF 时出现乱码PDF 文档编码或字体问题。尝试使用UnstructuredPDFLoader或pdfplumber后端它们对中文支持更好。安装pip install unstructured[pdf]。7. 最佳实践与工程建议将原型应用到生产环境需要考虑更多工程化细节。7.1 提示词工程清晰明确给模型的指令必须无歧义。明确角色、任务、输出格式。提供示例在 Prompt 中加入一两个示例Few-Shot Learning能显著提升模型在复杂任务上的表现。结构化输出要求模型以 JSON、XML 或特定标记格式输出便于后续程序解析。可以使用 LangChain 的PydanticOutputParser。迭代优化将 Prompt 单独存储在配置文件或数据库中方便进行 A/B 测试和迭代。7.2 数据预处理清洗与标准化在加载文档前去除无关字符、标准化日期格式、统一术语。元数据附加在切分文本块时为每个块附加来源、页码、章节等元数据。这在后续追溯答案来源时至关重要。from langchain.schema import Document chunk Document( page_contenttext, metadata{source: manual.pdf, page: 10, section: 安装指南} )分层索引对于大型文档可以创建不同粒度如章节、段落、句子的索引进行多级检索。7.3 应用架构异步处理文档加载、向量化、模型调用都是 I/O 密集型操作使用asyncio或 LangChain 的异步接口提升吞吐量。缓存机制对频繁相同的查询结果进行缓存如使用Redis减少对模型和向量库的调用降低成本、提高响应速度。监控与评估记录用户的提问、检索到的上下文、模型的回答。定期评估回答的准确性、相关性和有用性持续优化检索和 Prompt。7.4 安全与成本API 密钥管理永远不要将 API Key 硬编码在代码中。使用.env文件和环境变量并在生产环境使用密钥管理服务。输入输出过滤对用户输入进行必要的清洗和过滤防止 Prompt 注入攻击。对模型输出进行敏感词过滤和内容安全检查。成本控制使用 Token 计数器监控每次调用的开销。对于简单的检索可以优先使用本地嵌入模型和小型本地 LLM。设置用量告警。7.5 LangGraph 生产级考量状态持久化生产环境中图的状态需要持久化到数据库如 PostgreSQL、Redis以支持长时间运行或中断恢复的任务。可视化与调试利用workflow.get_graph().draw_mermaid()输出图的 Mermaid 图表便于理解和调试复杂的工作流。超时与错误处理为每个节点设置超时并实现完善的错误处理try...catch和重试逻辑避免整个图因单点故障而崩溃。从构建一个简单的 RAG 问答系统到设计一个多智能体协作的 LangGraph 工作流LangChain 为我们提供了将大语言模型能力融入实际应用的强大工具箱。核心在于理解其模块化思想将复杂问题分解为加载、分割、嵌入、检索、生成等标准化步骤再通过 Chain 或 Graph 进行灵活编排。实践是学习的最佳途径。建议你从本文的示例代码出发尝试更换不同的文档、嵌入模型、大语言模型并调整各种参数观察其对最终效果的影响。当你熟悉了基本流程后可以进一步探索 LangChain 丰富的工具集成如搜索引擎、计算器、API 调用构建能够自主使用工具的智能体Agent这将真正释放大语言模型在自动化领域的潜力。