前言

大语言模型已经不只是“一问一答”的玩具了。我们越来越需要它完成多步骤的任务：一边搜索资料一边思考、调用工具获取实时数据、在输出前经过人工审核……这些场景中，应用不再是单根直线，而是一张有分支、有循环、有记忆的网。

LangChain 的链式调用（Chain）把多个步骤串成一条流水线，但它难以优雅地处理循环、条件跳转和状态持久化。为此，LangChain 团队推出了 LangGraph —— 一个基于有向图和全局状态的大模型应用编排框架。

今天这篇博客，我们就从最核心的三个概念出发，带你写出第一个 LangGraph 程序，让图“跑”起来。

1. 为什么需要 LangGraph

假设我们要做一个智能客服：

• 用户提问后，模型先判断意图。
• 如果是“咨询”，需要去知识库搜索再回答。
• 如果是“投诉”，需要生成回复草稿，必须经人工审核才能发送。
• 对话结束前，模型可能需要反复确认信息。

如果用传统链式调用，你可能会写一堆 if-else，在链的中间插入人工交互也非常困难。LangGraph 的设计正是为了这些场景：把应用的流程定义成一张图，节点是具体的处理逻辑，边是流转的规则，而所有信息都存放在一个贯穿始终的“状态”对象里。

2. 核心概念一：状态（State）

状态是整个图执行期间共享的数据结构。LangGraph 图在每一步都会接收当前状态，节点处理完后返回一个部分状态更新，框架再把这些更新合并回全局状态。

通常用 Python 的 TypedDict 定义状态，并为字段指定一个合并策略（Reducer）。最常用的合并策略是给消息列表用的 add_messages。

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 新消息会追加到现有列表末尾
    counter: int                              # 普通字段，直接覆盖

Annotated[list, add_messages] 的意思是：当节点返回 {"messages": [new_msg]} 时，不直接替换整个列表，而是把 new_msg 追加到原来的消息列表中。这非常适合记录聊天历史，我们后面会看到它的效果。

counter 没有用 Annotated，则默认是直接覆盖：节点返回 {"counter": 2}，全局状态里的 counter 就会直接变成 2。

3. 核心概念二：图与节点

图是一个 StateGraph 对象，节点就是图中的执行单元——其实就是一个普通的 Python 函数。

节点函数必须遵守一个约定：

• 接收一个参数：当前全局状态（字典形式）。
• 返回一个字典，包含想要更新的字段和值。如果某些字段没变，可以不写。

下面定义一个最简单的节点，它模拟一个“智能体”，每被调用一次就让 counter 加 1，并生成一条 AI 消息。

def agent_node(state: State):
    count = state.get("counter", 0)
    new_count = count + 1
    # 假装这是 LLM 生成的回复
    reply = f"收到，这是第 {new_count} 次回复。"
    return {
        "messages": [AIMessage(content=reply)],
        "counter": new_count
    }

然后，我们创建一个 StateGraph，把节点添加进去，并告诉图“从哪里开始”。

from langgraph.graph import StateGraph, END

graph = StateGraph(State)               # 用我们定义的 State 创建图
graph.add_node("agent", agent_node)     # 添加名为 "agent" 的节点
graph.set_entry_point("agent")          # 设置入口节点

现在图上只有一个孤零零的节点，还没办法跑——我们需要用边把它连起来。

4. 核心概念三：边

边决定了节点执行完后下一步去哪里。LangGraph 中有两种边：

• 普通边：无条件地从一个节点走到另一个节点（或直接到 END）。
• 条件边：根据节点返回的结果动态选择下一个节点。

在我们这个简单例子里，希望“agent”节点被执行最多 3 次，然后结束。所以要定义一个路由函数，根据当前 counter 的值决定是继续循环回 agent 节点，还是走向结束。

def should_continue(state: State):
    if state["counter"] >= 3:
        return "end"         # 结束的信号
    else:
        return "continue"    # 继续循环的信号

# 给 "agent" 节点添加条件边
graph.add_conditional_edges(
    "agent",                # 从哪个节点出发
    should_continue,        # 路由函数
    {
        "continue": "agent",  # 如果路由返回 "continue"，就回到 agent 节点
        "end": END            # 如果路由返回 "end"，整个图执行结束
    }
)

至此，我们的图形成了一个循环：agent -> agent -> agent ... -> END。只要 counter 小于 3，就会不断回到 agent 节点。

5. 最小可运行示例

把所有零件组装起来，编译并执行。

app = graph.compile()

# 准备初始状态：一条用户消息和计数器初值 0
initial_state = {
    "messages": [HumanMessage(content="你好，请开始工作")],
    "counter": 0
}

# 执行图
final_state = app.invoke(initial_state)

# 打印最终的消息列表
for msg in final_state["messages"]:
    print(f"[{msg.type.upper()}] {msg.content}")

输出将会是：

[HUMAN] 你好，请开始工作
[AI] 收到，这是第 1 次回复。
[AI] 收到，这是第 2 次回复。
[AI] 收到，这是第 3 次回复。

可以看到，每经过一次 agent 节点，消息列表就会自动追加一条 AI 消息，而且 counter 也在同步递增。这就是 add_messages 和状态合并机制在起作用——节点只需要关心自己产出的新消息，旧的历史会被框架自动保留。

如果你只放一个节点并用普通边直连 END，那就得到了一个最朴素的“链”；如果再加上条件边和循环，就得到了一个具备自主决策能力的 Agent。LangGraph 的核心就是这么简单。

6. 小结与预告

今天我们一起抓住了 LangGraph 的三个核心要素：

• 状态（State）：全局共享的数据字典，通过 Reducer 控制字段的合并方式。
• 节点（Node）：接受状态、返回部分更新的 Python 函数。
• 边（Edge）：用普通边定义固定流程，用条件边实现智能路由和循环。

用这“三板斧”，你已经能搭建出带记忆、能循环、会分岔的智能流程了。

下一篇博客我们会深入 持久化与人工介入：如何让图在任意步骤暂停、保存、恢复，甚至让人在半路改一改状态再继续执行。届时，你的 Agent 就真正有了“记忆”和“可干预”的能力。