AI Agent 工程师 Day 22：拆解 LangGraph：Graph / State / Node / Edge

2026-10-11

AI Agent 工程师 Day 22：拆解 LangGraph：Graph / State / Node / Edge

6 个月学习计划 | M1 - LLM + Agent Runtime | Day 22/168

今日主题: LangGraph 架构分析

拆解 LangGraph：Graph / State / Node / Edge

核心结论（先给答案）：LangGraph 的本质是把 Agent 的执行过程从”一次性对话流”改造成”带状态的图计算”。Graph（图）定义控制流，State（状态）是共享内存，Node（节点）是计算单元，Edge（边）是路由决策，Checkpoint（检查点）让状态可持久化，Interrupt（中断）让人可以介入，Streaming（流式）让过程可观察。学 LangGraph 不是学 API，而是学一套 Agent Runtime 的架构范式——理解了它，你就知道自己的 Agent Runtime 为什么需要一个图，以及哪些部分可以不用图。

一、背景与动机：为什么第 22 天要拆 LangGraph

这是 6 个月 AI Agent 工程师路线的第 22 天，M1 第 4 周。前三周你已经亲手写过 Agent 的循环：LLM 调用、工具执行、结果回填、再调用。你大概率已经撞上了这些问题：

循环越写越复杂，if/else 开始失控；
Agent 跑到一半挂了，状态丢了，只能重跑；
想让人在关键步骤审批，只能靠轮询或者 hack；
想做并行工具调用，线程安全的共享状态让你头疼。

这些不是你的代码写得差，而是裸写 Agent Runtime 的结构性缺陷。LangGraph 是 LangChain 团队对这些问题给出的系统性答案，也是目前工业界 Agent 编排的事实标准之一（Anthropic 的 Claude Code 深度用户、众多生产级 Agent 的底座）。

这一天的目标很明确：拆开它，看懂它的骨架，然后把这套思想映射到你自己的 Runtime 上。不是学会 StateGraph 怎么 import，而是回答一个架构问题——“我的 Agent 需要图吗？需要哪一部分？”

二、核心内容

2.1 Graph：为什么控制流需要”图”而不是”循环”（Graph / 图）

中英文命名：Graph（图）/ Control Flow（控制流）

裸写 Agent 的典型结构是一个 while 循环：

while True:
    response = llm.invoke(messages)
    if response.tool_calls:
        results = run_tools(response.tool_calls)
        messages.append(results)
    else:
        break

这个循环的隐含假设是：Agent 只有两种状态——继续调工具，或者结束。这在单 Agent 单任务时够用，但一旦出现以下需求就会崩塌：

条件路由：根据工具结果走不同分支（比如”检索到了 → 生成答案；没检索到 → 改写查询再检索”）；
多角色协作：一个 Planner 节点 + 多个 Worker 节点 + 一个 Reviewer 节点；
循环子图：某个子任务需要”生成 → 自查 → 修正”的内部循环；
失败恢复：从任意一步重来，而不是从头跑。

这四件事的共同点是：执行路径不再是线性的，而是一个有向图。节点是处理步骤，边是”下一步去哪”的决策。用图来表达控制流，本质上是用声明式结构替代命令式分支——控制流不再是散落在代码里的 if/else，而是可以被整体审视、修改、可视化的数据结构。

这就是图的第一个工程红利：控制流成为一等公民。你可以画出来、序列化它、甚至在运行时动态修改它。

2.2 State：Agent 的共享内存（State / 状态）

中英文命名：State（状态）/ Shared Memory（共享内存）

如果 Node 是工人，State 就是他们共享的工作台。LangGraph 的核心设计决策是：状态不是散落在各节点的局部变量里，而是一个集中定义的、带类型的数据结构，通常是 TypedDict 或 Pydantic 模型：

from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class AgentState(TypedDict):
    messages: Annotated[list, add]   # reducer: 追加而非覆盖
    retrieved_docs: list[str]
    retry_count: int                 # 普通字段: 覆盖式更新

注意 Annotated[list, add] 这个细节——它引入了 Reducer（归约器） 的概念：每个节点返回的是状态的增量更新（delta），而不是完整状态。messages 字段用 add 作为 reducer，意味着节点返回的新消息会被追加进列表；而 retry_count 没有 reducer，直接覆盖。

这个设计为什么重要？因为它解决了并行节点的状态合并问题。当两个节点并行执行、都往 messages 里写东西时，LangGraph 知道用”追加”来合并结果，而不是后写覆盖先写。这就是 Superstep（超步）执行模型：同一层（superstep）内的节点并行执行，层与层之间做状态同步——思想源自 BSP（Bulk Synchronous Parallel）计算模型，和 Pregel 一脉相承。

工程直觉：State 设计就是 Agent 的数据库 schema 设计。字段越少、语义越清晰，图的行为越可预测。新手常见的错误是把一切都塞进 messages，导致状态膨胀且难以调试。

2.3 Node：最小化的计算单元（Node / 节点）

中英文命名：Node（节点）/ Compute Unit（计算单元）

Node 在 LangGraph 里就是一个函数：输入 State，输出 State 的增量。就这么简单：

def retrieve_node(state: AgentState) -> dict:
    docs = search(state["messages"][-1].content)
    return {"retrieved_docs": docs}     # 只返回增量

def agent_node(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}     # 走 add reducer, 被追加

注意 Node 里没有”下一步去哪”的逻辑——这是刻意的设计，叫做计算与路由分离（类似 TCP/IP 的分层思想）。节点只管”做什么”，边管”接下来谁做”。带来的好处：

节点可以独立测试：喂一个 State，断言输出增量，纯函数式验证；
节点可以自由重组：同一个 retrieve_node 可以被接进不同的图；
路由逻辑集中：想改流程？改边，不用改任何节点代码。

对比裸写 Runtime：你的 while 循环里，”调 LLM”、”判断分支”、”执行工具”混在一起。拆成 Node 之后，每一块都有了清晰的边界。这其实和微服务拆分、Unix 哲学（do one thing well）是同一个思想在不同抽象层的复现。

2.4 Edge：显式化的路由决策（Edge / Conditional Edge / 条件边）

中英文命名：Edge（边）/ Conditional Edge（条件边）/ Router（路由器）

边分两种：

普通边（Normal Edge��：无条件跳转，add_edge("retrieve", "generate")；
条件边（Conditional Edge）：根据当前 State 动态决定去向。

def should_continue(state: AgentState) -> str:
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "tools"
    return END

graph.add_conditional_edges("agent", should_continue, 
                            {"tools": "tools", END: END})

注意一个反直觉的点：条件边的返回值是字符串（节点名），不是 if/else 里直接执行代码。这意味着路由决策本身是一个可测试的纯函数：给定 State，输出下一个节点名。你可以在不改图结构的情况下单测所有路由分支。

工程直觉：条件边本质是把”Agent 的决策逻辑”从黑盒（LLM 自由发挥）中部分抽离出来，变成显式的、确定性的代码。LLM 负责产生”内容”（要不要调工具），条件边负责执行”规则”（调了工具就必去 tools 节点）。确定性与智能的边界，就画在 Edge 上——这是 Agent 工程里最重要的架构判断之一。

2.5 Checkpoint：让 Agent 可恢复（Checkpoint / 检查点 / 持久化）

中英文命名：Checkpoint（检查点）/ Persistence（持久化）/ Time Travel（时间旅行）

每个 Superstep 执行完，LangGraph 会把完整的 State 快照存下来（可存内存、SQLite、Postgres、Redis）。这带来的能力按价值排序：

故障恢复（Crash Recovery）：进程挂了，从最近的 checkpoint 继续跑，而不是从头再来。对跑几十分钟的长任务 Agent，这是刚需；
时间旅行（Time Travel）：加载历史任意一步的 State，从那里重跑——调试 Agent 的神器。”如果第 3 步检索换一个 query，后面会怎样？”
多轮会话记忆（Cross-thread Memory）：同一个 thread_id 的对话可以在不同进程、不同时间继续。

因果链：State 集中化 → 才能整体快照 → 才能恢��和回放。如果你的状态散落在各节点的闭包变量里，checkpoint 根本无从谈起。这再次印证了 State 设计是整个架构的地基。

2.6 Interrupt：人在环中的正确姿势（Human-in-the-loop / 人机协同 / Interrupt）

中英文命名：Interrupt（中断）/ Human-in-the-loop（人在环中）/ Human Approval（人工审批）

Agent 要上生产，绕不开一个问题：哪些动作可以自主执行，哪些需要人批准？ 比如发邮件、执行 SQL 删除、转账。

LangGraph 的方案：在图的关键节点前设置 interrupt_before=["tools"]。图跑到 tools 节点前会暂停，State 已经 checkpoint，控制权交还给人；人审批后调用 invoke(None, config) 从断点继续，拒绝则可以从上一个节点修改 State 重跑。

这个设计的巧妙之处：中断不是一个错误，而是图的一种正常执行状态。因为状态已经持久化，暂停一小时和暂停一天没有区别。对比裸写 Runtime 里”等待审批”需要自己搞队列、轮询、超时——你会体会到 checkpoint + interrupt 的组合拳价值。

2.7 Streaming：让 Agent 过程可观察（Streaming / 流式）

Agent 一跑几分钟，如果只返回最终结果，用户会以为它死了。LangGraph 的流式有几种粒度：

模式	内容	场景
`values`	每个 superstep 后的完整 State	调试、审计
`updates`	每个节点的增量输出	展示”当前进行到哪步”
`messages`	LLM token 级流式	前端打字机效果
`custom`	节点内自定义事件	进度上报（如”已抓取 3/10 个页面”）

工程直觉：Streaming 不只是 UI 体验问题，更是可观测性（Observability）的基石。updates 模式下你看到的就是 State 的变化序列——这本身就是最好的调试日志。

2.8 LangGraph 解决了什么、没解决什么（对照表）

维度	解决了 ✅	没解决 ❌
控制流	显式图结构、条件路由、循环子图	业务语义的建模（图怎么设计仍靠你）
状态	集中 State、reducer 合并、类型约束	状态设计的合理性（塞 messages 还是结构化字段，是你的决策）
可靠性	Checkpoint、故障恢复、时间旅行	业务级幂等（工具重复执行的副作用要自己处理）
协同	Interrupt、Human-in-the-loop	审批的业务规则（谁能批、超时怎么办）
可观测	多粒度 Streaming	深度 Tracing（需配 LangSmith / Langfuse）
成本	—	Token 成本控制、模型选型

一句话总结：**LangGraph 解决的是”执行基础设施”，不解决”业务智能”**。图设计得好不好，取决于你对任务的理解——这正是工程师不可替代的部分。

2.9 映射到自己的 Runtime：我为什么需要 Graph？

现在做这一天最重要的练习。假设你的 Runtime 目前是一个 while 循环，对照下表自查：

你遇到的问题	对应 LangGraph 概念	是否需要引入
控制流 if/else 超过 3 层	Graph + Conditional Edge	强烈建议
节点间共享数据靠全局变量	State + Reducer	强烈建议
挂了只能重跑	Checkpoint	长任务建议
审批靠轮询	Interrupt	生产环境建议
用户看不到进展	Streaming	用户侧产品建议

关键判断：如果你的 Agent 只是”单循环 + 两三个工具”，引入图是过度设计；但如果出现分支、并行、暂停恢复、多角色中任意两项，图的抽象就开始回本。

不想直接依赖 LangGraph？那就把它的四个概念移植进自己的 Runtime：

# 极简版思想移植: 不用框架, 用它的架构
@dataclass
class StepResult:
    state_delta: dict          # Node 输出增量 (State 思想)
    next_node: str | None      # 显式路由 (Edge 思想)

class Runtime:
    def step(self, state: dict) -> StepResult:
        ...

# 每 step 后: 持久化 state (Checkpoint 思想)
# 某些 next_node 前暂停等待审批 (Interrupt 思想)

50 行代码就能获得 LangGraph 60% 的架构收益——前提是你真的理解了 State 集中、计算路由分离、增量更新这三个决策背后的因果链。

三、ASCII 图示

                 ┌─────────────────────────────────────┐
                 │           State (共享内存)            │
                 │  messages[] / docs / retry_count    │
                 └──────△──────────────△──────────────┘
                        │ 增量写入       │ 增量写入
                 ┌──────┴─────┐  ┌─────┴──────┐
   START ──────▶ │ agent_node │  │ tools_node │
                 └──────┬─────┘  └─────△──────┘
                        │ conditional edge      │ normal edge
                        ▼ (tool_calls?)         │
                   有 → 路由到 tools ────────────┘
                   无 → END

Checkpoint: 每个箭头跨越处保存 State 快照
Interrupt:  tools_node 前可暂停等人工审批

四、实践练习与思考题

练习（必做）：拿出你前三周写的 Agent 循环代码，完成以下映射：

把循环体的每个逻辑块标注为”候选 Node”；
列出这些块之间共享的所有变量——这就是你的隐式 State，给它写一个 TypedDict；
找出所有 if/else 分支，写成条件边函数；
回答：你的 Runtime 现在需要完整 Graph 吗？还是只需要 State + 显式路由？

思考题：

两个并行节点都往同一个没有 reducer 的字段写数据，会发生什么？为什么 LangGraph 选择报错而不是静默合并？
如果把所有路由决策都交给 LLM（让模型输出下一个节点名），和用代码写条件边相比，各损失了什么？
你的业务里，哪个动作应该设 Interrupt？判断标准是什么？

五、与 Agent Engineering 的关联

在生产级 Agent 系统中，今天的知识对应四个真实工程决策：

可靠性工程：Checkpoint 是 Agent 从 Demo 走向生产的分水岭。没有持久化的 Agent 无法通过任何 SRE 审查；
合规与安全：Interrupt 机制是”Agent 自主性分级”的技术实现——高危操作人工审批，低危操作自动放行；
可观测性：State 变更流 + Streaming = Agent 的审计日志，出问题时能定位到具体节点；
团队协作：图结构让不同工程师可以并行开发不同 Node，用 State schema 作为接口契约——这就是 Agent 时代的”API 设计”。

面试高频问题”你怎么设计一个可靠的 Agent 系统”，标准答案骨架就是今天这七个概念。

六、FAQ

Q1：LangGraph 和 LangChain 是什么关系？必须一起用吗？
LangGraph 是独立的图编排框架，不依赖 LangChain。你可以用 LangGraph 编排原生 OpenAI SDK 或任何 LLM 调用。LangChain 的模型/工具抽象只是可选的胶水层。

Q2：简单的 Agent 也需要用 LangGraph 吗？
不需要。单循环、无分支、无状态恢复需求的 Agent，一个 while 循环更简单直接。图的抽象在出现分支、并行、审批、恢复需求时才开始回本。

Q3：State 越来越大怎么办？
三条路：① 把大对象（如文档全文）存外部存储，State 里只放引用 ID；② 定期在节点里做摘要压缩（Context 压缩）；③ 用 reducer 的自定义实现做去重和裁剪。

Q4：不用 LangGraph，能自己实现类似的 Runtime 吗？
完全可以，而且强烈建议至少做一次。核心就四件事：集中 State + 增量更新 + 显式路由函数 + State 快照持久化。自己实现一遍，你才真正理解框架帮你省了什么、藏了什么坑。

Q5：Conditional Edge 和让 LLM 决定路由（如 Router Agent）怎么选？
规则明确的用代码（确定性、可测试、零成本）；需要语义理解的用 LLM 路由（灵活但不确定）。成熟系统通常是混合：LLM 输出意图，代码根据意图路由。

七、参考资料

LangGraph 官方文档 — Graph API / State / Reducers 概念页：https://langchain-ai.github.io/langgraph/
LangGraph 官方教程 — Build a Basic Agent（对应本文的 agent/tools 循环图）：https://langchain-ai.github.io/langgraph/tutorials/introduction/
LangGraph 概念指南 — Persistence & Checkpointing：https://langchain-ai.github.io/langgraph/concepts/persistence/
LangGraph 概念指南 — Human-in-the-loop：https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/
LangGraph 概念指南 — Streaming：https://langchain-ai.github.io/langgraph/concepts/streaming/
LangGraph 源码中的 Pregel 执行模型（langgraph/pregel/ 目录），理解 Superstep 的最佳材料：https://github.com/langchain-ai/langgraph

下一篇预告：我们将把今天映射出的 State schema 和条件边，实际重构进自己的 Agent Runtime，并加上第一个 Checkpoint。

📋 本日知识点清单

LangGraph 的 Graph / State / Node / Edge / Checkpoint / Interrupt / Streaming
LangGraph 的思想映射到自己的 Runtime
我的 Runtime 为什么需要 Graph？
LangGraph 解决了什么问题、没解决什么
不是学 API 而是学架构思想

📝 实践练习

将 LangGraph 的思想映射到自己的 Runtime，回答”我的 Runtime 为什么需要 Graph”。

本文是「AI Agent 工程师」6 个月学习计划的一部分。学习路线：LLM → Inference → Runtime → Harness → Memory → Environment → Evaluation → Production。目标：Agent / AI Systems Engineer。