1. 先搞清楚 LangGraph 到底解决了什么 Agent 开发痛点如果你正在用 LangChain 或者类似的框架做 AI 应用尤其是涉及多步骤、有状态、需要协作的智能体Agent大概率会遇到几个头疼的问题任务流程一复杂代码就变成面条状态管理混乱Agent 之间不知道谁干了什么想加个循环、分支或者回退得写一堆胶水逻辑。LangGraph 就是来解决这些问题的。它不是要替代 LangChain而是 LangChain 生态里一个专门用来编排复杂、有状态工作流的库。你可以把它理解为一个专门为 AI Agent 设计的“流程图引擎”。它最核心的价值是让你能用声明式的方式清晰地定义 Agent 的执行步骤、状态流转和决策逻辑而不是在回调函数里硬编码。所以这篇文章适合两类人看一是已经用过 LangChain 做简单 Agent但一遇到复杂流程就感觉力不从心的开发者二是刚开始接触 Agent 开发想直接从更结构化的方式入手避免后期重构的人。最关键的一点是LangGraph 让你关注“流程设计”本身而不是陷入状态同步和流程控制的细节泥潭。2. 环境准备与核心概念拆解State、Node、Edge在动手写代码之前得先把环境搭好并且理解 LangGraph 里最关键的三个概念。这能帮你避开后面很多理解上的坑。2.1 基础环境与依赖安装LangGraph 通常和 LangChain 一起使用。假设你有一个干净的 Python 环境3.8我建议先创建一个虚拟环境然后安装核心包。# 创建并激活虚拟环境以 conda 为例 conda create -n langgraph-demo python3.10 conda activate langgraph-demo # 安装 LangChain 和 LangGraph。注意版本建议安装较新的稳定版。 # 这里不指定具体版本因为更新快以官方最新推荐为准。 pip install langchain langgraph # 通常还需要一个 LLM 的集成比如 OpenAI 或本地模型。这里以 OpenAI 为例需要你有 API Key pip install openai安装完成后不要急着跑复杂例子。先验证基础导入是否正常写个最简单的脚本test_import.pyfrom langgraph.graph import StateGraph, END print(导入成功基础环境OK。)能跑通说明环境没问题。这里最容易忽略的是 Python 版本和包冲突如果你之前装过很多 AI 相关的包最好用新环境。2.2 理解 State工作流的“共享内存”在 LangGraph 里State是整个工作流运行时所有节点Node都能读取和修改的共享数据。它通常是一个字典Dict或者 Pydantic 模型。这是和 LangChain 里简单的链式调用最大的不同——状态是显式声明和管理的。比如你要做一个客服 Agent状态里可能包含user_query: 用户当前问题conversation_history: 历史对话current_step: 当前执行到哪一步final_answer: 最终生成的答案needs_human: 是否需要转人工定义状态就是定义你的工作流需要关心哪些数据。我一般会先用一个简单的 TypedDict 来定义清晰明了from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): 定义智能体的状态 input: str # 用户输入 steps: List[str] # 已执行的步骤记录 knowledge: str # 检索或生成的知识 answer: str # 最终答案2.3 理解 Node 和 Edge流程的“步骤”与“路线”Node节点就是工作流中的一个步骤它是一个函数。这个函数接收当前的State执行一些操作比如调用 LLM、查询数据库、进行计算然后返回更新后的State。Edge边决定了流程的走向。一个节点执行完后下一步该去哪个节点这由边来决定。边可以是固定的比如总是去节点B也可以是条件判断conditional edge根据 State 里的某个值来决定下一步。把 State、Node、Edge 串起来就是一个完整的图Graph。LangGraph 会帮你处理节点间的状态传递和路由你只需要定义好每个节点做什么以及它们之间的连接规则。3. 从零构建你的第一个 LangGraph 智能体工作流理论讲多了容易晕我们直接动手用一个经典的“研究助手”智能体作为例子。这个智能体的任务是接收一个复杂问题先规划步骤然后依次执行每个步骤比如搜索、总结最后合成答案。3.1 定义状态与初始化图首先我们定义这个研究任务需要的状态。它比刚才的例子稍复杂一点。from typing import TypedDict, List, Optional from langgraph.graph import StateGraph, END class ResearchState(TypedDict): 研究任务的状态 original_query: str # 原始问题 plan: List[str] # 规划出的步骤列表 completed_steps: List[str] # 已完成的步骤 findings: List[str] # 每个步骤的发现 final_report: Optional[str] # 最终报告然后我们初始化一个图# 创建一个状态图并指定状态的结构 workflow StateGraph(ResearchState)3.2 创建节点函数规划、执行、汇总接下来创建三个节点函数。为了简化我们这里用模拟的 LLM 调用mock_llm_call来代替真实的 API 调用重点看流程。def planner_node(state: ResearchState) - ResearchState: 规划节点将复杂问题拆解为步骤 query state[“original_query”] # 模拟 LLM 生成规划 # 例如对于“如何学习深度学习”可能规划为[“1. 了解数学基础” “2. 学习编程框架” “3. 完成实战项目”] mock_plan [f“步骤{i}: 研究[{query}]的子主题{i}” for i in range(1, 4)] return {“plan”: mock_plan} def researcher_node(state: ResearchState) - ResearchState: 研究节点执行一个具体的研究步骤 # 假设我们按顺序执行计划中的第一个未完成步骤 all_steps state[“plan”] completed state.get(“completed_steps”, []) findings state.get(“findings”, []) next_step_index len(completed) if next_step_index len(all_steps): # 所有步骤已完成 return {“completed_steps”: completed, “findings”: findings} current_step all_steps[next_step_index] # 模拟执行研究例如调用搜索API或阅读文档 mock_finding f“关于{current_step}的模拟研究发现。” completed.append(current_step) findings.append(mock_finding) return {“completed_steps”: completed, “findings”: findings} def reporter_node(state: ResearchState) - ResearchState: 报告节点汇总所有发现生成最终报告 findings state[“findings”] query state[“original_query”] # 模拟 LLM 汇总发现 mock_report f“关于‘{query}’的研究报告\n” “\n”.join(findings) return {“final_report”: mock_report}3.3 将节点加入图并连接创建好函数后把它们作为节点添加到图中并指定连接关系。# 添加节点并给每个节点起个名字 workflow.add_node(“planner”, planner_node) workflow.add_node(“researcher”, researcher_node) workflow.add_node(“reporter”, reporter_node) # 设置边的连接关系 workflow.set_entry_point(“planner”) # 工作流从“规划器”开始 workflow.add_edge(“planner”, “researcher”) # 规划完后去研究 workflow.add_edge(“researcher”, “reporter”) # 研究完后去报告 workflow.add_edge(“reporter”, END) # 报告完成后结束 # 编译图得到一个可执行的对象 app workflow.compile()3.4 运行工作流并查看结果现在我们可以像调用函数一样运行这个编译好的图。传入初始状态。# 定义初始状态 initial_state ResearchState(original_query“如何学习LangGraph”, plan[], completed_steps[], findings[], final_reportNone) # 运行工作流 final_state app.invoke(initial_state) print(“最终报告”) print(final_state[“final_report”]) print(“\n执行步骤记录”) print(final_state[“completed_steps”])运行这段代码你会看到它按照“规划 - 研究多次循环- 报告”的流程执行并输出结果。这里的关键是我们并没有在代码里写循环但researcher_node会被多次调用直到计划中的所有步骤完成。这是因为我们将在下一步引入“循环”的概念。4. 实现动态工作流循环、条件分支与长期记忆基础线性流程用处有限。LangGraph 的强大在于能轻松实现循环比如一直研究直到满足条件和条件分支比如根据结果决定是继续还是转人工。4.1 实现静态循环让研究节点执行多次在上面的例子中研究节点只执行了一次。我们需要修改图的逻辑让研究节点在执行完一个步骤后判断是否还有未完成的步骤如果有就回到自己形成循环。这需要修改边的连接方式使用add_conditional_edges。首先我们修改researcher_node让它除了更新状态还返回一个标志指示是否全部完成。def researcher_node_with_flag(state: ResearchState) - dict: 研究节点返回更新后的状态和一个完成标志 all_steps state[“plan”] completed state.get(“completed_steps”, []) findings state.get(“findings”, []) next_step_index len(completed) if next_step_index len(all_steps): # 所有步骤已完成返回标志 “done” return {“updates”: {“completed_steps”: completed, “findings”: findings}, “next”: “done”} current_step all_steps[next_step_index] mock_finding f“执行{current_step}的详细发现。” completed.append(current_step) findings.append(mock_finding) # 还有步骤返回标志 “continue” return {“updates”: {“completed_steps”: completed, “findings”: findings}, “next”: “continue”}然后我们需要一个路由函数根据researcher_node返回的标志来决定下一步去哪。def route_after_research(state: ResearchState) - str: 根据研究节点的结果决定路由 # 这里我们需要一个约定研究节点把下一个动作放在 state 的某个字段里。 # 为了简化我们假设研究节点直接在 state 里设置了一个 should_continue 布尔值。 # 在实际更复杂的图中可能用 state[“_next”] 这种约定俗成的字段。 if state.get(“_should_continue”, True): return “researcher” # 继续研究 else: return “reporter” # 去生成报告修改图的构建方式使用条件边workflow StateGraph(ResearchState) workflow.add_node(“planner”, planner_node) workflow.add_node(“researcher”, researcher_node) # 注意这里用回最初版本的 researcher_node但逻辑需调整 workflow.add_node(“reporter”, reporter_node) workflow.set_entry_point(“planner”) workflow.add_edge(“planner”, “researcher”) # 关键为 researcher 节点添加条件边 workflow.add_conditional_edges( “researcher”, # 路由函数根据当前 state 决定下一个节点 lambda state: “researcher” if len(state[“completed_steps”]) len(state[“plan”]) else “reporter” ) workflow.add_edge(“reporter”, END) app workflow.compile()现在当你运行app.invoke时researcher节点会一直循环调用自己直到completed_steps的数量等于plan的数量然后自动跳转到reporter节点。这就是一个静态循环循环次数在规划时就确定了。4.2 实现动态条件分支根据结果决定流程更复杂的情况是流程走向需要根据 LLM 生成的内容或外部条件动态决定。例如在研究过程中如果发现信息不足可能转向“请求澄清”节点而不是继续研究。这需要我们在节点函数里做出判断并修改状态中的一个“路由指令”字段。class DynamicResearchState(TypedDict): original_query: str plan: List[str] completed_steps: List[str] findings: List[str] needs_clarification: bool # 新增是否需要澄清 clarification_question: str # 新增澄清问题 final_report: Optional[str] def smart_researcher_node(state: DynamicResearchState) - DynamicResearchState: 智能研究节点可能触发澄清 # ... 模拟研究过程 ... current_finding “一些研究发现...” # 模拟一个判断如果发现内容太模糊则请求澄清 if “模糊” in current_finding: # 这里用简单字符串模拟LLM判断 return { “needs_clarification”: True, “clarification_question”: “您能具体说明一下XX方面吗” } else: # 正常记录发现并继续 new_findings state.get(“findings”, []) [current_finding] new_completed state.get(“completed_steps”, []) [“一次研究”] return { “findings”: new_findings, “completed_steps”: new_completed, “needs_clarification”: False } def clarifier_node(state: DynamicResearchState) - DynamicResearchState: 澄清节点模拟获取用户澄清 # 在实际应用中这里可能会暂停工作流等待外部输入。 # 我们模拟用户提供了澄清。 print(f“Agent请求澄清: {state[‘clarification_question’]}”) # 假设我们收到了澄清并更新查询 updated_query state[“original_query”] “ (已澄清)” return { “original_query”: updated_query, “needs_clarification”: False, “clarification_question”: “” }然后构建一个包含分支的图workflow StateGraph(DynamicResearchState) workflow.add_node(“planner”, planner_node) workflow.add_node(“researcher”, smart_researcher_node) workflow.add_node(“clarifier”, clarifier_node) workflow.add_node(“reporter”, reporter_node) workflow.set_entry_point(“planner”) workflow.add_edge(“planner”, “researcher”) # researcher 后的路由根据是否需要澄清来决定 workflow.add_conditional_edges( “researcher”, lambda s: “clarifier” if s.get(“needs_clarification”, False) else (“reporter” if len(s.get(“completed_steps”, [])) 3 else “researcher”) # 这个路由函数如果需要澄清去clarifier否则如果完成步骤3去reporter否则继续研究。 ) workflow.add_edge(“clarifier”, “researcher”) # 澄清后回到研究者 workflow.add_edge(“reporter”, END)这个图就实现了动态分支研究过程中可以跳出主流程去请求澄清澄清后再回来继续研究。4.3 关于“长期记忆”的实现思路热搜词里有“langgraph 长期记忆”。在 LangGraph 中状态State本身就可以作为短期记忆在一次工作流执行过程中持续存在。而“长期记忆”通常指需要持久化到数据库如 Redis、PostgreSQL或向量存储中的信息供多次工作流调用共享。实现长期记忆一般有两种模式在节点函数中访问外部存储在planner_node或researcher_node里去查询向量数据库获取相关历史记忆并放入当前 State 中。使用 LangGraph 的检查点Checkpoint机制这是更高级的功能。它允许你将工作流在任意节点暂停持久化状态稍后从该点恢复。这对于需要等待人工审核、长时间运行或可中断的任务非常有用。这通常涉及配置一个持久化存储后端。对于大多数应用第一种模式节点内访问数据库更简单直接。你可以把记忆检索和更新封装成独立的节点或工具函数。5. 生产级考量错误处理、并发与监控当你的智能体从 Demo 走向实际应用时有三个问题必须提前考虑出错怎么办、慢怎么办、怎么知道它正在干什么。5.1 错误处理与重试节点函数可能因为网络、API限制、脏数据等原因失败。LangGraph 本身不自动处理错误需要你在节点函数内部或外部包装层实现。建议做法节点内部 try-catch在每个节点函数内部进行细致的异常捕获根据异常类型决定是重试、跳过还是标记失败并路由到特定“错误处理节点”。使用 LangChain 的Runnable组件LangChain 的很多组件如ChatOpenAI本身支持配置重试。将 LLM 调用封装成Runnable再在节点中调用能利用其重试机制。设置超时对于可能长时间挂起的操作如网络请求一定要设置超时。可以在节点函数内用asyncio.wait_for或使用支持超时的客户端。from langchain_core.runnables import RunnableConfig import asyncio from openai import APITimeoutError def robust_llm_node(state: State): try: # 假设 llm_chain 是一个 LangChain Runnable result llm_chain.invoke(state[“query”], configRunnableConfig(timeout30)) state[“answer”] result state[“_error”] None except APITimeoutError: state[“_error”] “LLM API 超时” state[“_should_retry”] True except Exception as e: state[“_error”] str(e) state[“_should_retry”] False return state然后你的条件路由函数可以检查state[“_error”]和state[“_should_retry”]决定是重试、告警还是终止。5.2 并发执行与性能LangGraph 的图是按定义顺序执行的一个节点做完才到下一个。但这不意味着不能并发。节点内并发你可以在一个节点函数内部使用asyncio.gather并发执行多个独立任务比如同时调用多个不同的 API 或查询多个数据库合并结果后再更新状态。子图并发对于可以完全并行执行的分支你可以将它们设计成不同的子图然后通过更高级的编排模式如使用StateGraph的并行分支或在外层用asyncio驱动多个图实例来实现。但这通常需要更精细的设计。对于 IO 密集型的 Agent大量网络请求节点内并发是提升性能最直接有效的方法。不要把并发的希望寄托在图引擎自动帮你做而要自己在业务节点里实现。5.3 日志、追踪与监控调试一个复杂的工作流没有清晰的日志是灾难。LangGraph 与 LangSmithLangChain 的官方监控平台集成非常好。基础日志在每个节点的开始和结束打印或记录关键状态字段。使用 Python 的logging模块为不同节点设置不同的 logger。集成 LangSmith设置环境变量LANGCHAIN_TRACING_V2true和LANGCHAIN_API_KEY。运行你的 LangGraph App 时所有步骤的输入输出、LLM 调用都会自动记录到 LangSmith可以可视化查看执行轨迹、耗时和 token 消耗。这是生产调试和优化的利器。自定义监控点在状态中增加_start_time,_end_time字段在关键节点计算耗时。将最终状态或错误信息推送到你的监控系统如 Prometheus, Sentry。6. LangGraph vs LangChain 与其他框架怎么选这是被问得最多的问题。简单来说LangChain是一个大工具箱提供了连接 LLM、数据、工具的各种组件Models, Indexes, Chains, Agents。它的AgentExecutor也能跑简单 Agent但复杂流程的编排能力弱。LangGraph是 LangChain 生态里的一个流程编排器。它擅长管理有状态、多步骤、带循环和分支的复杂工作流。它通常和 LangChain 的其他组件一起使用。其他 Workflow 框架如 Prefect, Airflow是通用的任务编排框架更侧重于调度、依赖管理和运维。它们也能跑 AI 任务但不像 LangGraph 那样原生为 Agent 的状态和决策逻辑设计需要更多胶水代码。Dify, Coze 等平台是低代码/无代码的 AI 应用构建平台提供了可视化编排界面。如果你不想写代码它们是很好的选择。LangGraph 则提供了代码级的灵活性和控制力。选择建议如果你的任务只是简单的“用户输入 - LLM - 输出”用 LangChain 的Chain或简单Agent就够了。如果你的任务需要多步决策、循环、内部状态记忆、复杂的工具调用流程果断用 LangGraph。如果你需要严格的定时调度、跨大量机器的任务分发、复杂的重试策略可以考虑在 LangGraph 上层再套用 Prefect 或 Airflow。如果你想快速原型验证且不关心代码部署可以用 Dify 这类平台。最后也是我最想强调的一点不要被“智能体”这个词吓到也不要追求一步到位做出拥有“长期记忆”、“自我反思”能力的超级 Agent。从解决一个具体的、有明确步骤的业务流程开始用 LangGraph 把它清晰地定义出来跑通它监控它然后再迭代增加复杂性。先让一个工作流稳定可靠远比堆砌炫酷但不稳定的功能更有价值。在真实项目中输入数据的清洗、异常边界的处理、日志的完备性往往比 Agent 本身的算法更能决定项目的成败。