LangChain与LangGraph实战:从RAG系统到多智能体协作开发指南 📅 发布时间:2026/8/24 11:23:07 👁 浏览次数: 1. 先搞清楚 LangChain、LangGraph 和 Agent 到底能帮你做什么如果你正在找 LangChain 和 LangGraph 的实战教程尤其是想搞懂怎么用它们来搭建多智能体Multi-Agent和 RAG检索增强生成系统那这篇内容就是为你准备的。我花了几天时间把市面上常见的教程、文档和项目都跑了一遍发现很多新手卡住的地方根本不是代码有多难而是从一开始就没理清这几个东西的关系和边界。结果就是要么对着一个简单的“Hello World”例子觉得没用要么一上来就想搞个复杂的系统结果在环境、依赖和概念上反复踩坑。简单来说你可以这么理解LangChain是一个“工具箱”和“连接器”。它帮你把大语言模型LLM、各种外部工具比如搜索引擎、数据库、API、以及你的数据文档、数据库方便地连接和组织起来。它的核心是Chain链你可以把它想象成一个流水线数据按顺序经过不同的处理环节。LangGraph是 LangChain 生态里一个更高级的“流程控制器”。当你的任务不是一条直线能走完而是需要根据上一步的结果循环、分支、或者让多个“工人”Agent协作时Chain 就有点力不从心了。LangGraph 引入了State状态和Graph图的概念让你能像画流程图一样设计复杂的、带状态的 AI 应用逻辑。多智能体系统就是它的典型应用场景。Agent不是一个具体的库而是一种设计模式或“智能体”。它通常由 LLM 作为“大脑”加上一些工具Tools和记忆Memory构成能够理解目标、规划步骤、使用工具、并持续执行。在 LangChain 里你可以用AgentExecutor来运行一个 Agent。而在 LangGraph 里你可以更优雅地实现多个 Agent 之间的协作与通信。RAG是一种技术架构目的是让 LLM 能基于你提供的、它本身不知道的外部知识来回答问题。通常的步骤是加载你的文档 - 切分成块 - 转换成向量存入向量数据库 - 用户提问时先从向量库检索相关片段 - 把这些片段和问题一起交给 LLM 生成答案。LangChain 提供了全套的 RAG 组件文档加载器、文本分割器、向量化、检索器是实现 RAG 最流行的框架之一。所以一个典型的进阶路径是先用 LangChain 搞懂 Chain 和简单的 Agent实现一个基础的 RAG 问答。当你需要处理更复杂的、需要多个步骤循环判断或者多个角色分工的任务时再上 LangGraph 来构建多智能体系统。接下来我会按照这个认知路径带你从环境搭建到项目实战走一遍。2. 环境准备别在第一步就卡住很多教程默认你已经配好了完美的 Python 环境但现实中版本冲突、依赖缺失才是最大的拦路虎。我的建议是无论如何先创建一个干净的虚拟环境。这是避免 99% 依赖地狱问题的前提。2.1 创建并激活虚拟环境你可以使用venv或conda这里以venv为例# 创建名为 langchain_env 的虚拟环境 python -m venv langchain_env # 激活环境 # Windows langchain_env\Scripts\activate # macOS/Linux source langchain_env/bin/activate激活后你的命令行提示符前面应该会出现(langchain_env)字样。2.2 安装核心依赖LangChain 和 LangGraph 的生态依赖较多我们分批安装先确保核心能跑起来。注意 Python 版本建议使用 3.8 到 3.113.12 及以上版本可能遇到某些底层库的兼容性问题。# 1. 安装 LangChain 全家桶核心包含标准版和社区版的一些工具 pip install langchain langchain-community # 2. 安装 LangGraph pip install langgraph # 3. 安装 OpenAI 库如果你使用 OpenAI 的模型如 GPT-3.5/4 pip install openai # 4. 安装用于嵌入Embedding和向量存储的库 # 这里以 OpenAI 的 Embeddings 和 Chroma一个轻量级向量数据库为例 pip install langchain-openai chromadb # 5. 安装用于文档处理的常用库 pip install pypdf python-dotenv tiktoken安装完成后用pip list检查一下主要包是否都在。2.3 配置 API 密钥大多数功能需要调用大模型 API你需要准备一个 API 密钥例如 OpenAI 的。永远不要将密钥硬编码在代码中在项目根目录创建一个名为.env的文件。在文件中写入OPENAI_API_KEY你的_OpenAI_API_密钥在 Python 代码中使用python-dotenv来加载from dotenv import load_dotenv load_dotenv() # 这会从 .env 文件加载环境变量 import os api_key os.getenv(OPENAI_API_KEY)现在基础环境就准备好了。我建议你先别急着写复杂应用用下面这个超简代码测试一下环境和 API 是否连通。2.4 连通性测试创建一个test_setup.py文件from dotenv import load_dotenv load_dotenv() import os from langchain_openai import ChatOpenAI # 初始化一个聊天模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 尝试一个简单的调用 response llm.invoke(请用一句话介绍你自己。) print(response.content)运行它python test_setup.py如果能看到一句 GPT 的自我介绍恭喜你环境通了。如果报错大概率是API_KEY 未设置或错误检查.env文件格式和变量名。网络问题确保你的网络环境可以访问 OpenAI API。余额不足检查 OpenAI 账户余额。3. 用 LangChain 快速构建你的第一个 RAG 系统理解了基础我们立刻动手做一个最有用的东西一个能读取你本地 PDF 文档并回答问题的 RAG 系统。这个过程会串联起 LangChain 的几个核心概念文档加载、文本分割、向量化、检索链。3.1 准备文档并加载假设你有一个名为report.pdf的技术报告在项目目录下。from langchain_community.document_loaders import PyPDFLoader # 加载 PDF 文档 loader PyPDFLoader(./report.pdf) documents loader.load() print(f加载了 {len(documents)} 页文档。) # 查看第一页的内容片段 print(documents[0].page_content[:500])PyPDFLoader将每一页转换成一个Document对象。Document是 LangChain 处理文本的基本单元主要包含page_content文本内容和metadata元数据如页码。3.2 将长文本切分成块LLM 有上下文长度限制直接扔整本书进去不行。我们需要把文档切成语义相关的小块。from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个块的最大字符数 chunk_overlap50, # 块之间的重叠字符避免语义断裂 separators[\n\n, \n, 。, , , , , ] # 分割符优先级 ) chunks text_splitter.split_documents(documents) print(f将文档切分成了 {len(chunks)} 个文本块。)关键参数理解chunk_size不是越小越好。太小会丢失上下文太大会超出模型限制且检索不精准。500-1000 是常见起点。chunk_overlap必要的重叠可以防止一个完整的句子或概念被硬生生切开。separators按此顺序尝试分割尽量保证块在段落、句子边界处断开。3.3 将文本块转换为向量并存储我们需要把文字变成计算机能计算相似度的向量Embedding然后存进向量数据库。from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma # 初始化嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 将切分好的文本块进行向量化并存入 Chroma 数据库 # persist_directory 指定持久化目录否则数据只在内存中 vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./chroma_db # 数据将保存到此文件夹 ) vectorstore.persist() # 显式持久化保存 print(向量数据库已创建并保存。)这一步可能耗时较长取决于文档大小和网络速度。Chroma会在本地./chroma_db目录生成文件下次启动可以直接加载无需重新向量化。3.4 构建检索问答链这是核心环节将检索器Retriever和语言模型LLM链接起来。from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI # 从已保存的向量库加载检索器 vectorstore Chroma(persist_directory./chroma_db, embedding_functionembeddings) retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 检索最相关的 3 个块 # 初始化一个 LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 创建检索问答链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 最常用的类型将检索到的所有文档“塞”给 LLM retrieverretriever, return_source_documentsTrue # 返回参考来源便于验证 ) # 提问 question 这份报告中提到的主要挑战是什么 result qa_chain.invoke({query: question}) print(答案, result[result]) print(\n参考来源) for doc in result[source_documents]: print(f- {doc.page_content[:200]}...)发生了什么你提问“主要挑战是什么”retriever从向量库中找出和问题最相关的 3 个文本块。qa_chain将这 3 个文本块和你的问题一起组合成一个详细的提示Prompt交给 LLM。LLM 基于你提供的“上下文”检索到的文本块生成答案。返回答案和来源你可以核对答案是否基于文档。至此一个最基础的、可运行的 RAG 系统就完成了。你可以更换不同的文档、调整chunk_size、尝试不同的search_kwargs如k值来观察效果变化。4. 从单链到智能体理解 LangChain Agent 的工作流RAG 链是确定性的流程检索 - 组合 - 生成。但很多任务需要“思考”和“决策”比如“帮我去网上查一下今天天气如果下雨就提醒我带伞”。这就需要Agent。Agent 的核心是LLM Tools 执行循环。LLM 决定何时使用哪个工具并根据工具返回的结果决定下一步动作。4.1 定义工具工具是 Agent 的手和脚。我们先定义两个简单的工具。from langchain.tools import tool import datetime tool def get_current_time(placeholder: str ) - str: 获取当前的日期和时间。当用户询问时间时使用此工具。 now datetime.datetime.now() return now.strftime(%Y-%m-%d %H:%M:%S) tool def calculate_length(text: str) - str: 计算输入字符串的字符长度。 return f字符串 {text} 的长度是 {len(text)} 个字符。 # 将工具放入列表 tools [get_current_time, calculate_length]4.2 创建 Agent 并运行我们将使用 OpenAI 函数调用Function Calling来创建 Agent这是目前最稳定和推荐的方式。from langchain_openai import ChatOpenAI from langchain.agents import create_openai_functions_agent, AgentExecutor from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder # 初始化 LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 定义提示模板告诉 Agent 它有什么工具以及如何行事 prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手。请根据需要使用工具来回答问题。), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # 预留位置用于存放 Agent 的思考过程 ]) # 创建 Agent agent create_openai_functions_agent(llmllm, toolstools, promptprompt) # 创建 Agent 执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 运行 Agent result agent_executor.invoke({input: 现在几点了然后告诉我‘Hello World’的长度是多少。}) print(result[output])运行上述代码你会看到verboseTrue模式下详细的思考过程LLM 识别出需要调用get_current_time工具。执行工具获取时间结果。LLM 看到结果后识别出还需要调用calculate_length工具。执行工具获取长度结果。LLM 将两个结果整合生成最终回复给用户。这就是一个简单的ReActReason Act模式。Agent 的威力在于你可以给它接入搜索引擎、数据库查询、代码执行等成百上千个工具它就能完成非常复杂的任务编排。5. 引入 LangGraph设计多智能体协作系统当任务复杂到单个 Agent 忙不过来或者需要多个角色如分析师、作家、评审员协作时就该LangGraph登场了。LangGraph 让你能用“图”来定义一组 Agent 或函数的工作流其中包含循环、条件分支和共享状态。我们设计一个经典场景一个多智能体审稿系统。流程是作家写初稿 -分析师检查事实和数据 -评审员评估逻辑和语言 - 如果评审员不满意则返回给作家修改直到通过。5.1 定义状态和节点首先定义整个工作流共享的“状态”State。我们使用TypedDict来明确结构。from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): # 整个流程的输入比如写作主题 topic: str # 当前轮次生成的草稿 draft: str # 分析师提供的事实核查意见 fact_check_feedback: str # 评审员提供的逻辑语言意见 review_feedback: str # 记录流程走到了哪一步 step: str # 评审是否通过 approved: bool然后我们定义图中的各个“节点”Node每个节点是一个函数接收状态修改状态。from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) def writer_node(state: AgentState) - AgentState: 作家节点根据主题或反馈撰写/修改草稿。 if state[step] start: instruction f请围绕以下主题撰写一篇短文初稿{state[topic]} else: instruction f请根据以下反馈修改草稿\n事实核查意见{state[fact_check_feedback]}\n评审意见{state[review_feedback]}\n\n原草稿{state[draft]} prompt f{instruction}\n请直接输出修改后的完整草稿不要附加任何解释。 response llm.invoke(prompt) return {draft: response.content, step: written} def analyst_node(state: AgentState) - AgentState: 分析师节点检查草稿中的事实和数据准确性。 prompt f请检查以下文本中事实和数据的准确性指出任何可能的问题或需要核实的地方\n\n{state[draft]}\n\n请直接给出核查意见。 response llm.invoke(prompt) return {fact_check_feedback: response.content, step: fact_checked} def reviewer_node(state: AgentState) - AgentState: 评审员节点评估草稿的逻辑、结构和语言。 prompt f请从逻辑连贯性、结构清晰度和语言流畅度评审以下文本\n\n{state[draft]}\n\n请直接给出评审意见并最终判断是否通过只需回答‘通过’或‘不通过’。 response llm.invoke(prompt) feedback response.content approved 通过 in feedback return {review_feedback: feedback, approved: approved, step: reviewed}5.2 构建图并定义流程这是 LangGraph 的核心将节点连接起来并定义路由逻辑。# 初始化一个图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(writer, writer_node) workflow.add_node(analyst, analyst_node) workflow.add_node(reviewer, reviewer_node) # 设置入口点 workflow.set_entry_point(writer) # 定义边连接关系 workflow.add_edge(writer, analyst) # 作家写完后交给分析师 workflow.add_edge(analyst, reviewer) # 分析师检查后交给评审员 # 定义条件边评审员之后去哪根据 approved 字段决定 def decide_after_review(state: AgentState) - str: if state[approved]: return END # 通过结束流程 else: return writer # 不通过返回作家节点修改 workflow.add_conditional_edges( reviewer, # 从哪个节点出发 decide_after_review, # 路由判断函数 {writer: writer, END: END} # 可能的下一个节点映射 ) # 编译图 app workflow.compile()5.3 运行多智能体工作流现在我们可以像运行一个函数一样运行这个复杂的协作流程。# 初始化输入状态 initial_state AgentState( topic人工智能对软件开发行业的影响, draft, fact_check_feedback, review_feedback, stepstart, approvedFalse ) # 运行图 final_state app.invoke(initial_state) print(最终草稿) print(final_state[draft]) print(\n最终状态, final_state[step], 是否通过, final_state[approved])通过 LangGraph 的app.invoke状态会自动在图定义的节点间流转。你可以在writer_node里加入更多逻辑比如限制修改次数防止无限循环。6. 实战避坑与进阶思考走通了基础流程但要真正用在项目里还有一堆细节要处理。下面是我从实际项目中总结的几个关键点和避坑指南。6.1 RAG 效果不佳先检查检索质量RAG 回答不对十有八九是检索没找到对的上下文。调整检索策略search_kwargs里的k返回数量和score_threshold相似度阈值需要调优。k太大可能引入噪声太小可能遗漏关键信息。尝试不同检索器Chroma默认使用余弦相似度。可以试试MMR最大边际相关性搜索它在相似的基础上兼顾多样性。retriever vectorstore.as_retriever( search_typemmr, # 使用 MMR search_kwargs{k: 4, fetch_k: 10} # fetch_k 是初始检索池大小 )优化文本分割chunk_size和chunk_overlap是玄学没有银弹。对于技术文档按章节或标题分割可能比按固定字符数更好。可以试试MarkdownHeaderTextSplitter。检查嵌入模型不同的 Embedding 模型效果差异巨大。text-embedding-3-small是性价比之选。对中文场景可以评估BGE、M3E等开源模型通过langchain_community.embeddings接入。6.2 Agent 胡言乱语或陷入循环给工具清晰的描述工具函数的docstring就是给 LLM 的说明书必须清晰、准确说明输入输出和用途。限制最大迭代次数AgentExecutor的max_iterations参数一定要设置防止 Agent 陷入死循环。agent_executor AgentExecutor(agentagent, toolstools, max_iterations10, verboseTrue)处理解析错误LLM 可能无法正确解析工具调用参数设置handle_parsing_errorsTrue可以让执行器尝试修复或提示用户。为工具提供示例在复杂工具中可以在提示词中为 Agent 提供几个工具调用的示例Few-shot大幅提升其使用工具的准确性。6.3 LangGraph 状态管理复杂状态设计要精简只把需要在节点间传递的变量放入State。避免状态过于庞大难以调试。使用send实现广播LangGraph 支持将状态同时发送到多个后续节点适合并行任务场景。可视化你的图对于复杂工作流画出来看最直观。from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法生成图片可以输出文本结构 print(app.get_graph().draw_ascii())持久化状态对于长时间运行的工作流需要将State保存到数据库以便中断后恢复。LangGraph 提供了与 LangSmith 等工具的集成。6.4 面向生产环境的考量异步与流式响应对于 Web 应用使用ainvoke进行异步调用避免阻塞。使用astream实现流式输出提升用户体验。缓存对 Embedding 和 LLM 的重复调用进行缓存能显著节省成本和时间。LangChain 内置了InMemoryCache、SQLiteCache等。可观测性与评估使用LangSmithLangChain 官方平台来追踪每次链、Agent 的调用详情、耗时、花费并进行效果评估。这对于调试和优化至关重要。模块化与配置化不要把所有代码写在一个文件里。将 Chain、Agent、Graph 的定义模块化并通过配置文件如 YAML来管理模型参数、提示词模板等。7. 总结从学习到应用的路径回过头看学习 LangChain 和 LangGraph 的核心不是记住所有 API而是理解其设计思想LangChain 提供了构建 AI 应用的标准组件和连接方式而 LangGraph 在此基础上解决了复杂、有状态工作流的编排问题。对于个人学习或快速原型我的建议路径是第一步配好环境跑通一个最简单的 RAG 例子本文第 3 节。理解文档-切分-向量化-检索-生成的完整流程。第二步尝试将一个简单的任务如查询天气、计算、查数据库改造成 Agent本文第 4 节。理解 LLM 如何规划、调用工具。第三步当你发现一个任务需要多个步骤循环或者需要多个角色分工时引入 LangGraph本文第 5 节。从简单的两节点循环开始再增加复杂度。第四步深入优化。根据你的具体场景文档类型、问题类型、性能要求去微调 RAG 的每一个环节分割、检索、提示或者为你的 Agent 设计更精准的工具和提示。最终判断一个 AI 应用框架是否好用标准就几条能不能快速把想法搭出来跑起来稳不稳定效果不好时方不方便调试和优化从这几个角度看LangChain 生态目前仍然是构建复杂 AI 应用最成熟、社区最活跃的选择之一。而 LangGraph 的引入确实让多智能体协作这类复杂场景的设计变得清晰和可控了许多。剩下的就是在具体项目里不断踩坑和填坑了。