先分清三个容易混的词#

1
state       = 当前整份工作数据
2
messages    = state 里可能存在的一个字段
3
checkpointer = 保存/恢复 state 快照的组件

例如后续聊天 Agent 的 state 可能是：

1
{
2
    "messages": [...],
3
    "retrieved_docs": [...],
4
    "risk_level": "low",
5
}

messages 不是整个 state；它只是其中一部分。

这段代码的执行顺序#

1
初始 state
2
{"question": " Checkpointer 是什么？ "}
3

4
START
5
  -> normalize_question
6
     返回 {"normalized_question": "checkpointer 是什么？"}
7
  -> create_answer
8
     返回 {"answer": "准备回答：checkpointer 是什么？"}
9
  -> END
10

11
最终 state
12
{
13
  "question": " Checkpointer 是什么？ ",
14
  "normalized_question": "checkpointer 是什么？",
15
  "answer": "准备回答：checkpointer 是什么？"
16
}

Node 不需要返回完整 state#

看这个 node：

1
def normalize_question(state: LearningState) -> dict[str, str]:
2
    return {"normalized_question": state["question"].strip().lower()}

它读取 state["question"]，但只返回自己更新的字段：

1
{"normalized_question": "checkpointer 是什么？"}

LangGraph 会把这份更新合并回工作流 state。入门阶段先记住默认直觉：同名字段的新值会覆盖旧值。

以后你会学 reducer，才处理“列表要追加，不是覆盖”这类规则；本章先不展开。

Node：做事#

1
builder.add_node("normalize_question", normalize_question)

左边字符串是 node 名称，右边是实际执行的 Python 函数。

Node 的职责是：

1
读 state
2
-> 做一次工作
3
-> 返回 state 更新

它不会天然决定下一步去哪里。

Edge：决定固定顺序#

1
builder.add_edge(START, "normalize_question")
2
builder.add_edge("normalize_question", "create_answer")
3
builder.add_edge("create_answer", END)

边把执行顺序写出来：

1
START -> normalize_question -> create_answer -> END

这就是为什么说：

1
node 负责做事
2
edge 负责安排下一步
3
state 负责传递工作数据

换成执行时序就是：

1
builder.add_edge("normalize_question", "create_answer")

这行代码此刻不会执行 normalize_question，也不会立刻执行 create_answer。它只是把一条规则登记到图里：

1
当 normalize_question 节点完成
2
-> 把更新后的 state 交给 create_answer 节点

等你之后调用 graph.invoke(...)，LangGraph 才会按照这条边安排节点执行。因此，Edge 不是“做具体工作的函数”，而是图的流程连接规则。普通边指定固定下一步；条件边则会根据当前 state 或消息判断应该走哪条路。

两种方法并排看：

1
# 普通边：终点固定
2
builder.add_edge("normalize_question", "create_answer")
3

4
# 条件边：先运行路由函数，再根据返回值选择终点
5
builder.add_conditional_edges("model", tools_condition)

因此，add_edge 的终点在注册时就确定；add_conditional_edges 的终点要等 invoke() 运行到 source 节点后，调用路由函数才能确定。

别把它理解成 node 返回下一个 node 名。这个例子的固定流向由 edge 定义；“根据 state 选不同边”是条件边，后面再学。

先认四个已有或即将出现的对象#

1
from app.tools.knowledge_base import search_knowledge_base
2
from langgraph.prebuilt import ToolNode, tools_condition
3

4
tools = [search_knowledge_base]
5
model_with_tools = llm.bind_tools(tools)

bind_tools(...) 不是执行工具。它只是把工具的 schema 交给模型，作用和你第 26 章手写 Function Calling 时“把 tools 发给模型”相同。

为什么同一个 tools 要用两次？因为它们面对的是两个不同的消费者：

1
model_with_tools = llm.bind_tools(tools)  # 给模型看：工具叫什么、参数是什么、何时使用
2
tool_node = ToolNode(tools)               # 给执行器用：真正找到并调用哪个 Python 函数

前者解决“模型能不能提出规范的 tool_call”，后者解决“提出后由谁真正执行”。ToolNode 不会自动把工具说明发送给模型；bind_tools 也不会自己执行 Python 工具。

1
没有 bind_tools：模型通常不知道可用工具，可能直接回答，工具节点没有调用可执行
2
没有 ToolNode：模型可能提出 tool_call，但没有节点执行并回传工具结果

注意：本章使用的名字是 model_with_tools，不是 LangGraph 固定提供的 call_with_tools。model_with_tools 只是我们给 llm.bind_tools(tools) 返回对象取的变量名；如果你自己写成 call_with_tools，也只是本地变量改名，不会新增一个特殊函数。

把上一章的 create_agent(...) 也放在这里对照：

1
llm = ChatDeepSeek(...)                 # ChatDeepSeek 类创建模型实例
2
agent = create_agent(model=llm, tools=tools)  # 工厂函数创建 Agent
3
model_with_tools = llm.bind_tools(tools)      # 模型实例绑定工具后的对象

create_agent(...) 不是模型类，也不是 ChatDeepSeek(...) 的别名；它接收模型和工具，并把模型调用、工具执行和循环封装成一个 Agent。第六关把这层封装拆开，分别展示 call_model、ToolNode 和 Edge。

Model Node 是你写的普通 Python 函数#

1
from langgraph.graph import MessagesState
2

3

4
def call_model(state: MessagesState):
5
    response = model_with_tools.invoke(state["messages"])
6
    return {"messages": [response]}

这段函数执行前后，State 怎么变#

假设执行前的 State 是：

1
state = {
2
    "messages": [HumanMessage(content="退款需要几天？")]
3
}

函数先把已有消息列表交给模型：

1
response = model_with_tools.invoke(state["messages"])

此时 response 是模型新返回的一条 AIMessage：没有工具需求时是普通回答；需要工具时则可能带有 tool_calls。函数返回的是一份 State 更新：

1
{"messages": [response]}

它不是完整的新 State，也不是函数直接修改旧 State。MessagesState 为 messages 配置了消息追加/合并规则，所以 LangGraph 会把这条新 AIMessage 合并回原消息列表：

1
执行前：HumanMessage
2
        -> call_model
3
函数返回：AIMessage
4
        -> LangGraph 合并
5
执行后：HumanMessage, AIMessage

如果 AIMessage 带有 tool_calls，后续 tools_condition 会把流程送到 ToolNode；如果没有，就把这条普通回答保留在 messages 后结束。

这里的 call_model 才是普通 Python 函数；函数内部调用的是 model_with_tools.invoke(...)。因此不要把 call_model、model_with_tools 和 call_with_tools 当成同一个东西。

call_model 不是 LangGraph 内置类。它就是一个普通 Python 函数，注册到图以后才成为一个 model node：

1
builder.add_node("model", call_model)

它读取 state["messages"]，调用模型，然后把模型的新消息写回 State。模型的新消息可能是普通回答，也可能携带 tool_call。

Tool Node 是 LangGraph 提供的预制对象#

1
tool_node = ToolNode(tools)
2

3
builder.add_node("tools", tool_node)

ToolNode 是一个类；ToolNode(tools) 创建的是能执行工具的对象。它读取模型消息里的 tool_call，找到对应的 search_knowledge_base，调用它，并把工具结果作为消息写回 State。

你当然也可以自己写一个普通函数来执行工具，但 ToolNode 已经处理了常见的工具调用、结果回写和错误处理。入门阶段先使用它，后面再拆开手写。

再把对象连成一张图#

1
from langgraph.graph import END, START, MessagesState, StateGraph
2
from langgraph.prebuilt import ToolNode, tools_condition
3

4

5
builder = StateGraph(MessagesState)
6
builder.add_node("model", call_model)
7
builder.add_node("tools", ToolNode(tools))
8

9
builder.add_edge(START, "model")
10
builder.add_conditional_edges("model", tools_condition)
11
builder.add_edge("tools", "model")
12

13
graph = builder.compile()

这里的 START 只在一次 graph.invoke(...) 的开头经过一次，不会被 tools -> model 这条回边重新触发。循环的准确位置是：

1
START
2
  -> model
3
  -> tools_condition
4
       ├─ 有 tool_call -> tools -> model
5
       └─ 没有 tool_call -> END

因此，工具执行后回到的只是 model 节点；这个新的 model 响应完成后，还会再次经过 tools_condition。如果这次没有新的 tool_call，就走 END；如果还有新的 tool_call，就再次走 tools。

注意：builder.add_edge("tools", "model") 不是流程终止，而是把工具结果送回模型继续处理。这里先结束的是“图的节点和边定义”，下一步本来应当是 compile()，再下一步才是 graph.invoke(...)。本节为了先看懂对象形状，暂不要求你继续完成真实 Agent 的调用。

tools_condition：决定模型下一步去哪#

tools_condition 是 LangGraph 提供的路由函数。它只检查一件事：模型刚才的消息里有没有 tool_call。

它的真实形态可以先看成一个普通函数：

1
next_step = tools_condition(state)

它读取 state["messages"] 的最后一条消息：

1
最后一条 AIMessage 有 tool_calls
2
  -> 返回 "tools"
3

4
最后一条 AIMessage 没有 tool_calls
5
  -> 返回 "__end__"（内部结束标签，对应 END）

builder.add_conditional_edges("model", tools_condition) 的意思就是：模型节点执行完后，把当前 state 交给这个函数，让它返回下一步的目的地。它不执行模型、不执行工具，也不修改 State；它只负责路由判断。

1
有 tool_call  -> 去 "tools"
2
没有 tool_call -> 去 END

因此本例不需要手写 builder.add_edge("model", END)：tools_condition 已经负责在“没有新工具调用”时结束流程。END 不是普通业务函数，而是 LangGraph 提供的结束标记。

这里的“没有 tool_call”不是说模型没有输出内容。模型可能已经生成了完整的普通回答：

1
model 节点等待模型完成本次响应
2
  -> 得到 AIMessage(content="最终回答", tool_calls=[])
3
  -> tools_condition 发现没有 tool_call
4
  -> 走 END，但 AIMessage 的回答已经保留在 State/messages 中

tools_condition 只会在 model 节点完成一次响应后运行，不会在模型生成过程中打断模型。普通回答已经写入 messages，再走 END；END 只结束后续节点，不会删除回答。

因此真实执行路径是：

1
用户消息
2
  -> call_model（普通函数，调用绑定工具后的模型）
3
  -> tools_condition（查看模型是否提出 tool_call）
4
     -> 有：ToolNode(tools) 执行 search_knowledge_base
5
           -> 工具结果写回 messages
6
           -> call_model 再次调用模型，生成最终回答
7
     -> 无：END

这正是你已学过的 Function Calling loop。区别只是：第 26 章你手写“判断、找函数、执行、回传”；这里 LangGraph 用 Node 和 Edge 把同一流程显式画出来。

多次调用工具时如何循环#

一次工具调用不是整个流程的终点。模型拿到工具结果后，可能继续提出另一个 tool_call；只要还有 tool_call，就会再次经过 tools 节点，再回到 model：

1
model：调用 search_weather
2
  -> tools：执行并返回天气
3
  -> model：根据天气，继续调用 search_calendar
4
  -> tools：执行并返回日历
5
  -> model：根据两个结果生成普通回答
6
  -> tools_condition：发现没有 tool_call
7
  -> END

所以 builder.add_edge("tools", "model") 形成的是循环回边，不是“工具只执行一次”。真正结束的条件是：某次 model 响应中没有新的 tool_call。如果图一直产生工具调用而始终不结束，LangGraph 会在达到递归限制后报错；生产代码还应限制工具次数、校验参数并处理异常。

和你当前 create_agent(...) 的关系#

本节边界#

这节的目标是认得真实代码形态和调用链，不要求你现在复制运行完整图。MessagesState、tools_condition 和完整 Agent 实战会在下一章逐个写出来；现在只需能指出：哪个是普通函数、哪个是框架类、哪个对象真正执行工具。

安全边界没有变：模型 node 只能提出 tool_call；ToolNode 才执行 Python 工具。LangGraph 不能替你跳过参数校验、权限校验或高风险确认。

1. 读懂#

不看代码，先口头说出：

1
初始 state 里有什么？
2
第一个 node 新增或更新了什么？
3
第二个 node 读取了什么、又更新了什么？
4
最终 result 是什么？

2. 跟写#

先确认下面三条 edge 组成的是一条直线：

1
builder.add_edge(START, "normalize_question")
2
builder.add_edge("normalize_question", "create_answer")
3
builder.add_edge("create_answer", END)

然后只做一个“读图实验”：删除这条结束边：

1
builder.add_edge("create_answer", END)

再改为：

1
builder.add_edge("create_answer", "normalize_question")

这时图才会变成：

1
normalize_question -> create_answer -> normalize_question

先不要运行它。没有结束边的循环图会在达到框架的递归限制后报 GraphRecursionError；这个练习只用于确认：改 edge 改的是流程，不是 node 内的业务逻辑。

3. 独立重写#

把最小图换成“学习计划”版本：

1
输入 topic
2
-> make_outline node 生成提纲
3
-> write_preview node 生成一段预览
4
-> END

要求：

1. State 至少有 topic、outline、preview。
2. 两个 node 都只返回自己更新的字段。
3. 运行后 result["preview"] 能读到 result["outline"] 的内容。

坑 1：以为 compile() 会执行图#

1
graph = builder.compile()

这一步只是得到可运行对象。只有 graph.invoke(...) 才会执行 node。

坑 2：以为 node 要返回整个 state#

1
# 不必重复返回所有字段
2
return {"answer": "..."}

Node 返回它负责更新的字段即可。

坑 3：把 state 当成 messages#

聊天工作流里 messages 很常见，但它只是一个字段。普通工作流可以完全没有 messages，例如本章的 question、normalized_question、answer。

坑 4：没有 checkpointer 就以为图不能运行#

最小图不需要 checkpointer。它负责跨步骤/跨调用保存恢复 state，不负责让 node 本身“能运行”。

坑 5：以为 LangGraph 自动保证安全#

图能规定流程，不能自动判断你是否该执行删除、付款、发送邮件等动作。工具参数、权限和人工确认仍由你的业务代码负责。

1. 核心思想是什么？#

LangGraph 用共享 state、执行 node、连接 edge 来显式定义工作流，而不是把执行顺序藏在框架内部。

2. 它解决什么问题？#

当 Agent 不再是一条固定直线，而需要明确分支、循环、状态恢复或人工介入时，你可以看见并控制每一步。

3. 为什么不一直用 create_agent 或手写 loop？#

简单标准 Agent 用 create_agent 更省事；为了学习底层或完全自定义，用手写 loop 最直观；当流程变成多个明确步骤和状态转移时，LangGraph 更容易组织、观察和扩展。

4. 在本项目里怎么识别？#

看到 StateGraph(...)、add_node(...)、add_edge(...)、compile()、invoke(...) 时，就知道它在显式构建状态工作流；看到 InMemorySaver() 和 thread_id 时，就知道它在给 state 增加保存和恢复能力。