1. 项目概述从LangChain到LangGraph的进阶之路最近在GitHub上看到一个项目叫“BrandPeng/Langchain1.0-Langgraph1.0-Learning”。光看这个标题很多朋友可能就有点懵了LangChain我知道是构建大语言模型应用的热门框架但这个LangGraph又是什么它和LangChain 1.0有什么关系这个项目到底在学什么简单来说这个项目像是一份精心编排的学习路线图它聚焦于一个核心的技术演进如何从使用基础的LangChain构建简单的链式应用升级到利用LangGraph构建复杂、有状态、可循环的智能体工作流。如果你已经用LangChain搭过几个聊天机器人或者文档问答系统感觉链条Chain的线性处理有点不够用了想实现更智能的、能“自己思考下一步该做什么”的AI应用那么这个项目所指向的学习路径正是你需要的。LangChain 1.0代表了一个成熟的范式它通过将各种工具模型、检索器、记忆等链接Chain起来形成确定的执行流程。这解决了大模型应用开发中的许多基础问题比如提示词模板化、工具调用标准化。但当你的应用逻辑变得复杂需要根据中间结果动态决定下一步行动或者需要维护一个跨多轮对话的“状态”时单纯的链就显得力不从心了。这时LangGraph登场了。它不是要取代LangChain而是构建在它之上的一个更强大的“编排层”。你可以把LangGraph想象成一个可视化的工作流引擎专门为构建有状态的、多智能体协作的AI应用而设计。这个学习项目本质上是在引导开发者跨越一个关键的能力门槛从编写线性的、脚本式的AI流程到设计并实现图结构的、具备自主决策能力的AI系统。接下来我会结合这个学习路径拆解其中的核心概念、实操要点以及我趟过的一些坑希望能帮你更平滑地完成这次升级。2. 核心概念辨析Chain、Agent与Graph在深入LangGraph之前我们必须先厘清LangChain生态中几个容易混淆的核心概念。理解它们的区别与联系是构建复杂应用的基础。2.1 LangChain Chain确定性的执行管道在LangChain 1.0中Chain是最核心的抽象之一。它代表了一个确定的、顺序的或少量分支的执行流程。一个典型的Chain由多个Component如PromptTemplate,LLM,OutputParser组成数据像在管道中一样流动。例如一个简单的检索问答链RetrievalQA Chain的工作流是确定的接收用户问题。将问题转换为向量用于检索相关文档片段。将问题和检索到的文档片段组合成一个详细的提示词。将提示词发送给大语言模型LLM。解析LLM的返回结果并输出。这个流程在每次运行时都遵循相同的步骤。它的优势在于结构清晰、易于调试和部署。Chain非常适合那些输入到输出映射关系明确、无需根据中间结果改变执行路径的场景。大部分早期的LangChain教程和项目都是围绕构建各种Chain来展开的。注意不要被“链”这个字面意思局限复杂的Chain可以通过SequentialChain,TransformChain等组合成有向无环图DAG但其执行路径在设计时依然是基本确定的运行时不会产生基于内容判断的“循环”或“跳转”。2.2 LangChain Agent引入不确定性决策Agent是LangChain中一个革命性的概念。它为大模型应用赋予了“工具使用”和“自主决策”的能力。一个Agent通常由几个部分组成一个核心的LLM、一系列它可以调用的Tools、一个决定何时及如何使用工具的AgentExecutor以及一个维护对话或任务状态的Memory。Agent的工作流程是动态的Agent接收用户输入和当前状态。核心LLM根据输入和状态思考下一步该做什么。它可能决定直接给出答案Final Answer也可能决定调用某个工具如搜索网络、查询数据库、执行计算。如果调用工具AgentExecutor会执行该工具并将工具执行结果返回给LLM。LLM根据工具结果再次思考决定是继续调用其他工具还是汇总信息给出最终答案。这个过程会循环进行直到LLM决定输出最终答案。这里的“不确定性”在于在编写代码时你无法预知LLM在每一步会具体选择调用哪个工具或者需要循环多少次。这依赖于LLM对当前任务和上下文的理解。Agent实现了基本的“思考-行动-观察”循环是构建智能应用的关键。2.3 LangGraph为复杂Agent提供结构化的工作流引擎那么有了Agent为什么还需要LangGraph当你的智能体逻辑变得极其复杂时原生的Agent框架可能会遇到一些挑战状态管理复杂当多个智能体协作或单个智能体的状态包含多个复杂变量如子任务列表、验证结果、历史决策时用简单的字典管理状态会变得混乱。流程控制薄弱原生的Agent循环相对简单思考-行动。但对于需要多阶段审批、条件分支、并行执行、或特定错误处理流程的场景缺乏直观的编排能力。调试与可视化困难一个动态的、多步的Agent执行过程像是一个黑盒很难清晰地看到整个决策路径和状态流转。LangGraph就是为了解决这些问题而生。它将工作流明确定义为一个图Graph。图中的节点Node代表一个执行单元可以是一个函数、一个工具调用或一个完整的Chain/Agent边Edge代表执行路径。最关键的是LangGraph引入了状态State的概念这是一个在所有节点间共享和传递的、强类型的数据结构。通过LangGraph你可以显式定义循环明确指定在什么条件下流程应该跳回之前的哪个节点实现可控的循环。实现复杂分支根据状态中的某个字段值决定下一步走哪个分支。编排多智能体轻松定义多个智能体角色并规定它们如何通过状态进行通信和协作。获得完整可追溯性整个图的执行过程、每个节点的输入输出、状态的变化都被完整记录极大方便了调试和优化。简而言之LangGraph将Agent那种“模糊”的自主决策纳入了“清晰”的工程化工作流管理之中。它让你既能享受AI决策的灵活性又能拥有软件工程的可控性和可维护性。这个学习项目目标就是带你掌握如何用LangGraph来构建这类下一代AI应用。3. 学习路径拆解从基础链到复杂图工作流基于“BrandPeng/Langchain1.0-Langgraph1.0-Learning”这个项目标题所暗示的路径我们可以梳理出一条循序渐进的学习曲线。这条路径不仅关乎工具的使用更关乎设计思维的转变。3.1 第一阶段夯实LangChain 1.0基础在接触LangGraph之前必须对LangChain的核心组件有扎实的理解。这个阶段的目标是能熟练搭建一个功能完整的、基于链的应用程序。核心学习点模型I/OModel I/O这是与LLM交互的基础。必须掌握ChatPromptTemplate如何结构化提示词了解不同消息角色SystemMessage,HumanMessage,AIMessage的作用熟悉ChatOpenAI,ChatAnthropic等模型类的调用方式以及如何使用OutputParser来规范模型输出。数据连接Retrieval这是让LLM获取外部知识的关键。需要掌握文档加载器DocumentLoader、文本分割器TextSplitter、向量化嵌入Embeddings以及向量数据库如Chroma,Pinecone的集成。最终能构建一个RetrievalQA链实现基于自有知识的问答。记忆Memory为了让对话具有连续性需要理解ConversationBufferMemory,ConversationSummaryMemory等记忆组件的工作原理并知道如何将它们集成到链中。工具Tools学习如何将外部API、函数或计算封装成Tool这是智能体能力的基石。要会使用tool装饰器创建自定义工具。链Chains超越简单的LLMChain学习使用SequentialChain,TransformChain来组合更复杂的工作流理解Runnable协议LangChain的新抽象它让一切组件都可以像函数一样被组合和调用。实操建议不要只看文档一定要动手。可以尝试复现一个经典的“带记忆的检索增强生成RAG聊天机器人”。这个项目会用到上述几乎所有组件。在实现过程中你会深刻体会到Chain的优缺点流程清晰但一旦想加入“根据答案质量决定是否重新检索”的逻辑代码就会变得笨拙。这种“痛点”正是你学习LangGraph的最大动力。3.2 第二阶段初探智能体与LangGraph核心概念当你用Chain构建的应用遇到灵活性瓶颈时就可以自然过渡到Agent和LangGraph。核心学习点智能体Agent初体验使用LangChain内置的智能体框架如create_react_agent创建一个能使用搜索工具和计算器的简单智能体。观察它的思考过程ReAct模式理解AgentExecutor如何驱动“思考-行动”循环。理解StateGraph这是LangGraph的核心类。学习如何定义一个State通常使用TypedDict它包含了工作流运行过程中所有需要流转的数据。例如对于一个写作助手State里可能有topic,outline,draft,feedback,final_content等字段。定义节点Node和边Edge节点是一个接收State、返回一个更新后的State或State修改指令的函数。边决定了从一个节点出来后下一步该去哪个节点。Conditional Edge条件边允许你根据State中的某个值动态选择路径。编译与运行将定义好的节点和边添加到StateGraph中然后调用graph.compile()将其编译成一个可执行对象。运行这个图对象并传入初始状态观察整个工作流的执行。一个最小化的LangGraph示例假设我们要做一个“决策-执行”工作流先让LLM生成一个计划然后决定是直接输出还是需要调用工具执行。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI # 1. 定义强类型状态 class State(TypedDict): input: str plan: str action: str result: str final_output: str # 2. 定义节点函数 def plan_node(state: State): 节点1生成计划 llm ChatOpenAI(modelgpt-4) message [HumanMessage(contentf请为以下任务制定一个步骤计划{state[input]})] response llm.invoke(message) return {plan: response.content} def decide_node(state: State): 节点2决定下一步行动 llm ChatOpenAI(modelgpt-4) message [HumanMessage(contentf基于这个计划{state[plan]}判断是否需要调用工具执行只需回答需要或不需要。)] response llm.invoke(message) action call_tool if 需要 in response.content else direct_output return {action: action} def tool_node(state: State): 节点3模拟工具调用 # 这里可以接入真实的工具 return {result: f已执行计划{state[plan]}模拟工具返回成功。} def output_node(state: State): 节点4生成最终输出 if state[action] call_tool: content f计划{state[plan]}\n执行结果{state[result]} else: content f计划{state[plan]}\n无需额外工具执行。 return {final_output: content} # 3. 构建图 graph_builder StateGraph(State) graph_builder.add_node(plan, plan_node) graph_builder.add_node(decide, decide_node) graph_builder.add_node(tool, tool_node) graph_builder.add_node(output, output_node) # 4. 设置边 graph_builder.set_entry_point(plan) graph_builder.add_edge(plan, decide) # 条件边根据decide节点的输出决定去向 graph_builder.add_conditional_edges( decide, lambda state: state[action], # 路由函数返回下一个节点的名称 { call_tool: tool, direct_output: output } ) graph_builder.add_edge(tool, output) graph_builder.add_edge(output, END) # 5. 编译图 graph graph_builder.compile() # 6. 运行图 initial_state {input: 写一份关于LangGraph的技术博客大纲} result graph.invoke(initial_state) print(result[final_output])这个简单的例子展示了定义状态、节点、条件分支和编译运行的全过程。通过这个练习你会对LangGraph的编程模型有一个直观的认识。3.3 第三阶段构建复杂的多角色智能体工作流这是LangGraph真正大放异彩的地方。你可以构建模拟一个团队协作的AI系统。核心学习点多角色定义为不同的AI角色定义不同的系统提示词System Prompt和工具集。例如一个“研究员”角色擅长搜索和分析一个“写手”角色擅长文案创作一个“评审员”角色擅长挑刺和优化。状态共享与通信这些角色如何协作答案是通过共享的State。研究员将找到的资料存入State写手从State中读取资料进行创作评审员将修改意见再写回State。State成为角色间唯一的、结构化的通信渠道。循环与终止条件工作流可以设计成循环。例如“写手”写完初稿后交给“评审员”评审评审员提出意见后流程可以自动跳回“写手”节点进行修改。这个循环可以设置一个最大次数或者当评审意见为“通过”时才流向最终节点。人工干预节点在关键节点如最终发布前可以设置一个“人工审核”节点将State中的内容暂停并等待外部输入实现人机协同。实操心得在设计多角色工作流时定义清晰的State结构是成功的一半。在动手写代码前最好先用纸笔画出角色关系图和数据流图明确每个角色需要读写State中的哪些字段。一个混乱的State定义会让后续的开发和调试变得异常痛苦。建议为State的每个字段都添加详细的注释说明其用途和格式。4. 关键配置与高级特性详解掌握了基本构建方法后我们需要深入一些关键配置和高级特性它们能极大提升应用的鲁棒性和用户体验。4.1 状态State管理的艺术State是LangGraph的血液它的设计至关重要。1. 使用TypedDict和Annotated进行强类型定义这不仅是代码规范更是利用IDE自动补全和类型检查来减少错误。Annotated可以用来添加更丰富的语义。from typing import TypedDict, Annotated, List from typing_extensions import Literal from langgraph.graph import add_messages class CollaborativeState(TypedDict): # 使用add_messages注解自动管理消息历史 messages: Annotated[List, add_messages] # 明确标注字段用途 research_materials: Annotated[List[str], 研究员收集的原始材料] draft: Annotated[str, 写手生成的草稿] feedback: Annotated[List[str], 评审员提出的反馈列表] current_stage: Annotated[Literal[research, writing, reviewing, final], 当前工作流阶段] iteration_count: Annotated[int, 当前循环次数用于防止无限循环]2. 状态的初始化与校验在graph.invoke()时你需要传入初始状态。对于可选字段要做好默认值处理。可以在第一个节点中加入状态校验逻辑。3. 状态持久化对于长时间运行或需要中断恢复的工作流需要将State持久化到数据库如SQLite、PostgreSQL。LangGraph本身不负责持久化但你可以很容易地在节点间插入保存/加载状态的节点或者使用checkpointer特性。4.2 条件边Conditional Edges与路由逻辑条件边是实现动态工作流的核心。路由函数应该保持简单、纯粹只基于State做判断。def route_after_review(state: CollaborativeState): 在评审节点之后的路由逻辑 if not state[feedback]: # 没有反馈直接通过 return finalize elif state[iteration_count] 3: # 循环超过3次强制结束 return finalize elif 重大修改 in state[feedback][-1]: # 有重大修改意见返回写手节点 return writer else: # 一般性修改意见也返回写手节点 return writer # 在构建图时添加条件边 graph_builder.add_conditional_edges( reviewer, route_after_review, { writer: writer_node, finalize: finalize_node } )提示路由函数的返回值必须与你在add_conditional_edges中提供的映射键名完全一致。建议将节点名称定义为常量以避免拼写错误。4.3 错误处理与中断Interrupt机制真实的AI应用必须考虑错误处理。LangGraph提供了几种方式节点内部的Try-Except在每个节点函数内部进行细致的异常捕获并选择是返回错误信息到State还是抛出特定异常。Graph层面的错误处理在编译图时可以设置interrupt_before或interrupt_after在特定节点前后插入中断点以便在出现错误时执行自定义的清理或通知逻辑。设置超时和重试对于调用外部API如LLM、搜索工具的节点务必设置超时和重试机制。可以使用tenacity库或LangChain内置的Runnable重试配置。from langchain_core.runnables import RunnableConfig from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def call_llm_with_retry(prompt, config: RunnableConfig): llm ChatOpenAI(modelgpt-4, timeout30) # 设置超时 return llm.invoke(prompt, configconfig)4.4 与LangChain生态的深度集成LangGraph不是孤立的它与LangChain的其他部分无缝集成。节点可以是任何Runnable这意味着一个LangChainChain、一个Tool、甚至一个简单的PromptTemplate | LLM组合都可以直接作为一个Node函数的核心。这保护了你之前在LangChain上的投资。使用ToolNode和tools_condition对于简单的工具调用LangGraph提供了便捷的ToolNode和tools_condition来快速创建类似传统Agent的“思考-调用工具”的循环模式但这比原生Agent框架提供了更强的状态和流程控制。消息历史管理对于对话应用使用add_messages注解来管理messages字段是最佳实践。它能自动处理消息的追加并确保格式符合LLM要求。5. 实战构建一个带自我修订的博客写作助手让我们综合以上知识构建一个相对复杂的实战项目一个能自动完成研究、起草、自我评审和修订的博客写作助手。这个项目将涵盖多角色、条件循环、状态管理等核心概念。5.1 系统设计与状态定义角色设计研究员Researcher负责根据主题进行网络搜索假设有搜索工具收集并总结相关资料。写手Writer根据研究员提供的材料和主题撰写博客草稿。评审员Critic从逻辑、文笔、技术准确性等角度评审草稿提出修改意见。工作流设计开始 - 研究员 - 写手 - 评审员。评审员判断如果无需修改 - 结束如果需要修改 - 携带意见返回写手节点。写手根据意见修改然后再次进入评审环节。循环最多3次。达到循环上限或评审通过进入终稿润色节点然后结束。状态定义from typing import TypedDict, Annotated, List, Literal from langgraph.graph import add_messages class BlogWritingState(TypedDict): # 对话历史用于记录整个流程的“思考”过程 messages: Annotated[List, add_messages] # 用户输入的主题 topic: str # 研究员收集的参考资料摘要列表 research_summaries: List[str] # 写手生成的草稿 draft: str # 评审员给出的最新一轮反馈 last_feedback: str # 所有历史反馈 all_feedback: List[str] # 当前阶段 stage: Literal[research, writing, reviewing, polishing, finished] # 修订次数 revision_count: int # 最终成品 final_blog: str5.2 关键节点实现我们重点看一下“评审员”节点和路由逻辑的实现这是循环的核心。from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI import json llm ChatOpenAI(modelgpt-4-turbo, temperature0.2) def reviewer_node(state: BlogWritingState): 评审员节点评审草稿并生成反馈 review_prompt ChatPromptTemplate.from_messages([ (system, 你是一位严谨的技术博客评审员。请仔细评审以下博客草稿给出具体、可操作的修改意见。如果草稿质量合格无需修改请说‘通过’。), (human, 博客主题{topic}\n\n参考资料摘要{research}\n\n博客草稿{draft}) ]) chain review_prompt | llm feedback chain.invoke({ topic: state[topic], research: \n.join(state[research_summaries]), draft: state[draft] }).content # 更新状态 new_state { stage: reviewing, last_feedback: feedback, all_feedback: state[all_feedback] [feedback] } return new_state def route_after_review(state: BlogWritingState): 评审后的路由逻辑 feedback state[last_feedback].lower() # 条件1评审员说“通过” if 通过 in feedback: return to_polishing # 条件2已达到最大修订次数3次 elif state[revision_count] 3: print(已达到最大修订次数3次强制进入终稿润色。) return to_polishing # 条件3其他情况返回写手节点进行修改 else: # 增加修订计数 new_state {revision_count: state[revision_count] 1} # 这里返回一个元组第一个元素是下一个节点名第二个元素是对状态的更新 # 注意在LangGraph中更标准的做法是在‘writer_node’里读取revision_count # 或者在进入‘writer_node’的边上通过函数更新状态。这里为演示简化逻辑。 # 实际应用中建议在writer_node函数内部根据state[revision_count]判断是初次写作还是修改。 return to_writer5.3 图的构建与执行from langgraph.graph import StateGraph, END # 初始化图构建器 workflow StateGraph(BlogWritingState) # 添加节点 (假设 researcher_node, writer_node, polisher_node 已定义) workflow.add_node(researcher, researcher_node) workflow.add_node(writer, writer_node) workflow.add_node(reviewer, reviewer_node) workflow.add_node(polisher, polisher_node) # 设置入口点 workflow.set_entry_point(researcher) # 添加边 workflow.add_edge(researcher, writer) workflow.add_edge(writer, reviewer) # 添加条件边 workflow.add_conditional_edges( reviewer, route_after_review, # 上面定义的路由函数 { to_writer: writer, to_polishing: polisher } ) workflow.add_edge(polisher, END) # 编译图 app workflow.compile() # 运行图 initial_state { topic: LangGraph如何革新大语言模型应用开发, research_summaries: [], draft: , last_feedback: , all_feedback: [], stage: research, revision_count: 0, final_blog: , messages: [] # 初始化消息列表 } # 运行并可视化状态变化这是一个简化示例实际可视化需要更多工具 final_state app.invoke(initial_state) print(f最终博客内容:\n{final_state[final_blog][:500]}...) # 打印前500字符 print(f修订次数: {final_state[revision_count]}) print(f所有反馈: {final_state[all_feedback]})通过这个实战项目你可以完整地体验到使用LangGraph构建一个具备循环和自我修正能力的AI应用的整个过程。它比传统的线性链强大得多结构也清晰得多。6. 调试、优化与部署经验谈构建复杂的图工作流调试是不可避免的挑战。以下是我在实践中总结的一些有效方法。6.1 调试技巧与可视化使用graph.get_graph().draw_mermaid()这是最直接的调试工具。它可以将你的图生成Mermaid代码你可以在支持Mermaid的Markdown编辑器如Typora、Obsidian或在线工具中渲染出可视化的流程图。一眼就能看出节点和边的连接关系是否正确尤其是条件边的逻辑。检查输入输出I/O在每个节点的函数开头和结尾打印state的关键字段。LangGraph也支持更正式的日志记录。确保数据在节点间传递时格式和内容符合预期。简化与分阶段测试不要一次性构建完整的复杂图。先构建一个只有2-3个节点的最小可行图跑通流程。然后逐步添加节点和边每加一部分都进行测试。利用State的快照在出现奇怪行为时保存出错前一步的State完整内容可以用json.dumps()打印然后用这个State单独测试出错的节点函数进行隔离调试。6.2 性能优化要点LLM调用优化这是最大的开销来源。缓存对相同的提示词使用LLM调用缓存如LangChain的InMemoryCache或RedisCache。在研究和评审环节相似的查询可能会重复出现。批处理如果可能将多个独立的小查询合并成一个批处理提示词一次调用LLM完成。模型选择在不需要顶级创造性的环节如信息提取、简单分类使用更小、更快的模型如gpt-3.5-turbo。避免不必要的循环仔细设计循环终止条件。像上面的博客助手设置最大修订次数是防止因评审员过于严苛或LLM“抽风”导致无限循环的必要措施。异步执行如果图中有可以并行执行的独立节点例如同时研究多个子主题可以探索使用async节点和LangGraph的异步执行能力来提升整体速度。6.3 部署与生产化考量状态持久化对于长时间运行或服务化的应用必须将State持久化到外部存储如数据库。你可以创建一个“检查点”节点定期或将State保存到DB。当需要恢复时从DB加载State并重新注入图中。错误恢复与重试如前所述为网络调用和LLM调用设置重试机制。考虑在图的层面设置一个“全局异常处理”节点作为所有错误边的终点进行统一日志记录和告警。API化使用FastAPI或Flask将编译好的app即graph.compile()的结果包装成HTTP API。注意每次调用app.invoke()应该是无状态的或者通过会话ID来关联持久化的状态。监控与可观测性记录每个工作流执行的trace_id记录每个节点的开始时间、结束时间、输入输出快照注意脱敏。这有助于分析性能瓶颈和调试线上问题。踩坑提醒在生产环境中要特别注意LLM API的速率限制和费用。可以在节点中加入令牌token使用量的估算和记录并在接近限额时触发告警或降级策略。另外传递给LLM的提示词中的用户输入一定要做好严格的清洗和过滤防止提示词注入攻击。从LangChain 1.0到LangGraph 1.0的学习是一个从“组装流水线”到“设计智能工厂”的思维跃迁。它要求开发者不仅关心单个组件的功能更要关注整个系统的状态流转和决策逻辑。这个过程虽然有挑战但当你看到自己设计的AI工作流像精密的仪器一样自动运转、迭代并产出高质量结果时那种成就感是无可比拟的。希望这份基于“BrandPeng/Langchain1.0-Langgraph1.0-Learning”项目的延伸解读和实战分享能成为你探索这一强大工具的有力踏板。