如果你正在寻找一套能快速上手、实战导向的LangChain与LangGraph学习路径那么这篇文章就是为你准备的。我们不再空谈概念而是直接切入核心如何用这两大框架结合Agent智能体与RAG检索增强生成技术构建一个真正可用的多智能体系统。本文的目标是让你在最短时间内掌握从环境搭建、核心概念理解到项目实战落地的完整流程避开那些文档里没写明的“坑”。我们将重点关注几个实用问题LangChain和LangGraph到底有什么区别在什么场景下该用谁如何设计一个由多个角色协作的智能体工作流如何将RAG知识库无缝集成到智能体的决策中整个过程我们会以一套可运行的代码示例为主线让你看到每一步的输出和效果。无论你是想快速集成一个智能客服助手还是构建一个复杂的自动化分析流水线这里的内容都能提供直接的参考。1. 核心能力速览LangChain vs. LangGraph在开始动手之前必须先理清这两个核心框架的定位。它们不是替代关系而是协作关系用于解决不同复杂度的问题。能力项LangChainLangGraph核心定位应用构建框架工作流编排框架抽象层级相对较高提供链Chain、代理Agent等预制模块相对较低基于图Graph模型提供极致的编排灵活性擅长场景快速构建标准的问答、总结、提取等应用集成各种工具和模型。构建有状态、多步骤、带循环或分支判断的复杂工作流如多智能体协作。状态管理隐式管理链之间传递上下文。显式状态管理整个工作流的状态State是一个可定义、可观察的核心对象。可视化有限。强大通过LangGraph Studio可直观看到工作流节点和边的执行过程。学习曲线入门平缓文档丰富。需要对图计算、状态机有基本理解入门稍陡但功能强大。本文焦点作为基础组件用于工具调用、模型交互。作为“大脑”和“调度中心”协调多个智能体角色完成复杂任务。简单来说用LangChain造“零件”工具、链用LangGraph组装“机器”复杂工作流。接下来我们将基于这个认知搭建一个融合了多智能体和RAG的实战项目。2. 项目目标与适用场景本次实战的目标是构建一个“技术调研助手”多智能体系统。它的工作流程是用户提出一个复杂的技术调研需求例如“帮我对比一下LangGraph和Dify在构建Agent方面的优劣”。主协调智能体Orchestrator解析需求将其拆解为子任务如资料检索、优势分析、劣势分析、报告生成。多个专业智能体如检索专家、分析专家、写作专家被依次或并行调用各司其职。检索增强生成RAG模块为智能体提供精准的外部知识如最新的技术文档、博客确保回答的时效性和准确性。最终系统整合所有结果生成一份结构清晰、有据可查的调研报告。适用场景企业内部知识问答与决策支持快速分析市场报告、技术文档提供综述。智能客服升级处理需要多步骤查询、判断和合成的复杂用户问题。自动化内容生产基于给定主题自动搜集资料、撰写初稿。教育辅助构建能够引导学生分步思考、提供资料的辅导智能体。使用边界与合规提醒本项目依赖于大语言模型LLM请确保你使用的模型API或本地模型符合相关法律法规和服务条款。RAG系统处理的知识库文档请确保你拥有使用权或文档是公开可用的避免版权风险。智能体的输出结果需要人工复核尤其涉及事实陈述、数据引用或重要建议时。3. 环境准备与前置条件我们将创建一个干净的Python环境来管理依赖。这是避免包冲突的最佳实践。基础环境要求操作系统Windows 10/11, macOS, 或 Linux (推荐 Ubuntu 20.04)Python版本3.10 或 3.11 (3.12可能存在部分库的兼容性问题建议使用3.11)包管理工具pip(建议版本21.3以上)网络能正常访问PyPI以及你所选LLM的API如OpenAI, Anthropic, 或国内合规模型平台核心依赖库我们将使用pip安装。首先创建一个新的项目目录并进入。mkdir langgraph-agent-rag-demo cd langgraph-agent-rag-demo然后创建一个requirements.txt文件内容如下# 核心框架 langchain0.1.0 langchain-community0.0.10 langgraph0.0.26 # 用于构建RAG的向量数据库与嵌入模型 chromadb0.4.22 sentence-transformers2.2.2 # 或者使用OpenAI的嵌入二选一 # openai1.6.1 # 文档加载与处理 unstructured0.10.30 pypdf3.17.4 markdown3.5.2 # LLM调用 (以OpenAI为例也可替换为其他) openai1.6.1 # 或者使用Ollama本地模型 # ollama0.1.2 # 其他工具 python-dotenv1.0.0 # 管理环境变量使用以下命令安装所有依赖pip install -r requirements.txtLLM API密钥配置本项目以OpenAI GPT-4为例。你需要准备一个API Key。在项目根目录创建.env文件来安全存储密钥# .env 文件内容 OPENAI_API_KEY你的OpenAI_API密钥 # 如果使用其他模型例如 Anthropic # ANTHROPIC_API_KEY你的Anthropic_API密钥然后在Python代码中通过dotenv加载。4. 项目结构设计与核心模块在写代码前先规划好项目结构这有助于管理复杂度。langgraph-agent-rag-demo/ ├── .env # 环境变量配置文件 ├── requirements.txt # 项目依赖 ├── main.py # 主程序入口 ├── config.py # 配置参数模型、路径等 ├── knowledge_base/ # 知识库文档存放目录 │ ├── langgraph_docs.pdf │ └── dify_blog.md ├── vector_store/ # 向量数据库持久化目录自动生成 ├── agents/ # 智能体模块 │ ├── __init__.py │ ├── orchestrator.py # 主协调智能体 │ ├── researcher.py # 检索分析智能体 │ └── writer.py # 报告撰写智能体 ├── graph/ # LangGraph工作流定义 │ ├── __init__.py │ └── tech_research_graph.py └── tools/ # 自定义工具 ├── __init__.py └── rag_tool.py # RAG检索工具5. 第一步构建RAG知识库多智能体需要可靠的信息来源。我们首先构建一个RAG系统它将作为所有智能体可查询的“外部大脑”。5.1 文档加载与处理在tools/rag_tool.py中我们创建一个文档加载和向量化函数。# tools/rag_tool.py import os from typing import List from langchain_community.document_loaders import PyPDFLoader, UnstructuredMarkdownLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.embeddings import OpenAIEmbeddings # 或者使用本地嵌入模型无需API # from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma class RAGTool: def __init__(self, persist_directory: str ./vector_store): self.persist_directory persist_directory # 使用OpenAI Embeddings (需要API Key) self.embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 或者使用本地模型例如 all-MiniLM-L6-v2 # self.embeddings HuggingFaceEmbeddings(model_nameall-MiniLM-L6-v2) self.vector_store None def load_and_split_documents(self, docs_dir: str ./knowledge_base) - List: 加载指定目录下的所有文档并进行分块 documents [] for filename in os.listdir(docs_dir): file_path os.path.join(docs_dir, filename) if filename.endswith(.pdf): loader PyPDFLoader(file_path) documents.extend(loader.load()) elif filename.endswith(.md): loader UnstructuredMarkdownLoader(file_path) documents.extend(loader.load()) # 可以添加更多文件类型支持如 .txt, .docx # 文本分割确保块大小适中有重叠以避免割裂上下文 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) split_docs text_splitter.split_documents(documents) print(f已加载 {len(documents)} 个文档分割为 {len(split_docs)} 个文本块。) return split_docs def create_vector_store(self, docs_dir: str ./knowledge_base): 创建并持久化向量数据库 split_docs self.load_and_split_documents(docs_dir) self.vector_store Chroma.from_documents( documentssplit_docs, embeddingself.embeddings, persist_directoryself.persist_directory ) self.vector_store.persist() print(f向量数据库已创建并保存至 {self.persist_directory}) def load_existing_vector_store(self): 加载已存在的向量数据库 self.vector_store Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings ) print(f已从 {self.persist_directory} 加载现有向量数据库。) def query(self, question: str, k: int 4) - List[str]: 查询知识库返回最相关的k个文本片段 if self.vector_store is None: self.load_existing_vector_store() docs self.vector_store.similarity_search(question, kk) contexts [doc.page_content for doc in docs] return contexts # 初始化并创建知识库只需运行一次 if __name__ __main__: rag_tool RAGTool() # 首次运行创建向量库 # rag_tool.create_vector_store() # 之后运行直接加载 rag_tool.load_existing_vector_store() test_result rag_tool.query(LangGraph的主要特点是什么) print(测试查询结果:, test_result)运行一次这个脚本将knowledge_base目录下的文档如LangGraph官方PDF、Dify技术博客Markdown向量化并存储。后续智能体将通过这个工具进行查询。6. 第二步定义多个智能体角色我们将定义三个具有不同职责的智能体。它们都基于LangChain的AgentExecutor构建但拥有不同的提示词Prompt和工具集。6.1 检索分析智能体 (Researcher)这个智能体负责从RAG知识库和网络如果需要中查找信息。agents/researcher.py# agents/researcher.py from langchain.agents import Tool, AgentExecutor, create_react_agent from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI from tools.rag_tool import RAGTool class ResearchAgent: def __init__(self, rag_tool: RAGTool): self.llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) self.rag_tool rag_tool self.tools self._define_tools() self.agent self._create_agent() def _define_tools(self): 定义研究员可用的工具 def rag_query(query: str) - str: 使用RAG知识库查询相关信息。输入应为一个明确的问题。 contexts self.rag_tool.query(query, k4) return \n\n.join(contexts) tools [ Tool( nameKnowledgeBaseSearch, funcrag_query, description当需要从内部知识库查找关于LangChain, LangGraph, Dify, Agent, RAG等技术概念、对比、用法或最新信息时使用此工具。输入应为一个具体的问题。 ), # 可以在此添加更多工具如网络搜索工具需API Key # Tool(nameWebSearch, funcweb_search, description...), ] return tools def _create_agent(self): 创建ReAct模式的智能体 prompt PromptTemplate.from_template( 你是一位资深技术研究员。你的任务是根据用户的问题利用所有可用的工具查找最相关、最准确的技术信息。 请严格遵循以下步骤思考 1. 理解问题仔细分析用户想问什么。 2. 制定搜索策略决定使用哪个工具以及输入什么查询词。 3. 执行搜索调用工具获取信息。 4. 分析信息评估获取的信息是否足够回答用户问题。如果不够调整查询词再次搜索。 5. 总结答案基于你找到的信息给出清晰、有条理的回答。务必注明信息来源的要点。 当前问题{input} 开始你的工作吧 ) agent create_react_agent(llmself.llm, toolsself.tools, promptprompt) return AgentExecutor(agentagent, toolsself.tools, verboseTrue, handle_parsing_errorsTrue) def run(self, query: str) - str: 执行研究任务 result self.agent.invoke({input: query}) return result[output]6.2 报告撰写智能体 (Writer)这个智能体负责将分析结果组织成结构化的报告。agents/writer.py# agents/writer.py from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate from langchain_openai import ChatOpenAI class WriterAgent: def __init__(self): self.llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0.7) # 温度稍高更有创造性 def write_report(self, topic: str, research_findings: str, analysis: str) - str: 根据研究结果和分析撰写最终报告 system_template 你是一位专业的科技报告撰写人。你擅长将复杂的技术信息整理成结构清晰、重点突出、易于理解的报告。 报告需要包含概述、核心要点分项阐述、对比表格如果适用、总结与建议。 语言风格专业、准确、简洁。 human_template 请根据以下材料撰写一份关于『{topic}』的技术调研报告。 【研究发现的原始信息】 {research_findings} 【初步分析结论】 {analysis} 请生成完整的报告。 prompt ChatPromptTemplate.from_messages([ SystemMessagePromptTemplate.from_template(system_template), HumanMessagePromptTemplate.from_template(human_template) ]) chain prompt | self.llm report chain.invoke({topic: topic, research_findings: research_findings, analysis: analysis}) return report.content6.3 主协调智能体 (Orchestrator)这是整个系统的“指挥官”它不直接调用工具而是理解用户意图拆解任务并调用其他智能体。agents/orchestrator.py# agents/orchestrator.py from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate from langchain_openai import ChatOpenAI from typing import Dict, Any class OrchestratorAgent: def __init__(self): self.llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) def plan_and_decompose(self, user_query: str) - Dict[str, Any]: 分析用户查询制定执行计划任务分解 system_template 你是一个多智能体系统的总调度员。你的职责是理解复杂任务并将其分解为一系列可以由专家智能体执行的子任务。 当前可调用的专家智能体有 1. 研究员Researcher负责从知识库中检索和汇总信息。 2. 撰写员Writer负责根据信息和指令生成结构化的报告。 请分析用户请求并输出一个JSON格式的计划包含以下字段 - original_query: 原始用户问题。 - sub_tasks: 一个列表每个元素是一个子任务对象。每个子任务对象包含 * agent: 执行该任务的智能体名称Researcher 或 Writer。 * instruction: 给该智能体的具体指令。 * depends_on: 该任务依赖的前置子任务ID列表从0开始如果没有则为空列表。 - final_integration_agent: 负责最终整合的智能体名称通常是Writer。 human_template 用户请求{query} prompt ChatPromptTemplate.from_messages([ SystemMessagePromptTemplate.from_template(system_template), HumanMessagePromptTemplate.from_template(human_template) ]) chain prompt | self.llm response chain.invoke({query: user_query}) # 这里需要解析LLM返回的JSON。实际应用中需要更健壮的解析这里简化为返回文本。 # 为简化演示我们返回一个预定义的计划结构。 # 在实际项目中你应该使用LLM的JSON模式或函数调用功能来获取结构化输出。 plan { original_query: user_query, sub_tasks: [ {id: 0, agent: Researcher, instruction: f检索并总结关于{user_query}的所有相关技术信息包括核心概念、功能、优缺点。, depends_on: []}, {id: 1, agent: Researcher, instruction: 重点对比其中涉及的主要技术或工具如LangGraph vs Dify在架构、易用性、灵活性方面的差异。, depends_on: [0]}, {id: 2, agent: Writer, instruction: 基于研究员提供的所有信息撰写一份详细的技术对比分析报告。, depends_on: [0, 1]}, ], final_integration_agent: Writer } print(f协调员生成计划{plan}) return plan7. 第三步用LangGraph编排工作流这是最核心的一步。我们将使用LangGraph把上述智能体连接成一个有状态的工作流。graph/tech_research_graph.py# graph/tech_research_graph.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from agents.orchestrator import OrchestratorAgent from agents.researcher import ResearchAgent from agents.writer import WriterAgent from tools.rag_tool import RAGTool # 1. 定义工作流的状态State class AgentState(TypedDict): 整个多智能体工作流的状态定义 messages: Annotated[List[str], add_messages] # 消息历史本示例简化处理 original_query: str # 原始用户问题 plan: dict # 协调员生成的计划 research_results: List[str] # 研究员的结果列表 analysis: str # 分析结论可来自研究员或中间步骤 final_report: str # 最终报告 # 2. 初始化所有智能体和工具 rag_tool RAGTool() rag_tool.load_existing_vector_store() # 加载已有向量库 orchestrator OrchestratorAgent() researcher ResearchAgent(rag_tool) writer WriterAgent() # 3. 定义各个节点Node的函数 def orchestrator_node(state: AgentState) - AgentState: 节点协调员负责任务规划 print(\n--- [Orchestrator Node] 开始任务规划 ---) plan orchestrator.plan_and_decompose(state[original_query]) state[plan] plan print(f规划完成: {plan[sub_tasks]}) return state def researcher_node(state: AgentState) - AgentState: 节点研究员执行检索分析任务 print(\n--- [Researcher Node] 开始执行研究任务 ---) # 简化处理执行计划中的第一个研究员任务 current_task None for task in state[plan][sub_tasks]: if task[agent] Researcher and len(task.get(depends_on, [])) 0: current_task task break if not current_task: print(未找到可立即执行的研究员任务。) return state instruction current_task[instruction] print(f执行指令: {instruction}) result researcher.run(instruction) state[research_results].append(result) print(f研究结果已保存当前结果数: {len(state[research_results])}) return state def writer_node(state: AgentState) - AgentState: 节点撰写员生成最终报告 print(\n--- [Writer Node] 开始生成报告 ---) # 整合所有研究结果 all_research \n---\n.join(state[research_results]) # 这里可以添加一个“分析员”节点来生成analysis本例简化直接使用部分研究结果作为分析 analysis_input state[research_results][0][:500] if state[research_results] else 无分析输入 report writer.write_report( topicstate[original_query], research_findingsall_research, analysisanalysis_input ) state[final_report] report print(报告生成完成) return state def decide_next_step(state: AgentState) - str: 条件判断边Edge决定下一步该执行哪个节点 print(\n--- [Conditional Edge] 判断下一步 ---) plan state.get(plan, {}) sub_tasks plan.get(sub_tasks, []) # 简化逻辑如果还没规划先去规划 if not plan: return orchestrator # 如果有规划且研究员任务还没执行完示例中只有一个就去研究员节点 researcher_task_done len(state.get(research_results, [])) 0 if not researcher_task_done: return researcher # 否则去撰写员节点 return writer # 4. 构建图Graph def create_tech_research_graph(): 创建并返回编译好的工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(orchestrator, orchestrator_node) workflow.add_node(researcher, researcher_node) workflow.add_node(writer, writer_node) # 设置入口点 workflow.set_entry_point(orchestrator) # 添加条件边 workflow.add_conditional_edges( orchestrator, decide_next_step, { researcher: researcher, writer: writer, # 理论上规划后可能直接去writer这里按逻辑走 } ) workflow.add_conditional_edges( researcher, decide_next_step, { writer: writer, researcher: researcher, # 如果需要多个研究任务可以循环 } ) # 从writer节点直接结束 workflow.add_edge(writer, END) # 编译图 app workflow.compile() return app # 5. 主执行函数 if __name__ __main__: # 创建图应用 graph_app create_tech_research_graph() # 定义初始状态 initial_state: AgentState { messages: [], original_query: 请详细对比LangGraph和Dify在构建AI Agent方面的优势和劣势。, plan: {}, research_results: [], analysis: , final_report: } print(*50) print(开始执行多智能体技术调研工作流...) print(f初始问题: {initial_state[original_query]}) print(*50) # 执行工作流 final_state graph_app.invoke(initial_state) print(\n *50) print(工作流执行完毕) print(*50) print(\n【最终生成的报告】) print(final_state[final_report])8. 运行与效果验证现在让我们运行这个完整的系统。8.1 执行步骤确保你的knowledge_base目录下已放置了一些关于LangGraph和Dify的文档PDF/MD。运行tools/rag_tool.py中的__main__部分创建并测试向量数据库。运行主工作流python graph/tech_research_graph.py。8.2 预期输出与观察执行后你将在控制台看到类似以下流程 开始执行多智能体技术调研工作流... 初始问题: 请详细对比LangGraph和Dify在构建AI Agent方面的优势和劣势。 --- [Orchestrator Node] 开始任务规划 --- 协调员生成计划{...} --- [Conditional Edge] 判断下一步 --- --- [Researcher Node] 开始执行研究任务 --- Entering new AgentExecutor chain... 思考用户想对比LangGraph和Dify。我需要先了解它们各自是什么。 行动使用KnowledgeBaseSearch工具查询“LangGraph和Dify是什么”。 观察[从向量库返回的相关文档片段]... 思考我获得了基本定义。现在需要对比它们在构建Agent方面的优劣。 行动使用KnowledgeBaseSearch工具查询“LangGraph 构建 Agent 优势 劣势”。 观察... ... (链式思考继续) ... Finished chain. 研究结果已保存当前结果数: 1 --- [Conditional Edge] 判断下一步 --- --- [Writer Node] 开始生成报告 --- 报告生成完成 工作流执行完毕 【最终生成的报告】 技术调研报告LangGraph与Dify在构建AI Agent方面的对比分析 1. 概述 本文旨在对比分析LangGraph与Dify两个框架在构建人工智能智能体AI Agent方面的特性、优势与适用场景... 2. 核心对比 ... | 维度 | LangGraph | Dify | |------|-----------|------| | 核心定位 | 工作流编排框架提供极致的灵活性 | 开箱即用的AI应用开发平台 | | 学习曲线 | 较陡需理解图、状态等概念 | 平缓可视化操作为主 | | 定制能力 | 极高可深度定制复杂逻辑 | 中等通过插件和配置实现 | | 部署运维 | 需自行处理 | 平台提供托管服务 | ... 3. 总结与建议 ...8.3 成功验证点流程贯通从用户输入到协调员规划到研究员检索再到撰写员成文整个流程自动执行。状态管理LangGraph清晰地管理了AgentState每个节点都能读取和修改状态。工具调用研究员智能体成功调用了RAG工具进行知识查询。条件路由decide_next_step函数根据当前状态决定下一个节点实现了简单的流程控制。9. 性能优化与扩展方向9.1 性能考量LLM调用成本与延迟这是主要瓶颈。可以通过以下方式优化对智能体进行更精细的提示词工程减少不必要的思考步骤。使用更快的模型如GPT-3.5-Turbo处理简单任务GPT-4处理复杂分析。实现异步调用当任务间无依赖时可并行执行。RAG检索效率向量检索速度很快但嵌入模型的计算或API调用可能有开销。考虑缓存常用查询的嵌入结果。9.2 功能扩展增加更多智能体例如CriticAgent批判性审查报告、CodeGenAgent生成示例代码。实现复杂循环让研究员和分析员可以多轮交互不断深化研究。集成外部API为智能体添加网络搜索、数据库查询、代码执行等工具。持久化与记忆利用LangGraph的检查点Checkpoint功能实现长对话记忆或工作流暂停/恢复。可视化监控使用LangGraph Studio来实时查看工作流的执行状态和流转。9.3 工程化建议错误处理与重试在图中的每个节点添加健壮的错误处理对于暂时性失败如API超时进行重试。日志与审计记录每个智能体的输入、输出和工具调用便于调试和效果分析。配置化管理将模型类型、API密钥、温度参数等抽离到配置文件中。10. 常见问题与排查方法问题现象可能原因排查方式解决方案导入LangChain/LangGraph失败版本不兼容或未安装检查pip list确认版本与requirements.txt一致。创建新的虚拟环境严格按requirements.txt安装。OpenAI API调用报错认证/额度API Key错误或余额不足检查.env文件格式确保密钥正确。在OpenAI后台检查用量。更正API Key或更换为其他模型如Ollama本地模型。RAG检索结果不相关文档分块策略不佳或嵌入模型不合适检查knowledge_base文档内容调整chunk_size和chunk_overlap。优化文本分割参数或尝试不同的嵌入模型如text-embedding-3-large。智能体陷入循环或逻辑错误提示词设计有歧义或ReAct循环失控查看智能体执行的详细日志verboseTrue。优化提示词明确停止条件或在AgentExecutor中设置max_iterations参数。LangGraph图编译错误节点或边定义有误状态定义不匹配仔细检查StateGraph的节点添加顺序和边的指向。确保所有节点函数都正确接收和返回AgentState类型。从简单图开始逐步构建。程序报错ModuleNotFoundError项目目录结构导致Python路径问题确认在项目根目录下运行脚本或使用PYTHONPATH。在根目录下运行python -m graph.tech_research_graph或使用IDE正确设置源根目录。通过以上步骤你已经完成了一个融合LangChain、LangGraph、多智能体和RAG的实战项目。这个项目骨架清晰地展示了如何将不同的AI组件组装成一个协同工作的系统。接下来你可以根据自己的需求替换其中的LLM、丰富工具集、设计更复杂的工作流从而构建出真正强大的AI应用。