引言上一篇我们知道了 LangGraph 的世界观是把 Agent 建模成一张图。这一篇把它真正跑起来:从最小的图开始,逐步加上工具循环、检查点持久化、人在环审批、可观测性,最后得到一个完整的研究型 Agent。每一步都能独立跑通,你可以跟着敲。关于语言的说明:本系列此前代码一律用 Java,但 LangGraph 只有 Python 和 JS 版本,这里必须用 Python(核心逻辑极简,有 Java 基础完全能读懂)。如果你是 Java 团队,请回看第 07 篇第八节的建议:主体用 Spring AI,复杂编排层独立成 Python 服务。本篇的价值不在语法,而在图 状态 检查点这套思想——它在任何语言里都成立。一、为什么需要图:从 for 循环到状态机回想第一篇文章那个手写 Agent:一个 for 循环,里面两个分支。它能跑,但你想加点东西试试:任务跑到第 8 步,进程崩了——能续上吗?不能,上下文全在内存里。执行到发邮件这步,想让人先确认——能暂停吗?不能,循环没法中途挂起再恢复。用户明天回来接着聊——记得上次聊到哪吗?不记得。想调试第 3 步为什么选错工具——能回到第 3 步重跑吗?只能整个重来。这四个需求的共同点是:它们都要求执行状态可以被存下来、取出来、改一改再继续。而 for 循环的状态是隐式的、活在栈帧里的,一断就没。图(状态机)的思路是把这些显式化:for 循环版(状态隐式在内存) 图版(状态显式可存取) while True: ┌─── State(可序列化)───┐ r call_llm(msgs) │ messages / 中间结果 │ if not r.tools: break └──────────┬───────────┘ for t in r.tools: 每步读写 ▲▼ msgs.append(run(t)) [节点A] ──▶ [节点B] ──▶ ... │ 每步后自动存档 状态一断就丢 └─ Checkpointer一旦状态从栈帧里的局部变量变成一个可序列化的对象 每步存档,上面四个需求就全部自然成立。这就是 LangGraph 存在的理由——不是为了让你换个写法,而是为了让有状态这件事成为一等公民。三个核心概念再复述一遍State:贯穿全程的共享状态。所有节点读它、往里写。Node:一个处理步骤,本质就是个函数 state - 要更新的字段。Edge:节点间的流转。固定边直连;条件边根据 state 决定下一步去哪(循环和分支都靠它)。二、第一步:最小的图(两个节点)先跑通一个最简单的图,建立肌肉记忆。安装:pip install langgraph langchain-openaifrom typing import TypedDict from langgraph.graph import StateGraph, START, END # ① 定义 State:图的共享内存,用 TypedDict 声明字段 class State(TypedDict): question: str topic: str # 中间结果 answer: str # ② 定义 Node:普通函数,入参是当前 state,返回要更新的字段 def classify(state: State): q state[question] topic 技术 if any(k in q for k in [代码, 架构, Bug]) else 通用 return {topic: topic} # 只返回要改的字段,不用返回整个 state def answer(state: State): return {answer: f[{state[topic]}类问题] 回答:{state[question]}} # ③ 组装图:加节点、连边 builder StateGraph(State) builder.add_node(classify, classify) builder.add_node(answer, answer) builder.add_edge(START, classify) # 入口 builder.add_edge(classify, answer) # 固定边 builder.add_edge(answer, END) # 出口 graph builder.compile() # ④ 跑 print(graph.invoke({question: 这段代码为什么有 Bug?})) # {question: ..., topic: 技术, answer: [技术类问题] 回答:...}三个关键点先记住:节点只返回要更新的字段,不用也不该返回整个 state,LangGraph 会自动合并。这是最容易写错的地方。State 是显式的契约。什么数据在图里流动,一眼看得清——比在 for 循环里散落一堆局部变量强得多。图是先声明后编译的。compile() 之后拿到的 graph 是不可变的可执行对象,这让它能被检查、可视化、也能被持久化机制包装三、第二步:加上工具循环(条件边)from typing import Annotated from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langgraph.prebuilt import ToolNode # ① 状态里放消息列表。注意 add_messages:它让节点返回的消息追加而非覆盖 class State(TypedDict): messages: Annotated[list, add_messages] # ② 定义工具(和第 04 篇一样,docstring 就是给模型看的描述) tool def search(query: str) - str: 搜索互联网获取信息。 return f关于「{query}」的搜索结果:……(此处为模拟数据) tools [search] llm ChatOpenAI(modelgpt-4o-mini).bind_tools(tools) # ③ 两个节点:调模型 / 执行工具 def call_model(state: State): return {messages: [llm.invoke(state[messages])]} tool_node ToolNode(tools) # 官方现成的工具执行节点 # ④ 条件边的判断函数:看最后一条消息有没有工具调用 def should_continue(state: State): last state[messages][-1] return tools if last.tool_calls else END builder StateGraph(State) builder.add_node(model, call_model) builder.add_node(tools, tool_node) builder.add_edge(START, model) builder.add_conditional_edges(model, should_continue) # 分支 builder.add_edge(tools, model) # 关键:绕回去,形成循环 graph builder.compile()画出来就是这样:START ──▶ [model] ──条件边──▶ END (没有工具调用 → 结束) ▲ │ │ └──条件边──▶ [tools] (有工具调用 → 执行) └──────────────────┘ 执行完绕回 model这张图和第 一篇那个 for 循环在语义上完全等价。区别在于:循环结构现在是数据(图的定义),而不是代码(嵌套的 if/for)。正因为它是数据,才能被存档、可视化、中途插入检查——下面两节马上用到。顺带一提:add_messages 这个 Annotated 写法叫reducer——它告诉 LangGraph 这个字段该怎么合并。默认是覆盖,消息列表要的是追加。多节点并行写同一字段时,reducer 尤其重要。四、第三步:检查点(Checkpointer)—— 状态活下来了这是 LangGraph 的杀手级特性,也是它区别于自己写循环的分水岭。加它只需两行:from langgraph.checkpoint.memory import InMemorySaver checkpointer InMemorySaver() # 开发用内存;生产换 Postgres/Redis graph builder.compile(checkpointercheckpointer) # 每次调用带一个 thread_id —— 相当于会话 ID config {configurable: {thread_id: user-42}} graph.invoke({messages: [(user, 帮我查一下 LangGraph 是什么)]}, config) # 稍后(哪怕重启了进程),同一个 thread_id 继续对话,它记得前面聊过什么 graph.invoke({messages: [(user, 那它和 LangChain 什么关系?)]}, config)就这么简单,你白得了四样东西:多轮记忆。同一个 thread_id 自动带上历史,不用自己拼消息数组。换个 thread_id 就是新会话,天然多用户隔离。崩溃恢复。每步执行后状态已落盘,进程挂了重启,用同一个 thread_id 就能从断点继续。时间旅行调试。可以列出历史检查点、回到任意一步、改掉状态再往下跑:# 查看历史(每步一个快照) for snap in graph.get_state_history(config): print(snap.config[configurable][checkpoint_id], snap.next) # 回到某个检查点重跑(可用于如果当时选了另一个工具会怎样) graph.invoke(None, {configurable: {thread_id: user-42, checkpoint_id: 某个快照 id}})人在环的基础。下一节全靠它。生产提醒:InMemorySaver 只用于开发。生产要用持久化实现(如 langgraph-checkpoint-postgres),否则重启即失忆。另外检查点会随对话增长,长会话要配合消息裁剪/摘要,否则状态越存越大。五、第四步:人在环(中断与恢复)真实业务里,有些动作不能让模型自己拍板:发对外邮件、改数据库、退款。我们希望执行到这一步时暂停,等人确认后再继续。有了检查点,这件事变得非常自然——因为暂停不过是存档后停下,而恢复就是读档后继续。from langgraph.types import interrupt, Command def send_email(state: State): draft state[draft] # 关键:interrupt 会在此处中断执行,并把 draft 抛给外部等待人工判断 decision interrupt({draft: draft, question: 这封邮件可以发吗?}) if decision approve: return {result: f已发送:{draft}} return {result: 已取消发送} builder.add_node(send_email, send_email) graph builder.compile(checkpointercheckpointer) # 人在环必须有 checkpointer # ① 第一次跑:执行到 interrupt 就停住,状态已存档 result graph.invoke({draft: 尊敬的客户,您的订单……}, config) print(result[__interrupt__]) # 拿到待确认的内容,展示给用户 # ② 人看完点了同意 —— 用 Command(resume...) 把答案送回去,从断点继续 final graph.invoke(Command(resumeapprove), config) print(final[result]) # 已发送:……[起草邮件] ──▶ [send_email] ──✋ interrupt ──▶ (进程可以退出!) 状态已存档 │ 人工在网页/IM 上点同意 ────────┘ ▼ Command(resumeapprove) ──▶ 从断点继续 ──▶ [已发送]请注意中间那句进程可以退出——这是它和阻塞等一个输入的本质区别。中断期间不占任何资源,人可能三分钟后确认,也可能明天才确认,甚至由另一个进程/服务来恢复。这是第 19 篇讲高风险操作拦截时最实用的落地手段。除了 interrupt,还有一种更简单的方式:编译时用 interrupt_before[send_email] 声明进这个节点前先停,适合只需要审一下而不需要传回复杂决策的场景。六、完整案例:一个可观测的研究型 Agent把前面所有部件拼起来,做一个真正有用的东西:给定一个主题,自主检索资料 → 整理成研究报告 → 交人审阅 → 定稿。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import interrupt, Command from langchain_openai import ChatOpenAI from langchain_core.tools import tool # ---------- State:显式声明图里流动的所有数据 ---------- class State(TypedDict): topic: str messages: Annotated[list, add_messages] # 研究过程的对话工具结果 report: str # 产出的报告 review: str # 人工意见 # ---------- 工具 ---------- tool def web_search(query: str) - str: 检索互联网资料,返回摘要片段。 return f[资料] 关于 {query}:市场规模持续增长,主要玩家有 A/B/C…… tools [web_search] llm ChatOpenAI(modelgpt-4o-mini, temperature0).bind_tools(tools) # ---------- 节点 ---------- def research(state: State): 反复检索直到模型认为资料够了(靠条件边形成循环) msgs state[messages] or [ (system, 你是研究员。用 web_search 收集资料,资料充分后停止调用工具。), (user, f研究主题:{state[topic]})] return {messages: [llm.invoke(msgs)]} def write_report(state: State): 把检索到的材料写成报告 prompt state[messages] [(user, 基于以上资料,写一份简明研究报告。)] return {report: ChatOpenAI(modelgpt-4o-mini).invoke(prompt).content} def human_review(state: State): 人在环:报告要人过一眼 decision interrupt({report: state[report], question: 报告是否通过?}) return {review: decision} def revise(state: State): 按意见修改(Evaluator-Optimizer 的味道,见第 03 篇下) prompt [(user, f按此意见修改报告。\n原报告:{state[report]}\n意见:{state[review]})] return {report: ChatOpenAI(modelgpt-4o-mini).invoke(prompt).content} # ---------- 条件边 ---------- def research_done(state: State): return tools if state[messages][-1].tool_calls else write_report def review_result(state: State): return END if state[review] approve else revise # ---------- 组装 ---------- b StateGraph(State) b.add_node(research, research) b.add_node(tools, ToolNode(tools)) b.add_node(write_report, write_report) b.add_node(human_review, human_review) b.add_node(revise, revise) b.add_edge(START, research) b.add_conditional_edges(research, research_done) # 检索循环 b.add_edge(tools, research) b.add_edge(write_report, human_review) b.add_conditional_edges(human_review, review_result) # 通过 or 返工 b.add_edge(revise, human_review) # 改完再审 graph b.compile(checkpointerInMemorySaver())整张图长这样,五种能力一目了然:START ──▶ [research] ──▶ [write_report] ──▶ [human_review] ──▶ END ▲ │ ▲ │ │ ▼ │ ▼ └[tools] ← 工具循环 [revise] ← 返工循环 ✋ 中断点在 human_review跑起来:config {configurable: {thread_id: research-001}} # 流式执行,能看到每个节点的产出(这就是最基础的可观测) for chunk in graph.stream({topic: 固态电池行业现状}, config): print(chunk) # 每个节点执行完都会 yield 一次 # 停在 human_review。人看完提意见: for chunk in graph.stream(Command(resume补充一段风险分析), config): print(chunk) # 走 revise → 再回 human_review # 这次通过: graph.invoke(Command(resumeapprove), config)关于可观测性,三个层次由浅入深:用 stream() 看每步产出——零成本,开发期最常用。也可以用 stream_modedebug 看更细的节点进出。用 get_state_history() 复盘——出问题后回看每一步的完整状态快照,比翻日志强得多。接 LangSmith 做全链路 trace——生产必备。设两个环境变量即可,不用改代码:export LANGSMITH_TRACINGtrue export LANGSMITH_API_KEYyour-key之后每次运行的每个节点、每次 LLM 调用的 prompt/输出/token/耗时,都会自动上报,可以在界面上一层层点开看。后面文章会讲可观测性和 OpenTelemetry 的对接,这里先知道接一下就有。七、小结这一篇我们把图与状态机真正跑通了。回看这条路径:从两个节点的直线图(建立 State/Node/Edge 肌肉记忆)→ 加条件边形成工具循环(等价于第 一 篇的 for 循环,但循环变成了数据)→ 加Checkpointer(白得多轮记忆、崩溃恢复、时间旅行)→ 加interrupt 人在环(暂停期间进程可退出,这是重点)→ 拼成一个带审阅返工的研究型 Agent,并接上流式与 trace。如果只能记一句话:LangGraph 的全部价值在于把执行状态显式化并每步存档。一旦状态可存取,多轮记忆、断点续跑、人工介入、可重放调试就不再是需要你苦苦造轮子的难题,而是自然的推论。