生命不息，折腾不止。上一篇把框架摆上台比完了，结论是「要上生产选 LangGraph」——那这篇就真开干，从 State / Node / Edge 三个词开始，带你把第一张 Agent 图画出来。

上一篇的选型表里我说过，LangGraph 是生产队的「耐久」担当，核心卖点两个：断点续跑和人审门禁。但它的代价也写得明明白白——概念重，你得先接受 state / node / edge 这套心智模型。

今天就专门把这层窗户纸捅破。不堆术语，就用一个能跑起来的例子，带你干四件事：看懂 State / Node / Edge 这仨词；画一张没工具的「空图」；给它接上一个会调工具的 Agent；最后加上断点续跑和人审门。全程代码你复制下来就能跑。

（先声明一下版本：LangGraph 在 2025 年 10 月发了 1.0，现在最新是 1.1.x。网上 2024 年的老教程里 MemorySaver、ToolExecutor 这些写法在 1.x 已经改名/删了，照着抄会报错。这篇按 1.x 的新 API 写，具体命令以你 pip 装到的官方版本为准。）

一、先懂三个词：State / Node / Edge

LangGraph 的骨架就一句话：你的 Agent 是一张有向图，数据在图里流转，每个节点改一点数据，边决定下一步去哪。

  • State（状态）：一张贯穿全图的数据表，是所有节点共享、都能读写的「公共黑板」。它用 TypedDict 定义，字段你自己定。
  • Node（节点）：就是一个个 Python 函数。每个函数吃进当前的 State，返回一个「要改哪些字段」的小字典。
  • Edge（边）：节点之间的箭头，决定谁跑完接着谁跑。普通边是死路线（A 完必去 B），条件边是活路线（看 State 内容决定去哪，这俩是 Agent 能「循环」的关键）。

一句话记住：State 是数据，Node 是动作，Edge 是顺序。 往下每个例子都拿这句话套，就不会晕。

二、搭环境：从零装出能跑的 LangGraph

先建个干净的虚拟环境（别直接怼在系统 Python 上）：

1
2
3
4
mkdir langgraph-first && cd langgraph-first
python -m venv .venv
source .venv/bin/activate          # Windows 用 .venv\Scripts\activate
pip install langgraph langchain-openai

说明几点：

  • 核心就是 langgraph 一个包；langchain-openai 是给 ChatOpenAI 用的模型接入（想接 Claude 换成 langchain-anthropic）。
  • 要持久化记忆/断点，后面会加 langgraph-checkpoint-sqlite，现在不急。
  • 强烈建议把版本 pin 住，老教程踩爆的坑就是 langgraph 和 langchain-core 版本打架，报 ImportError: cannot import name 'BaseMessage'。装完验一下：
1
2
python -c "import langgraph; print(langgraph.__version__)"
# 应当打印 1.1.x

三、最小骨架：先画一张没有工具的「空图」

先别上模型，用两个纯 Python 函数画一张最朴素的图，把 State / Node / Edge 看个分明。新建 01_bare.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# State：整张图共享的数据，就一个整数 count
class State(TypedDict):
    count: int

# Node 1：拿到 state，把 count 加一
def add_one(state: State):
    return {"count": state["count"] + 1}

# Node 2：把 count 翻倍
def double_it(state: State):
    return {"count": state["count"] * 2}

# 画图
builder = StateGraph(State)
builder.add_node("add_one", add_one)
builder.add_node("double_it", double_it)

# 边：START -> add_one -> double_it -> END
builder.add_edge(START, "add_one")
builder.add_edge("add_one", "double_it")
builder.add_edge("double_it", END)

graph = builder.compile()

# 跑图，count 从 3 出发
result = graph.invoke({"count": 3})
print(result)   # -> {'count': 8}

跑一下，输出 {'count': 8}。这 20 行把三样东西全讲透了：

  • State 就是那个 count，从 3 出发，一路被改。
  • add_one、double_it 是两个 Node，各拿 state、各回一个 {"count": ...} 的更新字典。
  • 三条 add_edge 是 Edge，把路径钉死成 START → add_one → double_it → END。

注意两个哨兵：START 和 END 是 1.x 里明确的「图开头/图结尾」，新 API 就是靠 add_edge(START, "xx") 标入口、add_edge("xx", END) 标出口，不再有 set_entry_point 那套老写法。

这就是 LangGraph 的「图」的全部秘密——你后面写的所有复杂 Agent，本质都是在这张图上加节点、加条件边、让状态循环起来而已。

四、给它接上脑子：一个会调工具的 Agent

空图跑通了，现在把 LLM 塞进去当「脑子」，再给它一个工具，让它学会「想一步、动一步、观察结果、再想」。这是 Agent 和 Chatbot 的分水岭。

先定义工具，用 @tool 装饰器：

1
2
3
4
5
6
7
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询某个城市的天气（示例，返回写死的数据）。"""
    fake = {"北京": "晴，18°C", "上海": "多云，22°C", "深圳": "雷阵雨，28°C"}
    return fake.get(city, f"没查到 {city}，就当今天晴天吧")

最省事的写法——一条 create_react_agent 把整套循环给你搭好：

1
2
3
4
5
6
7
8
9
10
11
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

agent = create_react_agent(model, tools=[get_weather])

result = agent.invoke(
    {"messages": [("user", "北京和上海现在天气怎么样？")]}
)
print(result["messages"][-1].content)

内部它自动跑了这个循环：模型读懂问题 → 决定调 get_weather('北京') 和 get_weather('上海') → 工具返回结果 → 模型把两个结果揉成一句话输给你。你只是 invoke 了一次，背后 ReAct（推理+行动）那套「想→做→看→再想」全是它替你跑完的。

想看清它内部长啥样，把这张图手动拼一遍（这就是 create_react_agent 的底裤）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

class State(TypedDict):
    messages: Annotated[list, add_messages]  # 消息用 add_messages 累加，别覆盖

def call_model(state: State):
    # bind_tools 让模型能「吐出工具调用」
    return {"messages": [model.bind_tools([get_weather]).invoke(state["messages"])]}

builder = StateGraph(State)
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode([get_weather]))   # ToolNode 负责真去执行工具

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)  # 有 tool call 就去 tools，没有就 END
builder.add_edge("tools", "agent")                      # 工具跑完，回到 agent 继续想

graph = builder.compile()

关键就两点：add_conditional_edges("agent", tools_condition) 是条条件边——模型这一步要是发了工具调用就走 tools，没发就直接 END，这一进一出就是 Agent 能「循环」的根；而 tools → agent 那条普通边，让工具结果回流给模型继续推理。所谓 Agent 的思考闭环，就是这俩边的组合。

（一个避坑：bind_tools 用的 tools 列表和 ToolNode 里传的必须是同一份，两边对不上就会「模型叫了工具，结果没人执行」。）

到这一步，你已经有了一只会调工具的真 Agent。但还有俩硬伤：关了就忘、它想干嘛就干嘛没人拦。下面两节分别把这两个洞补上——这也是 LangGraph 区别于别的框架的看家本领。

五、让它记得住：断点续跑（checkpoint + thread_id）

普通 invoke 是「无状态」的，每次跑完图就散。LangGraph 的解法是checkpoint（检查点）：每跑完一个节点，就存一份当前 state 的快照。配上 thread_id，就同时解锁了两件大事——短期记忆和断点续跑。

先来个最轻的内存版：

1
2
3
4
5
6
7
8
from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "1"}}   # thread_id 是找回状态的钥匙

graph.invoke({"messages": [("user", "我叫吴峰，记住了")]}, config=config)
res = graph.invoke({"messages": [("user", "我叫什么名字？")]}, config=config)
print(res["messages"][-1].content)   # 它能答出「吴峰」，因为同一 thread_id 读到了之前的快照

记住这条铁律：thread_id 就是你的指针。同一个 thread_id 下次 invoke 接着上次的状态跑；换个新的 thread_id 就是开一个全新的对话，互不干扰。这也是「跑了仨小时、第 17 步崩掉」能从第 17 步接着跑而不是从头来过的底层原理。

但 InMemorySaver 存在内存里，进程一重启就没了，只适合调试。生产上换成 SQLite 或 Postgres：

1
pip install langgraph-checkpoint-sqlite
1
2
3
4
5
6
from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)
    # 后面照常用 config={"configurable": {"thread_id": "..."}} 跑

多机多副本、要高可用，就上 Postgres（langgraph-checkpoint-postgres），套路一样，换个 Saver 而已。

六、加一道人审门：interrupt 暂停等你拍板

断点能续了，还差一个「该不该走下一步我说了算」。LangGraph 里这叫 human-in-the-loop（人审），核心就一个函数 interrupt()：节点跑到它，图整个暂停，把一个问题抛给你，等你拍板后再 resume 接着跑。

先看「发消息前先确认」这个经典场景：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

def send_with_approval(state: State):
    # 暂停，把草稿抛给人类确认；返回值就是人回传的决定的
    approved = interrupt({"question": "确认发送这条消息？", "draft": state["draft"]})
    return {"status": "已发送" if approved else "已取消"}

# 注意：用 interrupt 必须先挂 checkpointer，并传 thread_id，否则直接崩
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "t-1"}}

# 第一次 invoke：跑到 interrupt 停下来，把问题交给你
res = graph.invoke({"draft": "各位，明早十点开会"}, config=config)
print(res["__interrupt__"])   # -> [{'question': '确认发送这条消息？', 'draft': '...'}]

# 你看完点「确认」，用 Command(resume=...) 沿原 thread 接着跑
res2 = graph.invoke(Command(resume=True), config=config)
print(res2["status"])          # -> 已发送

理解它只记三点：

  1. interrupt() 会暂停整张图，返回的 res["__interrupt__"] 里就是刚才抛给你的问题。
  2. 恢复靠 Command(resume=值)，这个 值 会变回 interrupt() 的返回值，节点从拿到的那个值接着往下跑。
  3. 必须同一 thread_id 恢复——它得靠 checkpointer 找得到上次停在哪。

除了这种「动态」的 interrupt()，还有种「静态」写法更省事，想在某节点之前/之后统一停下，编译时标一句就行：

1
2
3
4
graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["tools"],   # 每次真要调工具前，先停一下让人过目
)

把 interrupt_before=["tools"] 加上，你的 Agent 就成了「工具调用前先打报告」的乖孩子——发邮件、改数据库这种带副作用的操作，点上这一句就多了一道保险。

（一个必须知道的坑：interrupt() 之前的副作用代码在 resume 时会从头重跑一遍，因为节点是从开头重执行的。所以 interrupt() 前面只放幂等的操作，真正「动手」的副作用放到 interrupt() 之后，或者单独拆一个节点。）

七、踩坑清单：替我趟过的这几个

  1. 老教程照抄报 ImportError：MemorySaver → 1.x 改叫 InMemorySaver，ToolExecutor → 已删，换成 ToolNode。看到 set_entry_point 也是老写法，现在用 add_edge(START, "xx")。
  2. 版本打架：langgraph 和 langchain-core 版本不匹配报 cannot import name 'BaseMessage'。装的时候把版本 pin 住，别裸 pip install langgraph langchain。
  3. bind_tools 和 ToolNode 的 tools 列表不一致：模型叫了工具没人执行，先查这个。
  4. 忘了传 thread_id 或用错值：加了 checkpointer 也没记忆、断点续不上，多半是 config 里的 thread_id 没传或变了。
  5. 用 interrupt() 却没挂 checkpointer：直接崩，这是硬前提。
  6. interrupt() 前放有副作用的操作：resume 会重跑，发邮件会发两遍。

到这儿，你的第一张 LangGraph 图就跑完整了：能引状态流转，能接模型调工具，能记住上下文、断了能续，还能在关键处停下来等人拍板。把这四个能力装进一张图里，你就已经摸到了「生产级 Agent」的门槛。

生命不息，折腾不止。下一篇咱们往上叠一层——LangGraph 实战第二课：给 Agent 加「短期记忆 + 长期记忆」，让它真正记得住人和事。这篇的 checkpoint 只够记住一次会话，下一篇用 InMemoryStore 跨会话存长期记忆、再教它把该记的主动写进知识库，顺便把多轮的 thread_id 会话管理捋顺。先把今天这张图跑通、那行 interrupt_before=["tools"] 亲手加上试一把，咱们下一篇见。