1. 从“图”说起LangGraph 的核心心智模型如果你之前接触过 LangChain可能会习惯性地将 LangGraph 视为一个“更高级的 Agent 框架”。这个理解没错但不够本质。LangGraph 真正的核心是它名字里的“Graph”——图。在计算机科学里图是由节点Node和边Edge组成的数据结构用来表示事物之间的复杂关系。LangGraph 正是将 AI 应用的工作流建模成了一个有向图。为什么是图因为现实中的 AI 应用尤其是多步骤、带状态、有分支判断的复杂任务其执行路径很少是一条直线。它更像是一个决策树或者一个状态机根据上一步的结果决定下一步该调用哪个工具或者是否要循环回去修正。用传统的线性脚本去硬编码这些逻辑代码会迅速变得臃肿且难以维护。而图模型天然适合描述这种“如果…就…”的流转关系。在 LangGraph 中每个节点代表一个可执行的函数比如调用大模型、执行工具、处理数据每条边定义了从一个节点到另一个节点的流转条件。你通过定义节点和边就构建了一个完整的、可视化的业务流程。这带来的最大好处是可观测性和可控性。你不仅能清晰地看到整个应用的逻辑全貌还能在任意节点注入检查点、添加日志、甚至动态修改流程这对于调试和生产环境下的运维至关重要。所以在学习 LangGraph 的 API 时请始终带着“我在构建一个图”的心智模型。我们不是在写顺序执行的脚本而是在绘制一张智能的工作流蓝图。本篇我们将深入 LangGraph 最核心的构建块StateGraph、节点Node、边Edge以及条件边Conditional Edge并通过一个比“Hello World”更实用的例子——一个具备自我反思和修正能力的写作助手——来彻底掌握它们。2. 构建图的基石深入理解 StateGraphStateGraph是 LangGraph 中用于构建图的容器类。你可以把它想象成一张画布我们在这张画布上添加节点和边。它的核心作用是管理整个图的状态流转。2.1 StateGraph 的初始化与状态模式创建一个图的第一步是定义“状态”State。状态是一个类似字典的结构它会在图的各个节点之间传递和更新。LangGraph 推荐使用TypedDict来定义状态的模式这能带来极佳的代码提示和类型安全。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END # 1. 定义状态模式 class WriterState(TypedDict): # 用户输入的原始指令 instruction: str # 当前生成的草稿 draft: str # 收集到的反馈或反思意见 feedback: List[str] # 一个标志位用于控制流程走向 needs_revision: bool这里我们定义了一个WriterState它包含四个字段。Annotated是 Python 的类型提示扩展LangGraph 用它来声明某些字段的聚合方式。例如Annotated[List[str], operator.add]表示feedback字段是一个字符串列表当多个节点都向这个字段写入数据时LangGraph 会自动使用operator.add即列表的操作来合并它们而不是覆盖。这对于收集日志、历史消息等场景非常有用。定义了状态模式后我们就可以创建StateGraph了# 2. 创建图构建器并传入状态模式 graph_builder StateGraph(WriterState)这个graph_builder对象就是我们操作画布的工具。接下来所有添加节点、定义边的操作都基于它进行。注意StateGraph本身并不执行任何逻辑它只是一个“蓝图”或“构建器”。真正的图是在调用graph_builder.compile()之后才生成的。这种设计模式建造者模式使得图的构建过程非常清晰和灵活。2.2 状态流转的底层机制理解状态如何在图中流转是关键。每个节点函数接收当前整个状态一个符合WriterState模式的字典作为输入并返回一个更新后的字典。这个返回的字典中只有发生变化的字段需要被包含。LangGraph 的运行时引擎会智能地将这个“增量更新”合并到全局状态中。例如一个节点只修改了draft字段它只需返回{“draft”: “新内容”}。引擎会确保其他字段如instruction,feedback保持不变。这种基于增量的更新机制使得节点可以专注于自己的职责无需关心状态的完整结构大大降低了耦合度。3. 节点的定义与最佳实践节点是图上执行实际工作的单元。在 LangGraph 中任何接收状态并返回状态更新字典的函数都可以成为一个节点。3.1 如何编写一个健壮的节点函数让我们为写作助手创建第一个节点generate_draft负责根据用户指令生成初稿。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate # 初始化大模型 llm ChatOpenAI(model“gpt-4-turbo-preview”) # 定义生成草稿的节点函数 def generate_draft(state: WriterState) - dict: 根据用户指令生成文章初稿。 # 1. 从状态中提取输入 user_instruction state[“instruction”] # 2. 构建提示词 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一位专业的写作助手。请根据用户的要求撰写一篇结构清晰、语言流畅的文章草稿。”), (“human”, “用户要求{instruction}\n\n请开始撰写”) ]) # 3. 调用大模型 chain prompt | llm response chain.invoke({“instruction”: user_instruction}) # 4. 处理响应更新状态 # 注意我们只返回需要更新的字段 return { “draft”: response.content, # 初始化 feedback 列表方便后续节点添加 “feedback”: [] }这个函数展示了节点编写的几个最佳实践明确的输入输出函数签名清晰表明它接收WriterState返回dict。返回的字典就是状态的增量更新。单一职责这个节点只做一件事——生成草稿。它不负责检查质量也不负责修改。从状态中提取所有输入都来自state参数这保证了节点的可复用性它不依赖外部变量。返回增量更新我们只返回了draft和feedback字段。即使instruction字段也存在我们也不需要返回它因为引擎知道我们没有修改它。3.2 将函数注册为图节点定义了函数之后需要将其“注册”到图构建器中赋予它一个在图中唯一的名称。# 将函数注册为节点节点名为 “generate_draft” graph_builder.add_node(“generate_draft”, generate_draft)add_node方法做了两件事一是将函数generate_draft与名称绑定二是根据函数的类型提示在内部记录下这个节点会读写哪些状态字段用于后续的优化和验证。你可以添加任意多个节点。例如我们再添加一个用于反思和评估草稿质量的节点def reflect_on_draft(state: WriterState) - dict: 对当前草稿进行反思提出修改意见。 draft state[“draft”] prompt ChatPromptTemplate.from_messages([ (“system”, “你是一位严格的编辑。请仔细审阅以下文章草稿从逻辑、结构、语言、事实准确性等方面提出具体、可操作的修改意见。如果草稿质量已非常高无需修改请明确指出。”), (“human”, “请审阅以下草稿\n\n{draft}”) ]) chain prompt | llm response chain.invoke({“draft”: draft}) # 将反思意见添加到 feedback 列表中 new_feedback [response.content] # 同时判断是否需要修订。这里用一个简单的启发式规则如果反馈意见超过一定长度且不包含“无需修改”等关键词则认为需要修订。 needs_revision len(response.content) 50 and “无需修改” not in response.content return { “feedback”: new_feedback, # 由于 feedback 字段定义为 add这里会追加到列表 “needs_revision”: needs_revision } # 注册反思节点 graph_builder.add_node(“reflect”, reflect_on_draft)注意reflect_on_draft节点的返回值。它更新了feedback列表追加了一条新反馈和needs_revision标志位。这个标志位将至关重要地决定流程的走向。4. 边的艺术连接节点与控制流程节点是孤立的边将它们连接起来形成了工作流。LangGraph 中有几种类型的边它们共同决定了状态在图中如何移动。4.1 普通边Edge顺序执行最简单的边是普通边它无条件地将一个节点的输出导向另一个节点。这用于定义固定的、必须执行的步骤序列。# 设置图的入口点从哪个节点开始执行 graph_builder.set_entry_point(“generate_draft”) # 添加一条从 “generate_draft” 到 “reflect” 的边 graph_builder.add_edge(“generate_draft”, “reflect”)现在图的流程是generate_draft-reflect。执行完generate_draft后状态会自动传递给reflect节点。4.2 条件边Conditional Edge实现分支逻辑这是 LangGraph 最强大特性之一。条件边允许根据当前状态的某个值动态决定下一个要执行的节点。它实现了if-else或switch-case的分支逻辑。在我们的写作助手例子中在reflect节点之后流程需要分支如果needs_revision为True则进入revise_draft修订草稿节点。如果needs_revision为False则说明草稿合格流程可以结束。首先我们需要创建修订节点def revise_draft(state: WriterState) - dict: 根据反馈意见修订草稿。 draft state[“draft”] # 获取所有的反馈意见是一个列表 all_feedback state[“feedback”] # 将列表合并成一段文本 combined_feedback “\n”.join(all_feedback) prompt ChatPromptTemplate.from_messages([ (“system”, “你是一位写作助手。请根据编辑的反馈意见认真修改以下文章草稿。确保修改后的内容完全回应了每一条反馈。”), (“human”, “原始草稿\n{draft}\n\n编辑反馈\n{feedback}\n\n请输出修改后的完整草稿”) ]) chain prompt | llm response chain.invoke({“draft”: draft, “feedback”: combined_feedback}) # 更新草稿并将 needs_revision 重置为 False准备进入下一轮评估 return { “draft”: response.content, “needs_revision”: False } # 注册修订节点 graph_builder.add_node(“revise”, revise_draft)接下来定义条件边。条件边需要一个“路由函数”Router Function。这个函数接收当前状态并返回下一个要执行的节点名称字符串或者返回特殊的END标记表示终止。def decide_after_reflection(state: WriterState) - str: 根据是否需要修订决定下一步是修订还是结束。 if state.get(“needs_revision”, False): # 需要修订前往 “revise” 节点 return “revise” else: # 无需修订工作流结束 return END现在我们将这条条件边添加到图中。注意条件边是从一个节点出发指向多个可能的目标。我们使用add_conditional_edges方法。# 从 “reflect” 节点出发添加条件边。 # 第一个参数是源节点 “reflect”。 # 第二个参数是路由函数 decide_after_reflection。 # 第三个参数是一个映射字典可选但推荐将路由函数可能返回的字符串映射到人类可读的描述主要用于可视化。 graph_builder.add_conditional_edges( “reflect”, decide_after_reflection, { “revise”: “需要修订草稿”, END: “草稿合格流程结束” } )4.3 闭环的形成添加循环边目前的图是generate_draft-reflect- (条件判断) -revise或END。 如果走到了revise节点修订完成后呢我们当然希望修订后的草稿能再次被reflect节点评估形成“生成-评估-修订”的循环直到质量达标。这就需要从revise节点添加一条边回到reflect节点。# 添加一条从 “revise” 回到 “reflect” 的普通边形成循环 graph_builder.add_edge(“revise”, “reflect”)现在图的逻辑完整了从generate_draft开始生成初稿。进入reflect节点进行评估获得反馈并设置needs_revision标志。如果needs_revision为True进入revise节点修订草稿修订后将needs_revision重置为False然后返回第2步reflect进行再次评估。如果needs_revision为False流程走向END工作流成功终止。这个循环可能执行多次直到reflect节点认为草稿无需再改。这实现了一个简单的自我迭代优化过程。5. 编译与运行让图活起来蓝图绘制完毕我们需要将其编译成一个可执行的对象。# 编译图得到可执行的“运行时图” graph graph_builder.compile()compile()方法会进行一系列检查如是否存在环、入口点是否设置等并生成一个优化后的图对象。这个graph对象有两个最常用的方法invoke和stream。5.1 使用invoke同步执行invoke方法接收一个初始状态字典运行整个图直到遇到END节点然后返回最终状态。# 定义初始状态 initial_state { “instruction”: “写一篇关于 LangGraph 条件边使用技巧的简短技术博客约500字。”, “draft”: “”, # 初始草稿为空 “feedback”: [], # 初始反馈为空列表 “needs_revision”: False # 初始不需要修订 } # 执行图 final_state graph.invoke(initial_state) print(“最终草稿”) print(final_state[“draft”]) print(“\n收集到的所有反馈”) for i, fb in enumerate(final_state[“feedback”]): print(f“反馈轮次 {i1}: {fb[:100]}...”) # 打印前100字符invoke是“一镜到底”的执行方式适合快速测试和简单的同步调用。5.2 使用stream进行流式执行与调试stream方法更为强大它返回一个生成器可以逐节点、甚至逐步骤地产出状态快照。这对于调试和构建交互式前端至关重要。# 流式执行观察每一步的状态变化 for step in graph.stream(initial_state): # step 是一个元组 (node_name, state_update) node_name, state_update list(step.items())[0] # 解包 print(f“\n 节点 ‘{node_name}’ 执行完毕 ) print(f“状态更新: {state_update}”) # 你可以在这里插入逻辑例如将草稿实时显示给用户通过stream你可以清晰地看到执行了哪个节点node_name。该节点对状态做了哪些修改state_update。整个工作流是如何一步步推进的。这是理解复杂图运行逻辑、定位问题节点的最有效工具。例如你可能会发现reflect节点反复将needs_revision设为True导致死循环这时你就需要去检查评估逻辑或修订逻辑是否有问题。6. 实战中的陷阱与进阶技巧掌握了基础 API我们来看看实际项目中容易踩的坑和一些进阶用法。6.1 状态字段的聚合策略冲突这是新手最常见的错误之一。回顾我们的状态定义feedback字段使用了Annotated[List[str], operator.add]。这意味着所有节点对feedback的更新都会以追加列表合并的方式进行。设想一个场景如果revise节点错误地返回了{“feedback”: [“修订完成”]}会发生什么feedback列表会不断增长包含大量重复的“修订完成”信息。这显然不是我们想要的。正确做法在revise节点中我们不应该更新feedback字段。我们的更新只针对draft和needs_revision。LangGraph 的引擎很聪明如果节点返回的字典中不包含某个字段该字段就会保持不变。经验法则仔细规划每个节点的职责和它应该更新的字段。对于用于收集历史信息的字段如feedback,conversation_history使用add聚合。对于代表当前核心状态的字段如draft,answer通常使用覆盖式更新。6.2 条件边路由函数的复杂性管理我们的decide_after_reflection函数很简单只判断一个布尔值。但在真实场景中路由逻辑可能非常复杂比如基于大模型的分析结果来决定下一步。def complex_router(state: State) - str: # 可能调用另一个LLM来分析状态决定下一步 analysis call_llm_to_analyze(state) if analysis “option_a”: return “node_a” elif analysis “option_b”: return “node_b” else: return “node_default”这里有一个性能陷阱这个路由函数本身可能很耗时因为调用了LLM。如果它位于一个循环中每次循环都会执行造成大量不必要的开销。优化方案将路由决策所需的信息在之前的节点中计算好并存入状态。让路由函数变成一个简单的、无副作用的判断器只读取状态中的标志位做决定。这符合“将决策逻辑与计算逻辑分离”的设计原则。6.3 图的可视化不可或缺的调试工具LangGraph 内置了可视化功能能将你定义的图生成一张 Mermaid 流程图。这对于沟通、设计和调试有巨大帮助。# 将图导出为 Mermaid 格式的字符串 mermaid_schema graph.get_graph().draw_mermaid() print(mermaid_schema) # 你可以将输出的字符串复制到支持 Mermaid 的编辑器如 Typora, Notion, GitHub Markdown中查看流程图。通过可视化你可以一眼看出节点之间的连接关系是否正确。条件边的分支是否覆盖了所有情况。是否存在意外的循环或无法到达的节点。我强烈建议在开发任何 LangGraph 应用时都将可视化作为第一步和常规检查步骤。6.4 中断与持久化生产级应用的关键我们上面的例子都在内存中一次性运行完毕。但在生产环境中一个工作流可能被用户中断比如关闭网页或者需要运行很长时间等待外部API。这就需要状态持久化和从断点恢复的能力。LangGraph 通过Checkpointer抽象来支持这一功能。简单来说你可以在图中配置检查点引擎会在每个检查点将完整状态保存到数据库如Redis、PostgreSQL。当需要恢复时只需提供检查点ID就能从上次中断的节点继续执行。from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph # 创建一个支持检查点的图构建器 graph_builder StateGraph(WriterState).add_node(...) # 添加节点... graph_builder.add_edge(...) # 使用 SQLite 作为检查点存储器生产环境可用其他后端 checkpointer SqliteSaver.from_conn_string(“:memory:”) # 内存数据库示例用 # 编译时传入 checkpointer graph graph_builder.compile(checkpointercheckpointer) # 执行时会返回一个配置ID用于后续恢复 config {“configurable”: {“thread_id”: “user_123_session_1”}} initial_state {…} result graph.invoke(initial_state, configconfig) # 假设在某个时刻中断了... # 稍后恢复执行只需要相同的 config recovered_result graph.invoke(None, configconfig) # 初始状态为 None因为会从检查点加载这是构建可靠、长周期 AI 应用的基础也是 LangGraph 相较于简单脚本的核心优势之一。7. 完整代码示例与总结让我们把上面的所有部分整合成一个完整的、可运行的写作助手示例。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate # —- 1. 定义状态 —- class WriterState(TypedDict): instruction: str draft: str feedback: Annotated[List[str], operator.add] needs_revision: bool # —- 2. 初始化组件 —- llm ChatOpenAI(model“gpt-4-turbo-preview”) # —- 3. 定义节点函数 —- def generate_draft(state: WriterState) - dict: prompt ChatPromptTemplate.from_messages([ (“system”, “你是一位专业的写作助手。请根据用户的要求撰写一篇结构清晰、语言流畅的文章草稿。”), (“human”, “用户要求{instruction}”) ]) chain prompt | llm response chain.invoke({“instruction”: state[“instruction”]}) return {“draft”: response.content, “feedback”: []} def reflect_on_draft(state: WriterState) - dict: prompt ChatPromptTemplate.from_messages([ (“system”, “你是一位严格的编辑。请审阅以下文章草稿提出具体、可操作的修改意见。如果草稿质量已非常高无需修改请明确指出。”), (“human”, “草稿{draft}”) ]) chain prompt | llm response chain.invoke({“draft”: state[“draft”]}) needs_revision len(response.content) 50 and “无需修改” not in response.content return {“feedback”: [response.content], “needs_revision”: needs_revision} def revise_draft(state: WriterState) - dict: combined_feedback “\n”.join(state[“feedback”]) prompt ChatPromptTemplate.from_messages([ (“system”, “请根据编辑反馈修改文章草稿。确保修改后的内容完全回应了每一条反馈。”), (“human”, “原始草稿\n{draft}\n\n编辑反馈\n{feedback}”) ]) chain prompt | llm response chain.invoke({“draft”: state[“draft”], “feedback”: combined_feedback}) return {“draft”: response.content, “needs_revision”: False} # —- 4. 构建图 —- graph_builder StateGraph(WriterState) # 添加节点 graph_builder.add_node(“generate_draft”, generate_draft) graph_builder.add_node(“reflect”, reflect_on_draft) graph_builder.add_node(“revise”, revise_draft) # 设置入口和边 graph_builder.set_entry_point(“generate_draft”) graph_builder.add_edge(“generate_draft”, “reflect”) graph_builder.add_edge(“revise”, “reflect”) # 循环边 # 添加条件边 def decide_after_reflection(state: WriterState) - str: return “revise” if state.get(“needs_revision”, False) else END graph_builder.add_conditional_edges( “reflect”, decide_after_reflection, {“revise”: “Revise”, END: “End”} ) # 编译图 graph graph_builder.compile() # —- 5. 运行与测试 —- if __name__ “__main__”: initial_state { “instruction”: “用通俗的语言解释 LangGraph 中的条件边Conditional Edge是什么并举例说明其用途。”, “draft”: “”, “feedback”: [], “needs_revision”: False } print(“开始执行写作助手工作流…\n”) # 使用 stream 来观察过程 for step in graph.stream(initial_state, stream_mode“values”): # stream_mode“values” 只输出状态值 node_name, state list(step.items())[0] print(f“[经过节点 {node_name}]”) print(f“当前草稿长度: {len(state[‘draft’])} 字符”) print(f“是否需要修订: {state[‘needs_revision’]}”) if state[‘feedback’]: print(f“最新反馈: {state[‘feedback’][-1][:80]}…\n”) else: print(“\n”) # 获取最终结果 final_state graph.invoke(initial_state) print(“\n 工作流结束 ) print(f“最终草稿已生成共 {len(final_state[‘draft’])} 字符。”) print(f“共进行了 {len(final_state[‘feedback’])} 轮反思/修订。”)通过这个从零构建的例子你应该对 LangGraph 的基础 API ——StateGraph、节点、边、条件边 —— 有了透彻的理解。它们是你绘制任何复杂 AI 工作流蓝图的画笔。记住设计图的关键在于清晰定义状态、合理划分节点职责、以及精确控制边的流向。下一章我们将探索更高级的图结构如子图、并行执行和人工干预节点来构建真正企业级的智能体应用。