1. 从“图”说起：LangGraph 的核心心智模型
如果你之前接触过 LangChain，可能会习惯性地将 LangGraph 视为一个“更高级的 Agent 框架”。这个理解没错，但不够本质。LangGraph 真正的核心，是它名字里的“Graph”——图。在计算机科学里，图是由节点（Node）和边（Edge）组成的数据结构，用来表示事物之间的复杂关系。LangGraph 正是将 AI 应用的工作流，建模成了一个有向图。
为什么是图？因为现实中的 AI 应用，尤其是多步骤、带状态、有分支判断的复杂任务，其执行路径很少是一条直线。它更像是一个决策树，或者一个状态机：根据上一步的结果，决定下一步该调用哪个工具，或者是否要循环回去修正。用传统的线性脚本去硬编码这些逻辑，代码会迅速变得臃肿且难以维护。而图模型天然适合描述这种“如果…就…”的流转关系。
在 LangGraph 中，每个节点代表一个可执行的函数（比如调用大模型、执行工具、处理数据），每条边定义了从一个节点到另一个节点的流转条件。你通过定义节点和边，就构建了一个完整的、可视化的业务流程。这带来的最大好处是可观测性和可控性。你不仅能清晰地看到整个应用的逻辑全貌，还能在任意节点注入检查点、添加日志、甚至动态修改流程，这对于调试和生产环境下的运维至关重要。
所以，在学习 LangGraph 的 API 时，请始终带着“我在构建一个图”的心智模型。我们不是在写顺序执行的脚本，而是在绘制一张智能的工作流蓝图。本篇我们将深入 LangGraph 最核心的构建块：StateGraph、节点（Node）、边（Edge）以及条件边（Conditional Edge），并通过一个比“Hello World”更实用的例子——一个具备自我反思和修正能力的写作助手——来彻底掌握它们。
2. 构建图的基石：深入理解 StateGraph
StateGraph是 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“反馈轮次 {i+1}: {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(checkpointer=checkpointer) # 执行时，会返回一个配置ID，用于后续恢复 config = {“configurable”: {“thread_id”: “user_123_session_1”}} initial_state = {…} result = graph.invoke(initial_state, config=config) # 假设在某个时刻中断了... # 稍后恢复执行，只需要相同的 config recovered_result = graph.invoke(None, config=config) # 初始状态为 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 工作流蓝图的画笔。记住，设计图的关键在于清晰定义状态、合理划分节点职责、以及精确控制边的流向。下一章，我们将探索更高级的图结构，如子图、并行执行和人工干预节点，来构建真正企业级的智能体应用。
