📚 课程目录第3章图结构核心概念

3. 图结构核心概念

LangGraph 的核心抽象是将智能体应用建模为一张有状态图（Stateful Graph）。与传统的线性链式调用（如 LangChain 的 SequentialChain）不同，StateGraph 允许开发者显式定义状态如何在节点之间流转、何时需要分支决策、以及如何构造循环来实现“反思—重试—继续”等复杂行为。理解 StateGraph 的构建范式，是掌握多智能体协作、人机交互中断、动态工具调用等高级特性的前提。本章将深入剖析 StateGraph 的四大支柱：状态（State）、节点（Node）、边（Edge）以及条件边（Conditional Edge），并通过一个可运行的案例展示循环结构的实际价值。

上图展示了 StateGraph 的核心运行逻辑：所有节点共享一个全局状态对象，节点通过返回值更新状态，边定义了状态流转的路径，而条件边则根据当前状态动态决定下一步走向。下面我们将逐一拆解这些概念。

3.1 状态（State）：图的“内存”与数据契约

核心原理

在 LangGraph 中，状态是贯穿整个图执行过程的数据容器。它通常是一个 TypedDict 或 Pydantic BaseModel，定义了图中所有节点可读写的字段。状态的设计直接决定了图的扩展性和可维护性。例如，一个客服机器人图的状态可能包含 messages（对话历史）、user_info（用户信息）、order_status（订单状态）等字段。

状态传递遵循局部更新（Partial Update）原则：每个节点只需返回它想要修改的字段子集，LangGraph 会自动将这些返回值合并（Merge）到全局状态中。这种机制避免了节点之间传递庞大的上下文对象，也使得每个节点可以独立测试。

操作流程：定义状态与Reducer

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

# 1. 定义状态结构
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # 使用Reducer处理消息追加
    retry_count: int
    final_answer: str

# 2. 初始化图
graph = StateGraph(AgentState)

这里的关键是 Annotated[list, add_messages]。它告诉 LangGraph：当多个节点返回 messages 字段时，使用 add_messages 函数进行合并（而不是覆盖）。对于普通字段如 retry_count，默认行为是覆盖（Overwrite）。

经验提醒：对于需要累加、追加或去重的字段（如消息列表、日志记录），务必使用 Reducer 函数。否则后一个节点的返回值会覆盖前一个节点的结果，导致数据丢失。

案例拆解：多轮对话中的状态管理

假设我们构建一个研究助手，需要先收集用户需求，再调用搜索工具，最后生成报告。状态设计如下：

字段名	类型	Reducer	用途
messages	list	add_messages	存储所有对话消息
search_query	str	覆盖	当前搜索关键词
search_results	list	覆盖	搜索结果列表
retry_count	int	覆盖	记录重试次数

在节点 search_tool 中，我们读取 state["search_query"]，调用外部 API，然后返回 {"search_results": results}。由于 search_results 使用覆盖策略，每次搜索都会替换旧结果，这符合业务预期。

3.2 节点（Node）与边（Edge）的定义

核心原理

节点是图执行的最小工作单元，本质上是一个接收状态并返回状态更新的 Python 函数。边则定义了节点之间的连接关系。LangGraph 支持两种边：普通边（Edge）和条件边（Conditional Edge）。普通边表示无条件跳转，即上一个节点执行完毕后，下一个节点必然执行。条件边则根据一个路由函数（Router Function）的返回值，动态选择下一个要执行的节点。

操作流程：添加节点与边

# 定义节点函数
def node_a(state: AgentState):
    return {"messages": [{"role": "assistant", "content": "处理中..."}]}

def node_b(state: AgentState):
    return {"retry_count": state["retry_count"] + 1}

def node_c(state: AgentState):
    return {"final_answer": "完成"}

# 添加节点
graph.add_node("A", node_a)
graph.add_node("B", node_b)
graph.add_node("C", node_c)

# 添加普通边
graph.add_edge("A", "B")
graph.add_edge("B", "C")

# 设置入口和出口
graph.set_entry_point("A")
graph.add_edge("C", END)

在上述代码中，END 是 LangGraph 内置的特殊节点，表示图执行结束。所有分支最终都必须连接到 END，否则图会报错。

风险提示：不要在一个节点函数内部直接修改传入的 state 对象。应该始终返回一个包含更新字段的新字典。直接修改会导致状态不可追踪，且在多线程或异步执行时产生竞态条件。

实战场景：工具调用链

考虑一个代码生成智能体：节点 generate_code 生成代码，节点 execute_code 执行代码并捕获错误，节点 fix_code 根据错误修复代码。这三个节点构成一个循环，直到代码执行成功或达到最大重试次数。边的定义如下：

generate_code → execute_code（普通边）
execute_code → fix_code（条件边：如果执行失败）
fix_code → execute_code（普通边，形成循环）
execute_code → END（条件边：如果执行成功）

3.3 条件边与循环结构：实现动态决策

核心原理

条件边是 StateGraph 区别于普通 DAG（有向无环图）工作流的核心特性。它允许图在执行过程中根据当前状态做出分支决策，甚至形成循环。循环结构使得智能体可以模拟人类的“反思—修正—再尝试”过程，这是实现高质量输出的关键。

操作流程：定义路由函数

def should_retry(state: AgentState):
    if state["retry_count"] < 3 and not state["final_answer"]:
        return "fix_code"   # 返回下一个节点的名称
    else:
        return "end"

# 添加条件边
graph.add_conditional_edges(
    "execute_code",
    should_retry,
    {
        "fix_code": "fix_code",
        "end": END
    }
)

路由函数接收当前状态，返回一个字符串键。第三个参数是一个映射字典，将键映射到具体的节点名称或 END。这种设计将决策逻辑与图结构解耦，便于单元测试。

  关键设计模式：循环结构必须包含“退出条件”。在上面的例子中，退出条件是 retry_count >= 3 或 final_answer 非空。如果缺少退出条件，图将无限循环，最终导致超时或栈溢出。

案例拆解：带质量评估的写作智能体

背景与约束：我们需要一个智能体撰写产品营销文案，要求文案必须包含关键词、字数在 200-300 之间，且语气积极。初始模型输出往往不满足要求。

初始现象：第一次生成的文案可能只有 150 字，且语气中性。

关键输入数据：状态中的 draft（草稿）、critique（评估结果）、revision_count（修改次数）。

分析判断过程：我们设计一个 evaluate 节点，使用一个轻量级 LLM 或规则检查草稿质量，返回 {"critique": "字数不足，缺少关键词"}。然后路由函数判断：如果 revision_count < 2 且 critique 不为空，则进入 revise 节点；否则进入 finalize 节点。

实际操作步骤：

def route_after_evaluate(state):
    if state["revision_count"] < 2 and state["critique"]:
        return "revise"
    return "finalize"

graph.add_conditional_edges("evaluate", route_after_evaluate, {
    "revise": "revise",
    "finalize": "finalize"
})

结果对比：第一次运行，revision_count=0，进入 revise 节点，修改后再次评估。第二次运行，revision_count=1，若仍不合格，继续修改。第三次运行，revision_count=2，强制进入 finalize，保证输出不无限循环。

风险与代价：循环会增加 LLM 调用次数，带来延迟和成本。因此设置最大重试次数是必要的。同时，评估节点的准确性直接影响最终质量，如果评估器过于宽松，循环可能提前退出。

复盘经验：将评估逻辑（规则或模型）与业务逻辑分离，便于后续替换更强大的评估器。同时，在状态中记录每次修改的 critique 历史，有助于调试和追溯。

3.4 常见误区与总结复盘

误区一：状态字段命名冲突

当多个节点返回相同的字段名但语义不同时，容易造成混乱。例如，一个节点返回 {"result": "success"}，另一个节点返回 {"result": "error"}，如果使用覆盖策略，后执行的节点会覆盖先前的结果。解决方案是使用更细粒度的字段名，如 tool_result 和 validation_result。

误区二：忽略 Reducer 的必要性

对于列表类型的字段，如果不使用 add_messages 等 Reducer，每次更新都会覆盖整个列表。这在多智能体协作场景中尤其危险——一个智能体的输出会抹掉另一个智能体的贡献。

误区三：条件边返回了未定义的节点名

路由函数返回的字符串必须存在于映射字典的键中，且对应的值必须是已添加的节点或 END。否则 LangGraph 会抛出 InvalidUpdateError。建议在开发阶段使用 assert 或日志记录路由结果。

总结复盘

概念	核心要点	常见错误
状态	TypedDict 定义，支持 Reducer 合并	直接修改传入的 state 对象
节点	纯函数，输入 state，输出更新字典	节点内执行耗时操作且无超时控制
普通边	无条件跳转，适合线性流程	忘记连接 END 节点
条件边	路由函数 + 映射字典，支持循环	缺少退出条件导致死循环

掌握 StateGraph 的构建，意味着你能够将复杂的智能体逻辑拆解为可复用、可测试的图组件。在实际项目中，建议先用纸笔画出状态流转图，明确哪些字段需要 Reducer，哪些节点可能形成循环，然后再编写代码。这种“先设计，后编码”的方式能显著减少调试时间。