L5 用 LangGraph 构建有状态 Agent

• 难度 ⭐⭐⭐（中级）
• 预计阅读 35 分钟
• 代码量 含完整可运行的 LangGraph Agent
• 先修 L1-L4 全部（特别是 L4 的 ReAct + Function Calling 原理）

课程目标：掌握 LangGraph 的核心抽象（StateGraph / Node / Edge / Checkpoint），能构建一个有长期记忆的多轮对话 Agent。

一、为什么需要 LangGraph？#

L3-L4 我们已经能用 smolagents 或手写 OpenAI API 实现 ReAct 循环。但生产环境会遇到三个问题：

1. 状态管理混乱：对话历史、工具调用结果、用户偏好——这些"状态"散落在 messages 里，难以管理

2. 没有长期记忆：每次调用都是"金鱼记忆"，聊过的事下次全忘

3. 流程不直观：ReAct 循环是"看不见的循环"，debug 时你只能看日志，画不出流程图

LangGraph 解决这三个问题的核心抽象是：把 Agent 建模成一张有向状态图。

START → assistant (LLM) → [决定调工具？]
                              ↓ 是
                          tools (执行)
                              ↓
                          assistant (循环)
                              ↓ 否
                            END

这张图既可视（能画出来），又可执行（LangGraph 帮你跑），还可记忆（每个状态都能存盘）。

二、用一个比喻：LangGraph = Agent 的"装配图纸"#

把 L3-L4 的 Agent 比作一辆手工拼装的车：

• smolagents 给你散件（@tool、CodeAgent），你按图自己装

• 手写 OpenAI API 是从轮子造起

而 LangGraph 是 Agent 的"装配图纸 + 半成品底盘"：

• 它告诉你"这辆车的发动机、车轮、座椅应该装在什么位置"

• 它提供标准的接口（节点），你只需要填内容

• 装好的车天然支持复杂的电气系统（状态管理）和车载电脑（长期记忆）

直觉：LangGraph 不是给你新功能，而是给你更好的组织方式。当你只有一个 Agent 时 LangGraph 显得重，但当你要做多步骤工作流、多 Agent 协作、带人审核的流程时，LangGraph 是唯一不让你代码乱成一团的选择。

三、LangGraph 三大核心概念#

加上两个辅助概念：

我们用 30 行代码演示这五个概念的最小例子。

四、第一个 LangGraph Agent：会思考的聊天机器人#

完整代码#

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

# ============ 1. 定义 State（状态）============

class State(TypedDict):
    """Agent 的"大脑记忆"——目前只有对话历史"""
    messages: Annotated[list, add_messages]
    # ↑ 这是 LangGraph 的精髓：
    #   - messages 是列表，存所有对话消息
    #   - Annotated[..., add_messages] 表示"新消息自动追加到列表末尾"
    #   - 这就是所谓的 "reducer"（归约函数）

# ============ 2. 定义 Node（节点）============

llm = ChatOpenAI(model="gpt-4o-mini")

def chatbot_node(state: State):
    """思考节点：调用 LLM，根据当前对话历史生成下一条回复"""
    return {"messages": [llm.invoke(state["messages"])]}
    # ↑ 返回 dict，LangGraph 会自动用 add_messages 归约函数合并到 state["messages"]

# ============ 3. 装配图 ============

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot_node)        # 加一个节点
graph_builder.add_edge(START, "chatbot")               # 开始 → 思考
graph_builder.add_edge("chatbot", END)                 # 思考 → 结束

graph = graph_builder.compile()

# ============ 4. 运行 ============

result = graph.invoke({"messages": [{"role": "user", "content": "你好，我是张三"}]})
print(result["messages"][-1].content)
# 输出：你好张三！很高兴认识你。

result = graph.invoke({"messages": [{"role": "user", "content": "我叫什么名字？"}]})
print(result["messages"][-1].content)
# 输出：抱歉，我没有你的名字信息。
# ↑ 问题来了：它忘了上一轮说过"我叫张三"！

问题暴露：上面的代码每次 invoke 都是全新开始——LangGraph 不知道上一次的对话。

解法：用 Checkpoint + Thread ID。我们马上讲。

五、Checkpoint + Thread：让 Agent 拥有"金鱼变金鱼王"的记忆#

LangGraph 的记忆机制由两个组件组成：

• Checkpoint（检查点）：每次状态变化都保存一份"快照"

• Thread ID（线程 ID）：给每次会话一个唯一标识，同一 Thread 的所有检查点串起来就是完整的对话历史

一句话：Checkpoint + Thread = 对话级长期记忆。

加记忆的完整代码（重点看第 24-28 行）#

from langgraph.checkpoint.memory import MemorySaver

# ... 前面的 State、Node 定义同上 ...

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot_node)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# ============ 关键改动：加 Checkpointer ============
memory = MemorySaver()  # 内存版 checkpointer（重启会丢）
graph = graph_builder.compile(checkpointer=memory)

# ============ 关键改动：加 thread_id ============
config = {"configurable": {"thread_id": "user-123"}}
# ↑ 这个 thread_id 就像"会话房间号"——同一房间内的对话 Agent 都能记住

# ============ 第一次对话 ============
result = graph.invoke(
    {"messages": [{"role": "user", "content": "你好，我是张三"}]},
    config=config  # ← 同一个 config
)
print(result["messages"][-1].content)
# 输出：你好张三！很高兴认识你。

# ============ 第二次对话（Agent 应该记住名字）============
result = graph.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？"}]},
    config=config  # ← 同一个 thread_id，对话历史自动恢复
)
print(result["messages"][-1].content)
# 输出：你叫张三呀！刚才你说过的。

这一次 Agent 记住了！

MemorySaver 把第一次的 messages 列表存下来。第二次 invoke 时 LangGraph 自动从同一 thread_id 加载历史，LLM 看到了"我是张三"——所以能正确回答。

六、加工具：让 Agent 能"做事"#

光聊天不够，我们加一个搜索工具（来自 LangGraph 官方 quickstart）：

from langchain_tavily import TavilySearch
from langgraph.prebuilt import ToolNode, tools_condition

# ============ 1. 定义工具 ============
search_tool = TavilySearch(max_results=2)
tools = [search_tool]
llm_with_tools = llm.bind_tools(tools)  # 把工具"绑"到 LLM 上

# ============ 2. 定义节点 ============
def chatbot_node(state: State):
    """思考节点"""
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

tool_node = ToolNode(tools)  # LangGraph 内置的工具执行节点

# ============ 3. 装配图（这次有条件边）============
graph_builder = StateGraph(State)

graph_builder.add_node("chatbot", chatbot_node)
graph_builder.add_node("tools", tool_node)  # 加一个工具执行节点

graph_builder.add_edge(START, "chatbot")   # 开始 → 思考

# ↓ 关键：条件边——思考完决定调工具还是结束
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,  # 内置函数：检查 LLM 是否想调工具
    {
        "tools": "tools",  # 如果想调 → 走 tools 节点
        END: END            # 如果不想 → 结束
    }
)

graph_builder.add_edge("tools", "chatbot")  # 工具执行完 → 回到思考（循环）

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

# ============ 4. 跑 ============
config = {"configurable": {"thread_id": "user-456"}}

for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "2026 年诺贝尔物理奖是谁？"}]},
    config=config,
    stream_mode="values"
):
    chunk["messages"][-1].pretty_print()

运行过程（简化）：

=== 第 1 轮：思考 ===
Human: 2026 年诺贝尔物理奖是谁？
AI: (decide to call search) → 工具调用

=== 第 2 轮：工具 ===
Tool: 搜索返回结果...

=== 第 3 轮：思考（基于工具结果）===
AI: 2026 年诺贝尔物理奖授予了 XXX，原因是 YYY。

=== 结束 ===

这张图完整画出来就是：

START
  ↓
chatbot (LLM 思考)
  ↓ tools_condition
  ├─ 想调工具 → tools (执行)
  │                ↓
  │              chatbot (再思考)  ← 循环
  └─ 不想调 → END

这就是 ReAct 循环的可视化！ L2 我们讲的 Thought-Action-Observation 循环，在 LangGraph 里就是 chatbot → tools_condition → tools → chatbot 这条路径。

七、State 进阶：自定义状态字段#

上面的 State 只有 messages。生产环境的 State 通常还要存很多东西：

from typing import Annotated
from langgraph.graph.message import add_messages

class State(TypedDict):
    # 对话历史（用内置归约函数）
    messages: Annotated[list, add_messages]
    
    # 用户偏好（不需要归约函数，直接覆盖）
    user_name: str
    
    # 工具调用计数（用自定义归约函数）
    tool_call_count: Annotated[int, lambda current, update: current + update]
    
    # 搜索结果缓存
    search_results: list[str]

reducer（归约函数） 是 LangGraph 的核心机制——它定义"新数据怎么和旧数据合并"。

八、检查历史：get_state 与 get_state_history#

调试时你想看 Agent 的每一步状态？LangGraph 提供两个工具：

# 查看当前状态
current_state = graph.get_state(config)
print(current_state.values)  # 完整 state 字典

# 查看完整历史（所有 checkpoints）
history = list(graph.get_state_history(config))
for i, state in enumerate(history):
    print(f"Step {i}: {state.values['messages'][-1].content[:50]}")

这在生产调试中救命：Agent 答错了，你能直接回看"它在哪一步开始走偏"。

九、生产级 Checkpoint：内存版 vs 持久版#

MemorySaver 是内存版，重启进程就丢。生产环境需要持久化：

# 内存版（适合 demo）
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()

# SQLite 版（适合小型生产）
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string("checkpoints.db")

# PostgreSQL 版（适合大型生产）
from langgraph.checkpoint.postgres import PostgresSaver
memory = PostgresSaver.from_conn_string("postgresql://...")

# Redis 版（适合高并发）
from langgraph.checkpoint.redis import RedisSaver
memory = RedisSaver.from_conn_string("redis://...")

选择标准：

• Demo / 测试：MemorySaver

• 个人项目：SqliteSaver（单文件，备份容易）

• 中型生产：PostgresSaver（支持事务、并发）

• 高并发生产：RedisSaver（速度最快）

十、和 L3-L4 的对比#

结论：

• L3 smolagents 适合快速原型和单 Agent

• L4 手写让你理解原理

• L5 LangGraph 是生产级多步骤 / 多 Agent 的标配

十一、动手时间：亲手跑 + 改#

步骤 1：跑通"加工具的 LangGraph Agent"（10 分钟）#

复制上面第六节的完整代码到 langgraph_v1.py：

1. pip install langgraph langchain-openai langchain-tavily tavily-python

2. 设置 OPENAI_API_KEY 和 TAVILY_API_KEY（Tavily 搜索有免费额度，到 tavily.com 注册获取）

3. 跑通，观察流式输出

步骤 2：可视化你的图（5 分钟）#

from IPython.display import Image, display

# 显示状态图（需要 Graphviz）
display(Image(graph.get_graph().draw_mermaid_png()))

你会看到一张真正的状态图——和你画的 ASCII 流程图一模一样，但更漂亮。

步骤 3：加一个自定义状态字段（10 分钟）#

把 tool_call_count 加进 State，每次调用工具时累加。最后 graph.get_state(config) 看计数是否正确。

class State(TypedDict):
    messages: Annotated[list, add_messages]
    tool_call_count: Annotated[int, lambda current, update: current + update]

def tool_node_with_counter(state: State):
    """包装 ToolNode，每次调用累加计数器"""
    result = tool_node.invoke(state)  # 调用原始 ToolNode
    return {
        **result,
        "tool_call_count": 1  # 每次调用 +1
    }

graph_builder.add_node("tools", tool_node_with_counter)

config = {"configurable": {"thread_id": "user-789"}}
graph.invoke({"messages": [{"role": "user", "content": "北京天气怎么样？"}]}, config=config)
state = graph.get_state(config)
print(state.values["tool_call_count"])  # 应该输出 1

十二、小结#

十三、Read next#

下一节课 L6 多 Agent 系统：从单兵作战到团队协作——这是本课程的最后一课。学完你会：

• 理解为什么复杂任务必须靠多 Agent 分工

• 掌握三种主流多 Agent 模式：Manager-Worker / Group Chat / Debate

• 能为一个真实业务场景设计多 Agent 架构图

主要参考：

• src-002 §Unit 3 LangGraph framework — HF 课程对 LangGraph 的官方阐述

• LangGraph 官方文档 - Persistence — Checkpoint 机制的官方说明

• Multi-Agent — 多 Agent 协作的设计模式

十四、Sources#

• src-001 — Microsoft AI Agents for Beginners · T1 · multi-agent

• src-002 — Hugging Face Agents Course · T1 · Course

• LangGraph 官方文档 · T1 · Persistence Concepts

• LangGraph 官方教程 · T1 · Quickstart

💬 有疑问？ 试试问我："为什么 LangGraph 不用 OpenAI 风格的 JSON function calling？"、"MemorySaver 和 SqliteSaver 性能差多少？"、"怎么可视化我设计的 Agent 流程？"——我会用代码或比喻为你解答。