1. 从LangChain到LangGraph为什么我们需要一个新的“图”如果你最近在折腾AI应用尤其是想搞点能自主决策、有工作流的智能体Agent那你大概率绕不开LangChain。LangChain确实是个好东西它把大语言模型LLM和外部工具、数据源连接起来让“提示词工程”变成了“应用工程”。但当你真的开始用它构建一个稍微复杂点的Agent比如一个能根据用户问题自动选择工具、执行、检查结果、再决定下一步的客服机器人时你可能会遇到一些麻烦。最典型的麻烦就是状态管理。在LangChain的早期Agent实现里整个执行流程更像是一个“黑盒”。你定义好工具和Agent调用run()然后等待结果。中间发生了什么Agent的思考过程Thought、执行的动作Action、观察到的结果Observation这些状态虽然能看到日志但如果你想在流程中插入一个自定义的步骤或者根据中间状态动态改变流程走向就会变得非常棘手。代码结构容易变成一堆if-else的嵌套可读性和可维护性直线下降。这就是LangGraph诞生的背景。它不是要取代LangChain而是LangChain生态中的一个专门用于构建有状态、多步骤工作流的框架。你可以把它理解成LangChain的“神经系统”或“流程引擎”。如果说LangChain提供了构建Agent所需的“器官”模型、工具、记忆那么LangGraph则定义了这些“器官”如何协同工作的“神经连接”和“反射弧”。它的核心思想非常直观用“图”Graph来建模工作流。图中的节点Node代表一个执行单元比如调用LLM、运行工具、检查条件边Edge定义了节点之间的流转逻辑。整个Agent的运行就是在这个图上根据当前的状态State沿着边从一个节点“走”到下一个节点。这种范式带来了几个立竿见影的好处可视化与可调试性整个工作流就是一张图一目了然。你可以清晰地看到Agent的决策路径哪里是分支哪里是循环调试起来不再是雾里看花。灵活的状态管理LangGraph有一个核心的State概念它是一个共享的数据结构在整个图的执行过程中传递和更新。所有节点都读写这个State这使得跨步骤的信息传递变得异常简单和规范。复杂的控制流实现循环比如让Agent反复尝试直到成功、条件分支根据结果走不同路径、甚至并行执行在图的模型下都变得非常自然。这是传统线性链式调用难以优雅实现的。所以当你看到“学习Agent开发”和“langgraph”出现在一起时背后的潜台词是你想构建的已经不是简单的单次问答机器人而是一个具备一定自主性、能处理复杂任务流程的智能体。LangGraph就是你实现这个目标的“脚手架”和“设计图”。2. LangGraph核心三要素State、Node、Edge拆解要玩转LangGraph必须吃透它的三个核心概念状态State、节点Node和边Edge。这三者构成了LangGraph应用的骨架。2.1 State工作流的共享记忆体State是LangGraph中最重要的概念没有之一。它是一个类似字典Dict的结构在整个图执行期间持续存在并被所有节点访问和修改。你可以把它想象成Agent的“短期工作记忆”或流程的“上下文白板”。定义State通常使用TypedDict类型化字典或Pydantic模型这能带来良好的类型提示和校验。一个典型的Agent State可能包含这些字段from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 消息历史记录用户、AI、工具之间的对话 messages: Annotated[List, add_messages] # 用户当前输入的问题 input: str # Agent的“思考”过程 reasoning: str # 下一步要采取的动作工具名 next_action: str # 工具执行的结果 tool_output: str # 迭代次数用于防止无限循环 iteration: int这里有个关键点messages字段使用了Annotated[List, add_messages]。add_messages是一个归约器Reducer这是LangGraph处理状态更新的一个精妙设计。当多个节点都可能向messages列表添加内容时归约器定义了如何合并这些更新。add_messages会确保消息按顺序合并而不是简单地覆盖。这解决了并发或异步场景下的状态冲突问题。实操心得在设计State时一定要想清楚哪些数据是需要跨节点共享的。不要一股脑把所有东西都塞进State这会让State变得臃肿且难以理解。通常输入、中间结果、最终输出、以及控制流程的标志位如should_continue是State的常客。使用Pydantic能进行更严格的字段校验比如限制iteration最大值防止死循环。2.2 Node执行具体任务的单元Node就是一个普通的Python函数或可调用对象它接收当前的State作为参数并返回一个对该State的更新一个字典。def call_llm(state: AgentState) - dict: 节点调用大语言模型生成思考过程和建议动作。 # 1. 从State中获取消息历史 chat_history state[“messages”] user_input state[“input”] # 2. 构造给LLM的提示词 prompt f 基于以下对话历史和用户最新问题请进行思考并决定下一步行动。 历史对话{chat_history} 用户问题{user_input} 请先输出你的推理过程Reasoning然后输出一个建议的动作Action。动作必须是以下之一search_web, calculate, final_answer。 # 3. 模拟调用LLM实际中会接入OpenAI、Anthropic等API # 这里用一个简单逻辑模拟 if “天气” in user_input: reasoning “用户询问天气我需要使用搜索工具获取实时信息。” next_action “search_web” elif “计算” in user_input: reasoning “用户需要计算我应使用计算器工具。” next_action “calculate” else: reasoning “这是一个通用问题我可以直接回答。” next_action “final_answer” # 4. 返回对State的更新 return {“reasoning”: reasoning, “next_action”: next_action}关键规则Node函数返回的字典其键必须对应State中定义的字段。LangGraph会自动用这个返回的字典去更新全局的State。这就是状态流转的方式。2.3 Edge决定流程走向的规则Edge定义了在某个Node执行完毕后接下来应该去哪个Node。它有两种主要类型条件边Conditional Edge根据State中的某个值动态决定下一个节点。普通边Fixed Edge无条件地指向下一个节点。条件边是实现分支和循环的关键。它通常是一个函数读取State返回下一个目标节点的名称字符串。def decide_next_step(state: AgentState) - str: 条件边函数根据Agent建议的动作决定下一步去哪。 next_action state.get(“next_action”) if next_action “search_web”: return “web_search_node” # 跳转到搜索节点 elif next_action “calculate”: return “calculator_node” # 跳转到计算节点 elif next_action “final_answer”: return “generate_response_node” # 跳转到生成回答节点 else: return “handle_error_node” # 兜底跳转到错误处理节点通过组合Node和Edge你就能构建出复杂的流程。例如一个经典的ReActReasoning ActingAgent循环可以这样构建思考节点 - 条件边 - 工具执行节点 - 条件边 - 思考节点...直到next_action变为final_answer。3. 手把手构建你的第一个LangGraph智能体理论说得再多不如动手跑一遍。我们来构建一个简单的“研究助手”Agent它能根据用户问题决定是直接回答还是需要联网搜索。3.1 环境准备与依赖安装首先确保你的Python环境建议3.10然后安装核心库pip install langgraph langchain-openai这里我们安装langchain-openai是为了方便接入OpenAI的模型。LangGraph本身是模型无关的你可以用任何兼容的LLM。3.2 定义State与初始化图我们定义一个精简的State并创建图对象from typing import TypedDict, List, Annotated from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages # 1. 定义State class ResearchState(TypedDict): messages: Annotated[List, add_messages] user_query: str needs_search: bool search_result: str final_answer: str # 2. 初始化图构建器 builder StateGraph(ResearchState)StateGraph是构建图的入口我们传入了ResearchState作为状态类型。3.3 实现核心功能节点接下来我们创建三个节点节点A路由节点Router- 分析用户问题判断是否需要搜索。def router_node(state: ResearchState) - dict: 判断问题类型 query state[“user_query”].lower() # 简单规则问题包含“最新”、“今天”、“2024”等词或明显需要实时信息则搜索 need_search_keywords [“最新”, “今天”, “2024”, “股价”, “天气”, “新闻”] needs_search any(keyword in query for keyword in need_search_keywords) # 更新State return {“needs_search”: needs_search}节点B搜索节点Web Search- 模拟调用搜索工具。def search_node(state: ResearchState) - dict: 模拟网络搜索实际应接入SerperAPI、Tavily等工具 query state[“user_query”] # 模拟返回搜索结果 mock_result f”关于‘{query}’的模拟搜索结果这是一个非常热门的话题根据模拟数据当前的主流观点是...“ return {“search_result”: mock_result}节点C回答生成节点Answer Generator- 综合信息生成最终答案。from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-3.5-turbo”) def answer_node(state: ResearchState) - dict: 生成最终回答 base_info state[“user_query”] search_info state.get(“search_result”, “”) prompt f””” 用户问题{base_info} {‘以下是网络搜索结果’ search_info if search_info else ‘没有进行网络搜索。’} 请生成一个友好、准确的回答。 “”” response llm.invoke(prompt) return {“final_answer”: response.content}3.4 添加节点与边构建完整流程现在将节点添加到图中并用边连接它们# 添加节点 builder.add_node(“router”, router_node) builder.add_node(“search”, search_node) builder.add_node(“answer”, answer_node) # 设置入口点 builder.set_entry_point(“router”) # 添加条件边根据router的结果决定走向 def decide_after_router(state: ResearchState) - str: if state[“needs_search”]: return “search” # 需要搜索去search节点 else: return “answer” # 不需要搜索直接去answer节点 builder.add_conditional_edges( “router”, # 源节点 decide_after_router, # 条件判断函数 {“search”: “search”, “answer”: “answer”} # 可能的目标节点映射 ) # 添加固定边从search节点到answer节点 builder.add_edge(“search”, “answer”) # 设置answer节点为终点 builder.add_edge(“answer”, END) # 编译图得到可执行对象 research_agent builder.compile()3.5 运行与可视化现在你可以运行这个Agent了# 初始化状态 initial_state ResearchState(user_query“今天北京的天气怎么样”, messages[], needs_searchFalse, search_result“”, final_answer“”) # 运行图 final_state research_agent.invoke(initial_state) print(final_state[“final_answer”])LangGraph还有一个强大的功能可视化。你可以将图导出为PNG图片直观看到工作流。from IPython.display import Image, display try: display(Image(research_agent.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以保存到文件 research_agent.get_graph().draw_mermaid_png(output_file_path“research_agent_graph.png”)生成的图会清晰显示router- (条件分支) -search-answer或者router- (条件分支) -answer的路径。这对于复杂流程的沟通和调试是无可替代的。4. 深入LangGraph高级特性子图、持久化与多Agent协作当你掌握了基础构建后LangGraph的一些高级特性能让你的Agent变得更强大、更健壮、更易管理。4.1 子图Subgraph模块化与复用子图允许你将一个复杂的图的一部分封装成一个独立的、可复用的单元。这就像编程中的函数或类极大地提升了代码的模块化程度。假设我们上面例子中的search_node内部逻辑很复杂包含“查询构造”、“调用API”、“结果过滤”等多个步骤我们可以将其抽离成一个子图。from langgraph.graph import StateGraph as SubGraphBuilder # 1. 定义子图专用的State可以是主State的子集或全新结构 class SearchSubState(TypedDict): query: str raw_results: List[str] filtered_results: str # 2. 构建子图 sub_builder SubGraphBuilder(SearchSubState) def construct_query_node(state: SearchSubState) - dict: # ... 构造搜索查询逻辑 return {“query”: f”优化后的查询{state[‘query’]}”} def call_api_node(state: SearchSubState) - dict: # ... 调用搜索API return {“raw_results”: [“结果1”, “结果2”]} def filter_results_node(state: SearchSubState) - dict: # ... 过滤结果 return {“filtered_results”: “; “.join(state[“raw_results”])} # 添加节点和边 sub_builder.add_node(“construct”, construct_query_node) sub_builder.add_node(“call_api”, call_api_node) sub_builder.add_node(“filter”, filter_results_node) sub_builder.set_entry_point(“construct”) sub_builder.add_edge(“construct”, “call_api”) sub_builder.add_edge(“call_api”, “filter”) sub_builder.add_edge(“filter”, END) # 编译子图 search_subgraph sub_builder.compile() # 3. 在主图中将子图作为一个节点添加 builder.add_node(“advanced_search”, search_subgraph)在主图中advanced_search节点就是一个黑盒它接收主State中相关的部分执行内部复杂的子流程然后返回更新。这使主图结构保持清晰。4.2 检查点Checkpoint与持久化实现长期记忆与恢复这是LangGraph用于构建“长期记忆”Agent的核心。检查点机制允许你在图执行的任何节点处将完整的State序列化保存下来。这意味着对话记忆Agent可以记住跨越多次调用的完整对话历史。流程暂停与恢复一个长时间运行的任务如处理一个复杂工单可以中途暂停稍后从断点继续。错误恢复如果执行过程中出错可以从上一个检查点重试而不是从头开始。启用检查点需要配置一个持久化存储后端如内存、SQLite、PostgreSQL等。from langgraph.checkpoint.sqlite import SqliteSaver # 使用SQLite作为检查点存储 memory SqliteSaver.from_conn_string(“:memory:”) # 内存数据库实际可用文件路径 # 在编译图时传入检查点存储器 persistent_agent builder.compile(checkpointermemory) # 第一次调用传入一个config其中包含线程IDthread_id用于标识这次会话/流程 config {“configurable”: {“thread_id”: “user_123_session_1”}} initial_state {“user_query”: “帮我制定一个学习计划”} result1 persistent_agent.invoke(initial_state, configconfig) # 第二次调用使用相同的thread_idAgent会从上次的检查点状态继续。 # 注意State的更新是累积的。你需要处理好初始状态。 new_state_for_update {“user_query”: “把计划里的Python换成Go语言”} result2 persistent_agent.invoke(new_state_for_update, configconfig) # result2中会包含完整的对话历史包括第一次的交互。踩坑实录使用检查点时最大的陷阱是状态合并。第二次调用invoke传入的initial_state并不是重置状态而是作为更新合并到已有的检查点状态中。如果你的State里有messages字段并使用add_messages归约器这很完美。但如果有些字段你希望每次都是全新的比如user_query就需要在节点逻辑中小心处理或者通过config中的recursion_limit等机制控制。务必在开发时打印出每次调用前后的完整State以理解其变化过程。4.3 多Agent协作构建智能体团队LangGraph非常适合编排多个各司其职的Agent协同工作。例如你可以构建一个“写作团队”包含“策划Agent”、“调研Agent”、“写作Agent”和“校对Agent”。实现模式通常有两种层级式一个“主管Agent”也是一个LangGraph图负责接收任务然后通过条件边调用不同的“员工Agent”子图或节点。广播式所有Agent节点并行接收同一份State各自处理然后由一个“聚合节点”汇总结果。这里展示一个简单的层级式协作框架class TeamState(TypedDict): task: str plan: str research_data: str draft: str final_output: str team_builder StateGraph(TeamState) def manager_node(state: TeamState) - dict: 主管节点分解任务决定流程 task state[“task”] # 简单规则如果任务需要研究先调研再写作否则直接写作。 if “调研” in task or “分析” in task: return {“next”: “research”} else: return {“next”: “write”} def research_agent_node(state: TeamState) - dict: 调研Agent节点这里可以嵌入另一个复杂的子图 # 模拟调研 return {“research_data”: f”关于‘{state[‘task’]}’的调研摘要...”} def writing_agent_node(state: TeamState) - dict: 写作Agent节点 plan state.get(“plan”, “”) research state.get(“research_data”, “”) # 综合规划和调研结果进行写作 return {“draft”: f”根据计划{plan}和调研{research}生成的初稿...”} def review_agent_node(state: TeamState) - dict: 校对Agent节点 draft state[“draft”] # 模拟校对 return {“final_output”: f”校对后的终稿{draft}已优化”} # 添加节点 team_builder.add_node(“manager”, manager_node) team_builder.add_node(“researcher”, research_agent_node) team_builder.add_node(“writer”, writing_agent_node) team_builder.add_node(“reviewer”, review_agent_node) team_builder.set_entry_point(“manager”) # 主管根据任务决定派发给谁 def manager_decision(state: TeamState) - str: return state.get(“next”, “writer”) team_builder.add_conditional_edges(“manager”, manager_decision) # 定义工作流调研 - 写作 - 校对 team_builder.add_edge(“researcher”, “writer”) team_builder.add_edge(“writer”, “reviewer”) team_builder.add_edge(“reviewer”, END) team_graph team_builder.compile()通过这种方式你可以构建出非常复杂的多智能体系统每个智能体可以专注于单一能力由LangGraph图来负责它们之间的通信和协调。5. 避坑指南与性能优化实战在实际开发中你会遇到一些预料之外的问题。下面是我在项目中总结的几个关键坑点和优化建议。5.1 状态更新冲突与归约器的正确使用这是新手最容易出错的地方。当多个节点可能并发或在逻辑上修改State的同一个字段时你需要归约器来定义合并规则。错误示例class State(TypedDict): log: List[str] # 多个节点都想往这里添加日志 def node_a(state: State): return {“log”: [“Node A executed”]} # 直接赋值会覆盖 def node_b(state: State): return {“log”: [“Node B executed”]} # 直接赋值会覆盖如果node_a和node_b都执行log里最终只会有一条记录因为后一次更新覆盖了前一次。正确做法from typing import Annotated from langgraph.graph import add_messages # 用于消息列表 import operator class State(TypedDict): # 使用归约器用operator.add来合并列表 log: Annotated[List[str], operator.add] # 或者对于消息使用LangGraph提供的专用归约器 messages: Annotated[List, add_messages] def node_a(state: State): return {“log”: [“Node A executed”]} # 返回一个列表片段 def node_b(state: State): return {“log”: [“Node B executed”]} # 返回一个列表片段现在log字段会正确地将两次返回的列表合并为[“Node A executed”, “Node B executed”]。核心原则如果State中的字段可能被多个节点更新且更新方式不是简单的覆盖如列表追加、数值累加、字典合并就必须使用归约器。LangGraph内置了一些常用的如add_messages对于自定义合并逻辑你需要实现自己的归约器函数。5.2 循环控制与超时机制Agent工作流经常包含循环比如ReAct的思考-行动循环。你必须设置明确的终止条件否则图会无限运行下去。在State中设置“循环哨兵”class State(TypedDict): messages: Annotated[List, add_messages] next_action: str iteration: int # 迭代计数器 max_iterations: int # 最大迭代次数 def should_continue(state: State) - str: 条件边函数决定继续循环还是结束 if state[“next_action”] “final_answer”: return “end” elif state[“iteration”] state[“max_iterations”]: return “end_too_many_loops” # 跳转到处理超限的节点 else: return “continue” # 在循环的每个周期结束时更新迭代次数 def update_iteration(state: State): return {“iteration”: state[“iteration”] 1}确保在图中有一个节点如update_iteration在每次循环后递增iteration并且条件边函数should_continue会检查这个值。5.3 异步执行与性能优化当你的节点涉及网络调用如LLM API、工具API时同步执行会导致大量时间浪费在等待I/O上。LangGraph原生支持异步节点。将节点函数定义为async defimport asyncio from langchain_openai import AsyncChatOpenAI async_llm AsyncChatOpenAI(model“gpt-3.5-turbo”) async def async_llm_node(state: State): prompt state[“prompt”] # 使用异步客户端调用 response await async_llm.ainvoke(prompt) return {“response”: response.content}使用ainvoke异步运行整个图final_state await graph.ainvoke(initial_state)对于有多个独立I/O操作的节点你可以结合asyncio.gather在单个节点内实现并行或者设计图的结构使得可以并行的分支使用LangGraph的并行执行特性这通常需要更复杂的设计。5.4 调试与日志记录调试一个复杂的图可能很困难。除了可视化还有以下实用技巧打印State在每个节点的开始和结束打印State的关键部分。你可以创建一个装饰器来自动完成这个操作。使用langsmith如果你使用LangChain生态强烈建议集成LangSmith。它能追踪每一次图执行的所有步骤、输入、输出和耗时是调试和监控的神器。简化测试先用一个极简的State和Mock的LLM/工具来测试图的逻辑流确保所有边和条件判断正确再接入真实的、耗时的服务。6. LangGraph vs LangChain如何选择与结合使用最后我们来澄清一个常见的困惑LangGraph和LangChain到底是什么关系我该用哪个LangChain是一个构建LLM应用的全栈框架。它提供了模型抽象层统一接口调用不同LLM。提示词管理模板、示例选择器。数据连接文档加载器、向量库、检索器。工具集成将函数、API封装成Agent可用的工具。链Chains将多个组件按固定顺序组合的简单工作流。Agent旧版基于“AgentExecutor”的早期智能体实现。LangGraph是LangChain生态中专门用于构建复杂、有状态、多步骤工作流特别是智能体的库。它提供了基于图的编程模型用于定义带有循环、分支、并行的复杂流程。健壮的状态管理通过State和归约器。持久化与记忆通过检查点机制。多Agent编排原生支持。如何选择如果你的应用是简单的“提示词 - LLM - 输出”或“检索 - 生成”这种线性管道用LangChain的Chain就足够了。它更简单、直接。如果你的应用需要智能体具备复杂的决策循环、工具使用、长期记忆或多步骤协作LangGraph是更优、更现代的选择。它用更清晰的代码结构解决了复杂控制流问题。最佳实践是结合使用使用LangChain来创建你的基础组件LLM模型、提示词模板、各种工具如搜索引擎、计算器、数据库查询、文档检索链。使用LangGraph来编排这些组件构建智能体的核心决策和工作流程。在LangGraph的Node函数内部你去调用LangChain创建好的链或工具。例如上面例子中的answer_node内部使用的llm就是LangChain的ChatOpenAI对象。search_node在实际项目中很可能会调用一个由LangChain的Tool封装好的搜索函数。这种组合让你既能享受LangChain丰富的生态和组件又能利用LangGraph强大的流程控制能力。从趋势来看LangChain团队正将开发重心向LangGraph倾斜未来更复杂的Agent用例都将以LangGraph为推荐实现方式。因此深入学习和掌握LangGraph无疑是开发现代AI智能体应用的必备技能。