六种LangGraph基本元素及其各自隐藏的故障模式

最初的 LangGraph 智能体通常很快就能完成工作：它回答一个问题，优化答案后便停止运行。随后有人添加了重试分支，结果图结构不再终止，不断消耗 API 信用额度，直到进程被强制终止。问题往往只是缺少一条边，但真正的症结在于缺乏对图结构为何如此运行的内在机制理解。

本指南将基于LangGraph所构成的六种基本要素来构建该模型：状态、节点、直接边、条件边、检查点机制以及人工干预机制。针对每一项要素，您都会看到一个最简示例、团队在使用时最常犯的错误，以及适合投入使用的版本。如果您想更全面地了解基于这些要素构建的智能体模式，《LangGraph实践：状态、节点、边及五种智能体模式》一文涵盖了相关内容；而本文的重点则是故障模式。

为何选择图结构而非链结构

LangChain的管道语法prompt | llm | parser非常适合对模型进行单次处理。一旦智能体需要做出决策——是搜索还是直接回答、重试还是放弃、询问人类还是继续处理——这种结构就不再适用了。由于链式结构没有“视情况而定”的概念，开发者不得不将链式调用封装在if语句中，久而久之就构建出了一个没有文档支持、更难调试的状态机。

LangGraph则让这种状态机变得清晰可见。它提供了节点、边以及一个可在任何时刻查看的共享状态对象。其中并无什么神奇之处，而这恰恰是其优势：智能体做出的每一个决策都能在图结构定义中找到对应的内容。

1. 状态：一个共享对象以及重要的归约器

状态是每个节点读取和写入的唯一对象。没有它，上下文往往会作为函数参数传递，这样就很难确定某个具体步骤实际知晓了什么。下面的定义是一个TypedDict，其中包含一个问题、一个答案以及一个消息列表，这些内容的更新会通过add_messages reducer进行合并。

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
    question: str
    answer: str
    messages: Annotated[list, add_messages]

operator.add并非消息reducer

许多教程会用 operator.add 来标注消息字段。这样做看似合理：add 会将新内容追加到列表中而非覆盖原有内容，而这正符合对话不断延伸的需求。但问题在于它会盲目地进行字符串拼接。一旦需要更新或删除现有消息，比如在截取历史记录或替换工具调用结果时，它反而会添加重复内容，导致对话历史中充斥着过时的记录，且不会引发任何错误。

add_messages正是为这一目的而设计的。它通过ID匹配消息，若该ID已存在则直接替换原有消息，仅添加真正新的消息。规则很简单：对于存储HumanMessage和AIMessage对象的字段，使用add_messages；而对于需要简单累积内容的列表，比如记录已调用的工具清单，则继续使用operator.add。

保持状态结构尽可能简洁

第二个常见的错误是将状态设计得像数据库模式一样，为将来可能出现的各种需求都预置字段。只有当节点真正需要读取或写入某个字段时才添加它。忽视这一点的后果是显而易见的：以一个存储完整原始LLM响应及令牌使用元数据的文档处理图为例，循环处理50份文档时每个检查点的大小约为180KB，Postgres的写入时间超过400毫秒，慢到足以让等待响应的用户察觉。解决办法其实很简单：将状态简化为下游节点真正使用的那三个字段。请记住，由于有检查点机制，状态中的所有内容都会在每一步被序列化并保存。

2. 节点：仅返回发生变化的内容

节点就是一个普通的 Python 函数。它接收当前状态，执行相应操作，然后返回一个仅包含其修改过的字段的字典。这就是全部的契约内容。第一个示例会用问题调用 OpenAI 的聊天模型，并将回复写入 answer 变量中。

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini")
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

在处理图结构时——这通常是初期工作的主要部分——你可能不希望每一步都调用付费的 API。Ollama 提供的本地模型实现了相同的接口，因此节点的实现代码保持不变，调试也无需额外成本：

from langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3.1", temperature=0)
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

此版本需要在本地运行 Ollama 并下载相应模型（ollama pull llama3.1），同时安装集成包（pip install langchain-ollama）。将 temperature=0 设置为默认值还能让运行结果更具重复性，这对测试路由逻辑非常有帮助。

返回整个状态会覆盖其他更新

一个常见的错误是节点返回整个状态字典，而非仅返回已更改的键值。在小型线性图中这似乎可行，因为没有其他部分会修改这些字段。但一旦有两个节点更新了重叠的字段，其中一个节点返回的完整状态就会用旧值覆盖另一个节点的更改。这种现象看起来像路由问题，因此开发人员往往会去检查边缘逻辑，而实际原因往往是某个节点返回了过多数据。仅返回最小限度的更新还能让还原器正常工作：没有对应还原器的字段会直接被节点返回的值替换。

3. 直接边：始终连接出口节点

边决定了下一步该执行什么操作。直接边是无条件的：当节点A执行完毕后，就会执行节点B的操作。下图定义了两个节点，将answer与refine相连，再将refine与END相连，同时指定了入口点并完成了编译。

from langgraph.graph import StateGraph, END
graph = StateGraph(AgentState)
graph.add_node("answer", answer_node)
graph.add_node("refine", refine_node)
graph.add_edge("answer", "refine")
graph.add_edge("refine", END)
graph.set_entry_point("answer")
app = graph.compile()

END节点往往是人们容易忽略的部分，也是导致图似乎无限循环的典型原因。最可靠的做法是明确让图中的每条路径都终止于END节点，这样就可以通过查看定义来判断程序是否结束。一旦出现循环，这一点就显得尤为重要：如果没有通往END节点的路径，或者存在永远不成立的条件，程序就会持续循环，直到LangGraph的递归限制触发GraphRecursionError异常才停止。这一限制只是安全措施而非设计考量——每一次循环仍然会消耗计算资源。当图出现卡住的情况时，首先应检查图的定义。

4. 条件边：智能体实际做决策的地方

条件边正是让图结构具备智能性而非成为固定流程的关键。路由函数会检查当前状态并返回一个标签，而映射功能则将每个标签转换为下一个节点。在这个例子中，长度在50个字符以内的简短答案会被发送到refine，其他所有内容则会被发送到END。

def route_based_on_quality(state: AgentState) -> str:
    if len(state["answer"]) < 50:
        return "refine"
    return "done"
graph.add_conditional_edges(
    "answer",
    route_based_on_quality,
    {"refine": "refine", "done": END},
)

请注意，这种条件边替代了之前代码片段中直接从answer到refine的边。如果同时注册了这两种路径，系统会走两条路，而这通常并非人们所期望的结果。

不匹配的路由标签会以明显但难以察觉的方式出错

这里常见的错误是路由函数返回了映射字典中不存在的字符串。由此产生的错误是一种相当通用的键错误，隐藏在堆栈跟踪的深层位置，而哪怕只是末尾多一个空格，都可能耗费大量时间。一个可靠的习惯是：先编写映射表，然后再从其中复制确切的键来编写路由器。更好的做法是将这些标签定义为常量，或者用Literal["refine", "done"]标注路由器的返回类型，这样类型检查工具和阅读者就能立即看到允许的值。

5. 检查点机制：在多次调用之间保持不变的内存

检查指针将无状态函数调用转化为与内存的交互。没有它的话，每次app.invoke()都会从零开始；有了它之后，每个线程的状态都会被保存，且任何在配置中传递相同thread_id的调用都能从上一次停止的地方继续执行。在示例中，user-session-42线程上的第二次调用会记住第一个问题。

from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-session-42"}}
app.invoke({"question": "What is LangGraph?"}, config)
app.invoke({"question": "Show me a code example"}, config)   # remembers the first turn

InMemorySaver仅适用于本地开发，除此之外别无他用。它存在于进程内存中，因此服务器重启会清除所有对话记录。真正需要被用户依赖的功能必须拥有持久化后端：单个服务器可使用SQLite，而多个实例需要共享状态时则应使用Postgres。

# single-server production — pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
# multi-instance production, needs shared state across servers
# pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

如安装说明所示，每个后端都以独立包的形式提供。在当前版本中，这些数据保存器通常是通过连接字符串创建的（例如通过from_conn_string），而Postgres需要执行一次setup()调用来创建表，因此请查阅对应版本的checkpointer文档以了解具体的初始化方式。

此处可能出现的故障是在生产环境中使用了内存中的数据保存器，而在阶段环境重启并清除实时演示数据时才发现问题。好消息是，如果整体架构已经较为完善，切换成本其实很低：checkpointer只是编译时的参数，并不需要重新设计，而转换为SqliteSaver通常不到一小时即可完成。

6. 人工干预：静态断点与动态中断

大多数教程展示的模式是 interrupt_before，即一份节点名称列表，表示编译后的图在执行前会在这些节点处暂停：

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"],
)

这种方法虽然有效且易于解释，但属于静态方案。暂停点由节点名称固定决定，无法实现条件控制，也无法附加说明要求审核者关注内容的有效载荷。实际需求很快就会超出它的能力范围，因为“在此节点前暂停”与“仅当退款金额超过500美元时暂停”是不同的规则，而只有前者能用这种方式表达。

使用 interrupt() 在节点内部暂停

更灵活的方式是在节点内部调用 interrupt()。下方的节点会检查退款金额：若超过500美元，则暂停执行并将草稿及金额呈现给人工处理。第一次 invoke 调用会一直运行到该暂停点。第二次调用会在同一线程上传递 Command(resume="approve")，而传给 resume 的值会成为 interrupt() 的返回值，这样节点要么继续发送数据，要么返回取消状态。由于需要暂停状态在等待期间被存储起来，因此必须使用检查指针。

from langgraph.types import interrupt, Command
def send_email_node(state: AgentState) -> dict:
    if state["refund_amount"] > 500:
        decision = interrupt({
            "draft": state["draft"],
            "amount": state["refund_amount"],
        })
        if decision != "approve":
            return {"status": "cancelled"}
    # send the email
    return {"status": "sent"}
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "task-99"}}
app.invoke({"task": "Draft and send a refund email"}, config)
# graph pauses inside send_email_node, surfaces the interrupt payload
app.invoke(Command(resume="approve"), config)

示例状态使用了 refund_amount、draft 和 task 等字段，这些字段并不存在于之前的 AgentState 中；在真实的图中，应将这些字段声明在 AgentState 中。

恢复执行会重新运行整个节点

令人惊讶的行为是：在继续执行时，LangGraph并不会从interrupt()那行代码继续往下跑。它会从头重新执行整个节点，而这次interrupt()会返回继续执行的值而非暂停。调用之前的所有代码都会再次被执行。那些在中断前会增加计数器的节点，每次获得批准时都会将计数器加两次。因此，请确保interrupt()之前的所有操作都是可重做的，或者将那些会产生副作用的操作移到更早的节点中。同样的原则也适用于放在暂停操作之前的API调用或数据库写入操作。

能够自我批准的模型并非有人类干预

无论你选择哪种机制，向模型询问“我应该继续吗？”并信任其回复都并非人类监督——无论它被如何定义。这是智能体对自己决策的确认。而真正的审批流程则是将控制权交给图结构之外的人，并等待他们的答复。

六大基本要素一览

下方的总结将每个概念与其功能以及相关的常见错误对应起来。

+----------------------+----------------------------------------+---------------------------+
| Concept              | What it does                            | The mistake I made        |
+----------------------+----------------------------------------+---------------------------+
| State                | Shared, typed dict every node touches   | operator.add instead of   |
|                      |                                          | add_messages for chat     |
+----------------------+----------------------------------------+---------------------------+
| Nodes                | Plain functions: state in, updates out  | Returning full state,     |
|                      |                                          | not just changed fields   |
+----------------------+----------------------------------------+---------------------------+
| Direct edges         | Always go to the same next node         | Forgetting to wire END    |
+----------------------+----------------------------------------+---------------------------+
| Conditional edges    | Function inspects state, picks next node| Return value doesn't      |
|                      |                                          | match a mapping key       |
+----------------------+----------------------------------------+---------------------------+
| Checkpointing        | Persists state per thread_id            | InMemorySaver in prod     |
+----------------------+----------------------------------------+---------------------------+
| Human-in-the-loop    | Pauses for a real person, then resumes  | Non-idempotent code       |
|                      |                                          | before interrupt()        |
+----------------------+----------------------------------------+---------------------------+

合理的构建顺序

对于第一个真正的图结构，应使用 InMemorySaver 且不设置中断，让完整的循环能够从头到尾正常运行。状态应保持简洁，仅包含节点所需的信息，并确保每条条件边都能返回其映射关系所期望的标签。只有当这一切都能顺畅运行时，才应引入持久性的检查指针，并在确实需要人工干预的步骤中添加中断——通常是指涉及资金转账、发送外部邮件或删除数据等操作。

更高级的功能，包括超出 add_messages 范围的自定义归约器、将大型图拆分为可测试部分的子图，以及基于令牌级的流式处理，都建立在相同的框架之上。在仅运用这六个理念构建、调试并修复过图结构之后，这些高级功能会更容易被采用。

核心要点

• 聊天历史记录请使用add_messages，普通列表则仅使用operator.add；由于状态会在每一步都被保存，因此要保持状态简洁。
• 仅返回节点中发生变化的字段；若返回完整状态，则会 silently 覆盖并行处理或之前进行的更新。
• 为每条路径指定一条明确的通往END的路由，并通过限制重试次数来控制流程，而非依赖递归深度上限。
• 从映射关系中获取路由标签，以避免它们出现偏差。
• 将InMemorySaver视为仅用于开发阶段的工具；检查点交换的操作成本很低，应在用户开始依赖该图结构之前执行。
• 对于条件性审批，优先使用动态的interrupt()机制；在代码具备幂等性之前，请先确保其不会因重新运行节点而导致问题。

参考文档：LangGraph 的 Graph API 文档、中断处理指南以及interrupt() API 参考。