09 · LangGraph 核心概念

本章目标：理解为什么 Agent 需要"图式编排"，并掌握 LangGraph 的四大核心概念——State、Node、Edge、Checkpointer。

本章你将学到

1. 线性 Chain 与 AgentExecutor 的局限，为什么需要图；
2. State（状态即唯一事实来源）与 add_messages 归约器；
3. 节点、边、条件边——如何把 Python 函数编排成图；
4. Checkpointer（检查点）——让对话可持久化、可恢复。

前置知识

• 已完成第 02 章的手写 Agent 循环；
• 了解 LangChain 的 LCEL（第 07 章）；
• 会用 @tool 定义工具（第 08 章）。

9.1 为什么需要"图"

回顾第 2 章手写的 Agent 循环：

while True:
    resp = model.invoke(messages)
    if not resp.tool_calls:
        return resp.content
    for tc in resp.tool_calls:
        messages.append(ToolMessage(execute(tc)))

它能跑，但遇到这些需求就吃力了：

这些问题本质上是控制流 + 状态管理问题。而控制流 + 状态，正是图（Graph）最擅长表达的——LangGraph 就是把 Agent 的每一步画成一张有向图：节点是"做一件事"，边是"做完去哪"，状态是"共享的黑板"。

Java 工程师类比：StateGraph ≈ Spring StateMachine / Activiti 这类工作流引擎。区别在于 LangGraph 的"节点"是 LLM 推理或工具执行，状态就是对话上下文；而 Activiti 的节点是人工/系统任务。

9.2 四个核心概念

9.2.1 State（状态）：唯一的"黑板"

LangGraph 里所有节点通过共享的 State 传递数据。State 就是一个 TypedDict：

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

class State(TypedDict):
    # messages：对话历史。add_messages 是"归约器(reducer)"，见 9.3
    messages: Annotated[list, add_messages]
    # intent：普通字段，后写的覆盖先写的
    intent: str

规则只有两条：

1. 节点读取 State 的字段，返回要更新的字段（返回值会合并进 State）；
2. 普通字段是"覆盖"语义；带归约器的字段是"累积"语义（add_messages 把新消息追加进列表）。

这就像工作流引擎里的流程变量——只不过这里存的是消息与中间结果。

9.2.2 Node（节点）：一个普通 Python 函数

def greet_node(state: State) -> dict:
    """读取 state，返回要更新的字段。"""
    return {"messages": [{"role": "assistant", "content": "你好，我是客服。"}]}

节点就是函数：入参是 State，返回值是"要写入 State 的增量"。没有魔法。

9.2.3 Edge（边）与条件边：控制流

• add_edge(A, B)：A 执行完无条件去 B；
• add_conditional_edges(A, router_fn, mapping)：A 执行完，把结果交给 router_fn(state)，按返回的字符串路由到不同节点。

9.2.4 Checkpointer（检查点）：持久化与恢复

给图配一个 checkpointer 后，每次执行结束都会保存一份"状态快照"。下次带上同一个 thread_id 调用，就能接着上次的状态继续——这就是多轮对话的记忆底座，也顺带实现了"回放与时间旅行"。

Java 类比：Checkpointer ≈ 给流程状态做持久化快照，thread_id ≈ 会话 ID / 流程实例 ID。

9.3 add_messages 归约器（reducer）

State 字段的默认语义是"后写覆盖先写"。但消息必须追加，不能覆盖——所以 messages 字段用 Annotated[list, add_messages] 声明归约器。

add_messages 做的事：把新消息拼进已有列表。它还支持"按 id 合并"：如果新消息带 id 且与旧消息相同，会覆盖而不是重复追加——这在"重新生成/时间旅行"时很有用。

from langchain_core.messages import AIMessage, HumanMessage

old = [HumanMessage(content="你好")]
merged = add_messages(old, [AIMessage(content="您好，有什么可以帮您？")])
# merged = [HumanMessage("你好"), AIMessage("您好，有什么可以帮您？")]

9.4 最小可运行示例：两节点图

"""LangGraph 最小示例：两节点线性图。"""
from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    text: str


def normalize_node(state: State) -> dict:
    """节点1：清洗输入。"""
    return {"text": state["text"].strip()}


def upper_node(state: State) -> dict:
    """节点2：大写化。"""
    return {"text": state["text"].upper()}


builder = StateGraph(State)
builder.add_node("normalize", normalize_node)
builder.add_node("upper", upper_node)
builder.add_edge(START, "normalize")
builder.add_edge("normalize", "upper")
builder.add_edge("upper", END)

graph = builder.compile()

result = graph.invoke({"text": "  hello agent  "})
print(result["text"])   # "HELLO AGENT"

流程：START → normalize → upper → END。节点返回的 dict 被合并进 State，下一个节点读到更新后的值。

9.5 条件边：意图路由

把 Agent 的第一步做成"判断意图"：查航班走工具链，闲聊直接答。

"""条件边示例：意图路由。"""
from typing import Literal, TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph

import os
from dotenv import load_dotenv

load_dotenv()

model = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)


class State(TypedDict):
    messages: list
    intent: str


def classify_node(state: State) -> dict:
    """节点A：用结构化输出判断意图（fly=航班类问题，chat=闲聊）。"""
    resp = model.with_structured_output(
        type("Intent", (), {"__annotations__": {"intent": Literal["fly", "chat"]}})
    ).invoke(state["messages"])
    return {"intent": resp.intent}


def fly_node(state: State) -> dict:
    """航班意图分支：走工具/检索（本章示意，第 10 章补全）。"""
    return {"messages": [AIMessage(content="好的，我来为您查询航班信息。")]}


def chat_node(state: State) -> dict:
    """闲聊分支。"""
    reply = model.invoke(state["messages"])
    return {"messages": [AIMessage(content=reply.content)]}


def route_by_intent(state: State) -> str:
    """路由函数：根据 intent 字段决定下一步去哪个节点。"""
    return "fly" if state["intent"] == "fly" else "chat"


builder = StateGraph(State)
builder.add_node("classify", classify_node)
builder.add_node("fly", fly_node)
builder.add_node("chat", chat_node)

builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    route_by_intent,
    {"fly": "fly", "chat": "chat"},   # 返回值 → 目标节点
)
builder.add_edge("fly", END)
builder.add_edge("chat", END)

graph = builder.compile()

print(graph.invoke({"messages": [HumanMessage(content="查一下 CA1501")]}))
# intent=fly，走 fly 节点

关键语法：add_conditional_edges(来源节点, 路由函数, {返回值: 目标节点})。路由函数接收 State、返回一个字符串，图按映射表跳转。

type("Intent", (), {...}) 是运行时动态造一个带类型注解的类，配合 with_structured_output 让模型输出 fly/chat。生产中建议用 pydantic BaseModel 定义（见第 03 章）。

9.6 Checkpointer：多轮对话与恢复

没有 Checkpointer 的图是无状态的：每次 invoke 都是全新开始。加上它，同一个 thread_id 的多轮调用会共享历史——这就是聊天机器人的会话记忆。

"""Checkpointer 示例：多轮对话记忆。"""
from typing import Annotated, TypedDict

from dotenv import load_dotenv
from langchain_core.messages import AIMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

load_dotenv()

model = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0.3,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)


class State(TypedDict):
    messages: Annotated[list, add_messages]   # 关键：消息用归约器累积


def chat_node(state: State) -> dict:
    reply = model.invoke(state["messages"])
    return {"messages": [AIMessage(content=reply.content)]}


builder = StateGraph(State)
builder.add_node("chat", chat_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)

# 配 checkpointer：每次执行后保存状态快照
graph = builder.compile(checkpointer=MemorySaver())

# thread_id 标识"同一会话"，跨多次 invoke 共享历史
config = {"configurable": {"thread_id": "user-1001"}}

graph.invoke({"messages": [HumanMessage(content="我叫小明")]}, config=config)
graph.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config=config)
# 第二次提问能答出"小明"——因为第一次的对话被保存并在第二次读到了

• MemorySaver 是内存版 Checkpointer（重启丢数据），生产用 SqliteSaver / Postgres/Redis 实现；
• 换个 thread_id 就是全新会话；不传 thread_id 则每次都是新会话。

这正好印证了第 2 章的结论：LLM 本身无状态，"记忆"= 在 State 里累积消息 + 每轮全量带上。LangGraph 把这件事做成了基础设施。

9.7 调试与可视化

LangGraph 内置工具，能让你的图"看得见"：

# 打印图结构（Mermaid 源码）
print(graph.get_graph().draw_mermaid())

# 流式观察每一步的执行过程
for event in graph.stream({"text": "hi"}, stream_mode="updates"):
    print(event)   # 每步是哪个节点、更新了什么

• get_graph().draw_mermaid()：输出 Mermaid 图，粘贴到任意 mermaid 渲染器即可看到流程图；
• stream(..., stream_mode="updates")：逐步吐出"哪个节点更新了 State 的什么"，排查卡点和状态污染非常好用。

9.8 小结

• State 是唯一事实来源：节点读它、返回增量更新它；普通字段覆盖，add_messages 字段累积；
• Node 就是普通 Python 函数，入参 State、返回 dict，没有魔法；
• Edge / 条件边 表达控制流：顺序、分支、循环（循环=边的指向可以让节点回环，第 10 章就是靠它实现 Agent 循环）；
• Checkpointer + thread_id 提供会话记忆与恢复能力，是"多轮对话"的地基；
• 对比第 2 章手写循环：手写版把控制流写在 while 里，LangGraph 把控制流画在图上——可分支、可持久化、可观察。

下一章，我们用这些零件搭出真正的 Agent：第 10 章。

练习

1. 在 9.4 的两节点图中间加一个节点 reverse（反转字符串），把三条边连对，跑通。
2. 给 9.5 的意图路由加第三个意图 "refund"（退改签），走第三个分支。
3. 把 9.6 的 thread_id 换成新的，验证"新会话不带旧记忆"；再试试两个 thread_id 交替调用，观察 State 相互独立。
4. 用 get_graph().draw_mermaid() 导出 9.5 的图，肉眼确认分支是否正确。

参考答案

<details>
<summary>练习1（插入 reverse 节点）</summary>

在 9.4 的"uppercase → concat"两节点图中插入 reverse，让数据流变为：uppercase → reverse → concat。

def reverse_node(state: State) -> dict:
    return {"text": state["text"][::-1]}     # 字符串反转

graph.add_node("reverse", reverse_node)
graph.add_edge("uppercase", "reverse")       # 关键改动：原 9.4 是 uppercase → concat
graph.add_edge("reverse", "concat")

三条边：START → uppercase（原样）、uppercase → reverse（改这里）、reverse → concat（新加）。验证：

graph.invoke({"text": "abc"})
# uppercase: "ABC" → reverse: "CBA" → concat 拼接

要点：节点只通过 State 通信，调整数据流 = 调整边的指向，函数本体一行都不用动——这就是把控制流画在图上（而不是写在 while 里）的好处。
</details>

<details>
<summary>练习2（意图路由加 refund 分支）</summary>

把路由函数从二分支改三分支：

def route_intent(state: State) -> str:
    intent = state["intent"]
    if intent == "flight":
        return "flight_node"
    elif intent == "hotel":
        return "hotel_node"
    elif intent == "refund":          # 新增意图
        return "refund_node"
    return "fallback_node"            # 兜底分支别忘了

# 条件边加一行映射
graph.add_conditional_edges(
    "router",
    route_intent,
    {"flight": "flight_node", "hotel": "hotel_node",
     "refund": "refund_node", "fallback": "fallback_node"},
)

同时给 route_intent 前的意图分类 Prompt 加第三个意图的 few-shot 示例（意图标签枚举要一致，否则路由失配会掉进 fallback）。验证：分别问三类问题 + 一个"都不是"的问题，确认四路都走对。
</details>

<details>
<summary>练习3（thread_id 与 State 隔离）</summary>

config_a = {"configurable": {"thread_id": "session-a"}}
config_b = {"configurable": {"thread_id": "session-b"}}

graph.invoke({"messages": [("human", "我叫张三")]}, config=config_a)
graph.invoke({"messages": [("human", "我叫李四")]}, config=config_b)

# 新 thread_id 不带旧记忆
print(graph.invoke({"messages": [("human", "我叫什么？")]}, config=config_b))
#  → 回答"李四"：A 会话的记忆没有污染 B

# 交替调用，State 相互独立
graph.invoke({"messages": [("human", "我说我多大了吗？")]}, config=config_a)
graph.invoke({"messages": [("human", "继续说，你记得什么")]}, config=config_b)

验证两个结论：

1. 新 thread_id = 新会话：thread_id 不同，Checkpointer 按 thread_id 分桶存 State，互不可见；
2. 交替调用不串台：A、B 两个会话交替 invoke，各自的 messages 独立累积——线程隔离的粒度就是 thread_id。

生产映射：一个用户/一个会话 = 一个 thread_id；换设备/换会话就换 thread_id（同一用户多端共享会话，则显式传同一 id）。
</details>

<details>
<summary>练习4（draw_mermaid 检查路由图）</summary>

print(graph.get_graph().draw_mermaid())

输出粘到 mermaid 渲染器（如 mermaid.live 或 IDE 预览），肉眼核对：

• START 是否连到 router；
• router 的三个分支（flight / hotel / fallback 或你的分支）各自指向正确节点；
• 每个分支节点是否最终收敛到 END；
• 有没有指向错误的边（比如 flight 连到了 hotel_node）。

技巧：get_graph().print_ascii() 在终端直接输出 ASCII 图，不用开渲染器就能快速核对；再配合 stream(stream_mode="updates") 打印实际执行路径（第 10 章练习 4），"静态图 + 动态轨迹"两边对照，排查分支错误最快。

</details>

延伸阅读

• 第 10 章：用 LangGraph 实现完整 ReAct Agent、人工介入、多 Agent
• LangGraph 官方文档（langgraph 0.4.x，接口以官方文档为准）：State、checkpointer、streaming