这是我的第487篇原创文章。

一、引言

LangGraph 不是“更复杂的 Chain”，而是把 Agent 应用拆成状态、节点、边、工具、人类反馈这些角色，让一次 AI 执行像一张可控制、可回放、可中断的流程图。

• Node (节点): 一个代表流程图中具体“步骤”的 Python 函数，是工作流中的处理单元。它接收当前的状态作为输入，执行独立的任务，然后返回一个字典来更新状态。

• Edge (边): 负责连接节点，定义了任务从一个节点流向另一个节点的路径。

• • 普通边 (Normal Edge): A → B，无条件转移，workflow.add_edge

  • 条件边 (Conditional Edge): 根据条件选择不同路径，workflow.add_conditional_edges

  • 循环边: 从 tools 回到 Agent，形成循环，workflow.add_edge("tools", "agent")

LangGraph 的核心不是“图”，而是可控的状态流转。

State 是中心，所有节点都围绕 State 读写。

Node 负责做事，Edge 负责决定下一步。

Tool 让 Agent 能连接真实系统。

Checkpointer 让 Agent 具备恢复、审计和人工介入能力。

如果只记一个类比：LangGraph 就像一个可存档的任务流程系统。State 是任务单，Node 是处理人，Edge 是流转规则，Tool 是外部系统，Checkpointer 是每一步留痕。

二、实现过程

2.1 整体设计

项目结构：

Agent.py:

Tool、State、Node、Edge

Checkpointer

流程图：

2.2 State：所有节点共享的“任务现场”

State 是 LangGraph 里最重要的概念。它不是普通参数，而是整张图的共享上下文。节点不应该靠隐式变量沟通，而应该通过 State 读写信息。

State 里通常放什么

• 对话消息messages

• 当前用户意图intent

• 工具调用结果search_results

• 中间判断need_human_review

• 最终答案final_answer

State定义：

State设计：是一个字典，两个key分别是messages和call_count。messages的value是列表，每个元素是一条message，call_count的value是int数值。

messages: 用add_messages_limit做缩减器（reducer），最多保留30条消息，防止上下文过长。

call_count：记录重试次数，防止无限循环。

message结构：每条message是一个字典，有content，type，id等key

不调用工具：一轮是2条message：用户输入（human）--AI最终返回（ai）

调用工具：一轮是4条message：用户输入（human）--AI判断决定调工具（ai）--工具返回（tool）--AI最终返回（ai）

如果是单轮(turn)调用多次工具：用户输入（human）--AI判断决定调工具（ai）--工具返回（tool）--判断决定调工具（ai）--工具返回（tool）--判断决定调工具（ai）--(...........)--AI最终返回（ai）

2.3 Checkpointer：让 Agent 不再“一跑到底”

Checkpointer 用来保存每一步 State。它带来三个能力：

• 恢复：执行中断后，可以从上次状态继续。

• 回放：可以检查 Agent 每一步为什么这么走。

• 中断：可以在关键节点暂停，等人确认。

这也是 LangGraph 相比普通调用链更适合生产环境的原因。

定义：

LangGraph的SqliteSaver自动创建两张持久化表：

checkpoints表：存的是状态快照，每次图执行到一个节点时，LangGraph把当前的完整State保存下来

• thread_id：

• thread_ts：时间戳

• checkpoint：序列化的完整状态（包括messages、call_count）

• id：checkpoint的唯一标识

writes表：存的是节点输出缓存。每个节点的返回值会暂存到这里

• thread_id：

• thread_ts：时间戳

• node_id：节点名称（包括call_tools_llm、invoke_tools）

• write：序列化的节点输出数据

2.4 Node：每一步真正干活的人

Node 是一个代表流程图中具体“步骤”的 Python 函数，是工作流中的处理单元。它接收当前的状态作为输入，执行独立的任务，然后返回一个字典来更新状态。

常见 Node 类型：

Node 类型	做什么	例子
LLM 节点	让模型推理/生成	判断用户意图、生成回复
Tool 节点	执行外部工具	搜索网页、查数据库
Router 节点	只负责判断方向	决定走人工审核还是自动回复
Formatter 节点	整理输出	把草稿变成最终答案

好的 Node 应该小而清楚：一个节点只做一类事。

LLM 节点：

tool节点：

tool节点1：

tool节点2：

这个节点不返回消息，只是做一个检测和计数。

tool节点3：

Tool：模型之外的真实能力

content_polish_submit.py: 工作清单润色提交

2.5 Edge：把节点连成可控流程

Edge 决定图怎么走。负责连接节点，定义了任务从一个节点流向另一个节点的路径。最简单的是固定边：A 执行完就去 B。更常见的是条件边：根据 State 决定下一步。

普通边 (Normal Edge): A → B，无条件转移，workflow.add_edge

条件边 (Conditional Edge): 根据条件选择不同路径，workflow.add_conditional_edges

循环边: 从 tools节点 回到 Agent（LLM节点），形成循环，workflow.add_edge("tools", "agent")

条件	下一步
模型需要外部信息	去 Tool 节点
工具结果还不够	回到 LLM 节点继续判断
涉及高风险操作	去人工确认节点
已经有答案	结束

条件边1：

关键点：判断大模型最后一条消息是否有tool_calls，有则走{'more_tools': 'invoke_tools'}执行，无则走{'finsh': 'check_final_reply'}检查是否虚假回复

条件边2：

关键点：判断大模型是否虚假回复，如果倒数第二条message是人，最后一条message存在虚假关键词），有则走{'retry': 'force_tool_retry'}执行，无则走{'success': END}。

2.6 图执行入口

invoke函数

核心是graph.invoke(initial_state, config=config)

2.7 Agent入口

引入：

执行：

2.9 附录

作者简介：

读研期间发表6篇SCI数据挖掘相关论文，现在某研究院从事数据算法相关科研工作，结合自身科研实践经历不定期分享关于Python、机器学习、深度学习、人工智能系列基础知识与应用案例。致力于只做原创，以最简单的方式理解和学习，关注我一起交流成长。需要数据集和源码的小伙伴可以关注底部公众号添加作者微信。