1. 项目概述为什么是LangGraph如果你正在构建一个基于大语言模型的复杂应用比如一个能处理多轮对话、执行多步骤任务、或者需要维护长期记忆的智能体那么你很可能已经感受到了LangChain的局限性。LangChain提供了丰富的工具链但当业务流程变得复杂、状态需要跨步骤持久化、或者决策路径需要根据条件动态调整时仅仅依靠链式调用就显得有些力不从心了。这时你会需要一个更强大的工具来清晰地定义和控制你的应用流程——这就是LangGraph诞生的背景。简单来说LangGraph是LangChain生态中的一个库它专门用于构建有状态的、多智能体Agent的工作流。它把整个应用流程抽象成一个“图”图中的节点代表一个执行单元比如调用一个LLM、执行一个工具、或者运行一段代码边则定义了节点之间的流转逻辑。这种图结构让你能够直观地设计出包含循环、条件分支、并行执行等复杂逻辑的流程并且能轻松地管理和追踪整个流程的“状态”。“HelloLangGraph”这个标题意味着我们将从最基础的概念和最简单的例子入手带你一步步走进LangGraph的世界。这不是一个简单的“Hello World”打印而是理解LangGraph核心三要素——状态State、节点Node和边Edge的绝佳起点。通过这个入门项目你将学会如何搭建一个最基本的LangGraph应用骨架为后续构建更复杂的智能体工作流打下坚实的基础。无论你是LangChain的老用户想升级技能还是直接想用更现代的方式构建LLM应用这篇文章都将为你提供一个清晰的起点。2. LangGraph核心三要素深度解析要玩转LangGraph你必须吃透它的三个核心概念状态、节点和边。它们共同构成了LangGraph工作流的灵魂。2.1 状态工作流的记忆中枢在LangGraph中状态是一个贯穿整个工作流执行周期的共享数据容器。你可以把它想象成一个全局的、可变的“白板”或者“上下文字典”。每个节点在执行时都可以读取和修改这个状态下一个节点看到的就是被上一个节点更新后的状态。这是LangGraph实现有状态工作流的关键。状态通常用一个Pydantic模型来定义这确保了类型安全和清晰的接口。例如在一个对话机器人应用中你的状态模型可能包含以下字段from typing import List, Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages class State(TypedDict): # 对话历史这是一个特殊字段LangGraph提供了add_messages操作来方便地追加消息 messages: Annotated[List[dict], add_messages] # 用户当前查询 user_query: str # 从查询中提取的关键信息 extracted_info: dict # 需要调用的工具名称 next_tool: str # 最终给用户的回复 final_answer: str这里有几个关键点需要注意。第一TypedDict是定义状态结构的推荐方式它比普通的字典更清晰。第二Annotated类型提示与add_messages的结合是LangGraph的一个精妙设计。add_messages是一个“归约器”它定义了当多个节点试图修改messages列表时如何合并这些修改这里是追加而不是覆盖。这完美契合了对话场景中消息不断累积的需求。注意状态的设计至关重要。一个好的状态模型应该只包含工作流真正需要共享和传递的数据。避免将临时变量或节点内部计算的大量中间结果塞进状态这会让状态变得臃肿且难以理解。通常状态只保存输入、输出、关键的决策标志以及需要持久化的上下文信息。2.2 节点具体的执行单元节点是工作流中实际“干活”的部分。每个节点都是一个函数它接收当前的状态作为输入执行一些操作如调用LLM、查询数据库、运行计算然后返回一个对状态的更新。在LangGraph中定义一个节点非常简单它就是一个接收状态并返回状态更新的函数。返回的更新通常是一个字典其中的键对应状态模型中的字段值则是要更新或设置的新值。LangGraph会自动将这个更新应用到全局状态上。def llm_node(state: State) - dict: 一个调用LLM生成回复的节点 # 1. 从状态中获取所需信息 conversation_history state[“messages”] query state[“user_query”] # 2. 构建LLM提示词 prompt f“基于以下对话历史和当前问题请生成回复。\n历史{conversation_history[-5:]}\n问题{query}” # 3. 调用LLM这里用伪代码示意 # 实际中你会使用LangChain的LLM接口如ChatOpenAI llm_response call_llm(prompt) # 4. 返回状态更新 # 更新消息列表添加助手的回复 new_message {“role”: “assistant”, “content”: llm_response} return {“messages”: [new_message], “final_answer”: llm_response}节点的设计原则是“单一职责”。一个节点最好只完成一件明确的事情。例如一个节点专门负责调用LLM另一个节点专门负责解析用户意图再一个节点专门负责调用某个API。这样的设计使得工作流易于理解、调试和测试。2.3 边控制流转的逻辑边决定了工作流的走向。它定义了在当前节点执行完毕后接下来应该执行哪个或哪些节点。边可以分为两种主要类型条件边和普通边。普通边是固定的连接表示无条件地从一个节点跳转到下一个节点。在代码中你通过add_edge方法来建立这种连接。条件边则提供了分支逻辑。它根据当前状态的某些属性动态决定下一步该走哪条路。这是实现循环、判断和复杂工作流的核心。条件边通过add_conditional_edges方法添加并关联一个“路由函数”。def route_after_tool_call(state: State) - str: 根据工具调用结果决定下一步 # 假设状态中有一个字段表示工具执行是否成功 if state.get(“tool_success”, False): # 成功则前往“处理结果”节点 return “process_result” else: # 失败则前往“处理错误”节点 return “handle_error”在这个例子中路由函数检查state[“tool_success”]的值并返回下一个目标节点的名称。LangGraph会根据这个返回值将工作流转发到对应的节点。边和节点共同构成了工作流的“骨架图”。通过巧妙地组合条件边和普通边你可以构建出任意复杂的流程比如“循环直到满足某个条件”、“尝试方案A失败后回退到方案B”、“并行执行多个任务后汇总结果”等等。这正是LangGraph比简单链式调用强大得多的地方。3. 从零构建你的第一个LangGraph应用理论说得再多不如亲手搭建一个。下面我们将一步步构建一个最简单的LangGraph应用一个能记住对话历史的回声机器人。它会把用户说的话原样返回并维护整个对话历史。这个例子虽小但完整涵盖了定义状态、创建节点、连接边和运行工作流的全过程。3.1 环境准备与依赖安装首先确保你的Python环境在3.8以上。然后安装必要的库。除了LangGraph我们通常还需要LangChain的核心包以及一个LLM的接口比如OpenAI。pip install langgraph langchain-openai如果你使用其他LLM提供商如Anthropic、Google等请安装对应的LangChain集成包。为了示例简单我们暂时不真正调用LLM而是用模拟函数代替这样你可以无需API密钥直接运行。3.2 定义工作流状态我们回顾一下之前提到的状态设计。对于这个回声机器人我们需要记录完整的对话历史。我们将使用LangGraph推荐的TypedDict和消息归约器模式。from typing import List, Annotated, TypedDict from langgraph.graph.message import add_messages # 定义状态结构 class EchoBotState(TypedDict): # 核心对话消息列表。Annotated注解指定了修改此字段时的归约操作。 messages: Annotated[List[dict], add_messages] # 我们可以额外加一个字段记录最近一次用户输入方便节点使用 latest_input: stradd_messages这个归约器是关键。它确保了无论哪个节点返回了对messages字段的更新通常是一个包含新消息的列表这些新消息都会被追加到现有的消息列表末尾而不是替换掉整个列表。这完美符合对话场景的需求。3.3 创建核心功能节点我们的工作流只需要一个核心节点echo_node。它负责从状态中取出最新的用户输入生成一个回声回复并更新状态。def echo_node(state: EchoBotState) - dict: 回声节点将用户的最新输入原样返回。 # 1. 获取最新的用户消息。 # 状态中的messages列表包含了所有历史消息。 # 我们假设最后一条消息是用户刚发送的。 message_list state[“messages”] if not message_list: # 如果没有消息则返回一个默认回复 return {“messages”: [{“role”: “assistant”, “content”: “你好请说点什么。”}]} last_message message_list[-1] # 确保最后一条消息来自用户 if last_message[“role”] ! “user”: return {“messages”: [{“role”: “assistant”, “content”: “我在等待你的输入。”}]} user_input last_message[“content”] # 2. 生成“回声”回复。在实际复杂应用中这里可能是调用LLM、工具等。 assistant_response f“我听到你说{user_input}” # 3. 构造状态更新。 # 我们要做两件事 # a) 将助手的回复追加到messages列表。归约器add_messages会处理这个追加操作。 # b) 更新latest_input字段可选用于演示。 update { “messages”: [{“role”: “assistant”, “content”: assistant_response}], “latest_input”: user_input } return update这个节点函数做了几件典型的事情从状态中读取数据、执行业务逻辑、返回一个包含更新字段的字典。返回的字典中的键必须与状态模型EchoBotState中定义的字段名相匹配。3.4 组装图并设置入口与出口有了状态和节点现在我们需要用StateGraph这个类把它们组装起来并指定工作流的开始和结束。from langgraph.graph import StateGraph # 1. 创建一个以EchoBotState为状态类型的工作流构建器 workflow_builder StateGraph(EchoBotState) # 2. 将我们定义的节点添加到图中并给它起个名字比如“echo” workflow_builder.add_node(“echo”, echo_node) # 3. 设置入口点。指定工作流开始时第一个执行的节点。 workflow_builder.set_entry_point(“echo”) # 4. 设置出口点。指定工作流结束后最后一个执行的节点。 # 在这个简单例子中echo节点执行完就直接结束了。 workflow_builder.set_finish_point(“echo”) # 5. 编译图得到一个可执行的工作流对象 echo_graph workflow_builder.compile()compile()方法非常关键它会检查图的结构比如所有被边引用的节点是否都已添加是否存在无法到达的节点等并返回一个CompiledGraph对象。这个对象就是我们可以直接调用的“应用程序”。3.5 运行与测试工作流现在让我们运行这个工作流模拟几轮对话。# 导入用于可视化图的函数可选但非常有用 from langgraph.graph import draw_mermaid import IPython.display as display # 生成Mermaid图代码并显示在Jupyter Notebook中 # graph_image draw_mermaid(echo_graph) # display.Image(graph_image) # 初始化工作流的输入状态。注意必须符合EchoBotState的结构。 initial_state { “messages”: [{“role”: “user”, “content”: “你好世界”}], “latest_input”: “” } # 运行工作流。使用.stream()方法进行流式执行它返回一个迭代器。 # 对于简单图我们也可以直接用.invoke()获取最终结果。 for step in echo_graph.stream(initial_state): # stream()会产出每个节点执行后的状态快照 node_name list(step.keys())[0] # 这里只有一个节点‘echo’ state_snapshot step[node_name] print(f“节点 ‘{node_name}’ 执行后的状态:”) print(f“ 最新消息: {state_snapshot[‘messages’][-1]}”) print(f“ 最新输入: {state_snapshot[‘latest_input’]}”) print(“-” * 30) # 如果要直接获取最终状态使用.invoke() final_state echo_graph.invoke(initial_state) print(“\n最终对话历史:”) for msg in final_state[“messages”]: print(f“ {msg[‘role’]}: {msg[‘content’]}”)运行上述代码你会看到工作流执行了一次echo节点被调用状态被更新。messages列表里现在有两条记录用户的“你好世界”和助手的“我听到你说你好世界”。latest_input字段也被更新了。你可以修改initial_state中的messages模拟多轮对话。因为状态中的messages会一直累积所以每次调用invoke时它都会基于完整的对话历史进行回应。这就是有状态工作流的威力——它记住了之前发生的一切。4. 进阶引入条件边与循环一个只会回声的机器人显然没什么用。让我们把它升级一下加入简单的意图识别和循环让它更像一个能处理多轮任务的智能体。新功能如果用户说“结束”则工作流终止并道别否则持续进行回声对话。4.1 设计新的状态与节点状态可以沿用EchoBotState。我们需要新增两个节点classify_intent_node: 判断用户意图是普通对话还是要求结束。goodbye_node: 专门处理结束对话的逻辑。同时我们需要修改echo_node让它只在普通对话时被调用。def classify_intent_node(state: EchoBotState) - dict: 意图分类节点。这是一个非常简单的基于规则的分发器。 message_list state[“messages”] if not message_list: return {“intent”: “unknown”} last_message message_list[-1] if last_message[“role”] ! “user”: return {“intent”: “unknown”} user_input last_message[“content”].strip().lower() # 简单的规则如果用户输入包含“结束”、“再见”、“退出”等词则认为是结束意图 end_keywords [“结束”, “再见”, “退出”, “bye”, “goodbye”] if any(keyword in user_input for keyword in end_keywords): return {“intent”: “end_conversation”} else: return {“intent”: “continue_chat”} def goodbye_node(state: EchoBotState) - dict: 结束对话节点。 farewell_msg “感谢你的交流期待下次再见” return { “messages”: [{“role”: “assistant”, “content”: farewell_msg}], “latest_input”: state.get(“latest_input”, “”) } # 修改echo_node使其更专注于生成回复 def echo_node_v2(state: EchoBotState) - dict: message_list state[“messages”] last_message message_list[-1] user_input last_message[“content”] assistant_response f“我听到你说{user_input}。我们还在聊天中。” return { “messages”: [{“role”: “assistant”, “content”: assistant_response}], “latest_input”: user_input }4.2 构建带条件分支的图现在我们来构建一个更复杂的图入口点是classify_intent_node根据它的输出决定是走向echo_node_v2继续聊天还是走向goodbye_node结束对话。并且在echo_node_v2执行后我们希望工作流能循环回到classify_intent_node以等待和处理用户的下一条消息。这就形成了一个“循环”直到用户触发结束条件。from langgraph.graph import StateGraph, END # 创建新的图构建器 workflow_builder_v2 StateGraph(EchoBotState) # 添加三个节点 workflow_builder_v2.add_node(“classify_intent”, classify_intent_node) workflow_builder_v2.add_node(“echo”, echo_node_v2) workflow_builder_v2.add_node(“goodbye”, goodbye_node) # 设置入口点 workflow_builder_v2.set_entry_point(“classify_intent”) # 添加条件边从‘classify_intent’节点出发根据其输出决定去向 workflow_builder_v2.add_conditional_edges( “classify_intent”, # 源节点 # 路由函数读取classify_intent节点输出中的“intent”字段 lambda state: state.get(“intent”, “unknown”), # 映射将路由函数返回的值映射到下一个目标节点 { “continue_chat”: “echo”, # 意图为继续聊天去echo节点 “end_conversation”: “goodbye”, # 意图为结束去goodbye节点 “unknown”: “echo” # 未知意图默认去echo节点安全处理 } ) # 添加普通边从‘echo’节点执行完后无条件地回到‘classify_intent’节点形成循环 workflow_builder_v2.add_edge(“echo”, “classify_intent”) # 添加普通边从‘goodbye’节点执行完后指向特殊的‘END’表示工作流终止 workflow_builder_v2.add_edge(“goodbye”, END) # 编译图 conversation_graph workflow_builder_v2.compile()这段代码是LangGraph能力的集中体现add_conditional_edges创建了动态分支。classify_intent_node的输出state[“intent”]决定了工作流的走向。add_edge(“echo”, “classify_intent”)创建了一个循环。这使得系统可以持续处理用户的多轮输入。END是LangGraph内置的一个特殊节点代表工作流的终点。4.3 运行与调试循环工作流现在运行这个升级版的工作流。我们需要以“流式”的方式调用它并持续提供新的用户输入。在实际应用中这通常由一个外部循环如Web服务器接口来驱动。这里我们模拟一个简单的对话序列。# 初始化状态只有一条用户消息 current_state { “messages”: [{“role”: “user”, “content”: “你好”}], “latest_input”: “” } print(“开始对话输入‘结束’以退出...”) print(f“用户: {current_state[‘messages’][-1][‘content’]}”) # 我们手动模拟几轮交互 max_turns 10 # 防止无限循环设置最大轮数 for turn in range(max_turns): # 调用工作流传入当前状态 # 注意.invoke()会执行直到遇到END。对于循环图我们需要用.stream()并控制输入。 # 更常见的模式是每次用户新输入后我们将新消息追加到state[‘messages’]然后调用一次图。 # 图会从classify_intent开始走完一个循环或结束。 output conversation_graph.invoke(current_state) # 获取最新的助手回复 latest_assistant_msg None for msg in output[“messages”]: if msg[“role”] “assistant”: latest_assistant_msg msg if latest_assistant_msg: print(f“助手: {latest_assistant_msg[‘content’]}”) # 检查是否已经结束即goodbye节点已被执行 # 一个简单的判断如果助手最后一条消息是告别语则结束 if latest_assistant_msg and “感谢你的交流” in latest_assistant_msg[“content”]: print(“对话已结束。”) break # 模拟用户下一轮输入在实际中是等待用户真实输入 if turn 0: user_input “今天天气怎么样” elif turn 1: user_input “我们结束吧。” else: break # 防止继续 print(f“用户: {user_input}”) # 更新状态将新的用户输入追加到消息列表作为下一轮图的输入 new_message {“role”: “user”, “content”: user_input} current_state[“messages”].append(new_message) current_state[“latest_input”] user_input print(“\n完整的对话历史:”) for msg in current_state[“messages”]: print(f“ {msg[‘role’]}: {msg[‘content’]}”)运行这段代码你会看到类似以下的输出开始对话输入‘结束’以退出... 用户: 你好 助手: 我听到你说你好。我们还在聊天中。 用户: 今天天气怎么样 助手: 我听到你说今天天气怎么样。我们还在聊天中。 用户: 我们结束吧。 助手: 感谢你的交流期待下次再见 对话已结束。这个工作流成功实现了多轮对话和条件终止。classify_intent_node在每一轮都起作用当它检测到结束意图时将工作流导向goodbye_node最终到达END循环终止。5. 常见问题与实战避坑指南在实际使用LangGraph构建复杂应用时你会遇到一些典型的问题。下面是我在项目中踩过的一些坑和总结的经验。5.1 状态管理中的常见陷阱问题1状态更新冲突或覆盖。当多个节点可能并发修改同一个状态字段时虽然在单线程中顺序执行但设计时要考虑逻辑上的冲突如果没有使用正确的归约器可能会导致更新丢失。例如两个节点都返回{“messages”: [new_msg]}后执行的会覆盖先执行的。解决方案对于列表、集合等需要累积数据的字段务必使用Annotated和相应的归约器如add_messages。对于数值可能需要自定义归约器如求和、求最大值。仔细设计状态结构让每个节点只修改自己“负责”的那部分状态。问题2状态过于臃肿。把大量中间计算结果、临时变量都塞进状态导致状态对象巨大难以调试也影响序列化/反序列化性能。解决方案遵循“最小共享状态”原则。状态只存储需要在节点间传递和需要持久化的数据。节点内部的复杂计算产生的临时数据应尽量保存在节点函数的局部变量中。如果某些中间结果确实需要共享考虑将其设计为明确的、语义清晰的字段。问题3状态类型错误。返回的状态更新字典中的键或值类型与TypedDict定义不匹配导致运行时错误。解决方案充分利用IDE的类型提示和mypy等静态类型检查工具。在返回更新字典时显式地构造符合类型定义的数据结构。对于复杂的数据转换可以写一些小辅助函数来确保类型安全。5.2 图结构设计与调试技巧问题4图陷入无限循环。这是初学者最常见的问题之一。比如你设置了一个从节点A到节点B的边又从节点B回到节点A但没有设置任何退出条件。解决方案明确终止条件在设计中必须至少有一条路径能到达END节点。通常这会由一个专门的条件节点如should_continue来控制。使用interruptsLangGraph支持“中断”机制允许从外部例如根据超时或用户取消强制停止一个运行中的图。对于可能长时间运行的工作流考虑集成这个功能。设置最大步数在开发调试阶段可以在调用.invoke()或遍历.stream()时手动限制循环次数。可视化养成画图的习惯。使用draw_mermaid函数生成工作流的可视化图能帮你一眼看出循环结构和潜在的无限循环路径。问题5条件边路由函数逻辑复杂导致难以维护。路由函数如果包含大量复杂的if-elif-else逻辑会变得难以理解和测试。解决方案将路由逻辑下沉到专用节点可以创建一个router_node它专门负责计算下一个节点名称并将其写入状态例如state[“next_node”]。然后使用一个简单的条件边读取这个字段进行路由。这样路由逻辑就变成了一个可测试的普通节点。使用映射表将条件判断抽象成从状态特征到节点名的映射关系使路由逻辑更声明式。单元测试为路由函数编写独立的单元测试覆盖各种边界情况。5.3 与LangChain及其他组件的集成问题6如何在节点中调用LangChain的链、工具或Agent这是LangGraph最常用的场景。集成非常直接。解决方案在你的节点函数中像平常一样初始化或传入你需要的LangChain组件。例如from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(model“gpt-4”) prompt ChatPromptTemplate.from_template(“你是一个助手。问题{question}”) chain prompt | llm def llm_chain_node(state: State): question state[“user_question”] response chain.invoke({“question”: question}) return {“answer”: response.content}确保你的LLM调用、工具调用等是异步友好的如果工作流使用异步执行。通常建议在节点函数内部处理可能的异常并将错误信息写入状态由后续的错误处理节点来接管。问题7如何实现“长期记忆”或与数据库交互工作流的状态默认是临时的存在于单次执行过程中。要实现跨会话的记忆需要将状态持久化。解决方案检查点LangGraph内置了检查点机制可以将图执行到某个节点时的完整状态保存下来例如存到数据库并赋予一个唯一的线程ID。下次可以用这个线程ID恢复状态继续执行。这是实现多轮对话、长任务暂停/恢复的核心。外部存储在节点函数中直接连接你的数据库向量数据库、SQL数据库等。将需要长期记忆的信息如对话摘要、用户偏好读写到外部存储中。状态本身可以只保存一个指向外部存储的引用ID。结合使用通常的模式是使用检查点保存完整的对话流程状态同时使用外部数据库存储更大量的、结构化的知识或历史记录。5.4 性能与生产环境考量问题8工作流执行慢如何优化复杂的图可能包含多个LLM调用这些通常是性能瓶颈。优化策略并行执行如果多个节点间没有数据依赖可以使用StateGraph的.add_parallel_edges或创建子图来实现并行显著减少总耗时。缓存对于纯函数节点或结果稳定的LLM调用如意图分类考虑引入缓存机制。LangChain有LLM缓存支持可以集成进来。流式输出对于LLM节点使用.stream()方法而不是.invoke()来逐步获取输出可以提升用户体验感知上的速度。精简状态如前所述避免在状态中存储大对象减少序列化开销。问题9如何监控和记录工作流的执行在生产环境中你需要知道工作流每一步发生了什么尤其是出问题时。实战技巧利用stream()事件graph.stream()不仅返回状态还会返回一个包含元数据的流。你可以监听这些事件来记录每个节点的开始、结束、输入、输出。结构化日志在每个节点函数的开始和结束处打日志记录关键信息如节点名、状态摘要、耗时。使用像structlog这样的库方便后续聚合和查询。集成追踪系统LangChain/LangGraph天然支持与LangSmith集成。LangSmith提供了强大的可视化、追踪、调试和测试功能是生产级LLM应用开发的利器。强烈建议在项目早期就接入LangSmith。自定义检查点除了自动保存状态你还可以在关键节点手动将状态快照和元数据保存到监控系统便于事后复盘。从“HelloLangGraph”这个最简单的回声机器人到能够处理条件分支和循环的对话智能体我们走完了LangGraph入门的关键一步。核心在于理解“状态驱动的工作流”这一范式转变——不再是一条链走到底而是一个由状态流转所控制的、灵活可变的图。当你开始用节点和边来思考你的应用逻辑时你会发现许多复杂的业务场景都能被优雅地建模。下一步你可以尝试集成真实的LLM和工具构建能处理查询、搜索、推理的智能体或者利用检查点功能实现跨会话的长期记忆。LangGraph提供的是一套强大而简洁的框架真正的魔法来自于你用它构建的解决实际问题的图。