游乐游手机版
首页/AI教程/文章详情

LangGraph原理深度解析:构建Agent最适合的框架

时间:2026-06-11 17:06
LangGraph将Agent运行过程抽象为有向图,通过State、Node、Edge和Checkpoint等核心概念实现复杂控制流。State作为全局不可变数据,使用Reducer处理并发更新;Node是无副作用的计算单元;Edge显式定义流转,条件边实现分支与循环。Checkpoint持久化每步快照,支持多轮对话与断点恢复,并提供人工中断原语。该框架不预

LangGraph 原理深度解析:为什么它是目前最适合构建 Agent 的框架

说个具体场景。假设你要搭建一个 AI Agent,它需要: - 接收用户问题 - 判断是否需要查资料 - 如果需要,调用搜索工具 - 把搜索结果交给 LLM 继续推理 - 如果还不够,继续搜索;否则输出答案 LangGraph 原理深度解析:为什么它是目前最适合构建 Agent 的框架 最朴素的写法就是一堆嵌套的 `if/else` 加 `while` 循环。可一旦逻辑稍微复杂一点,代码就很容易腐烂——状态传来传去、循环条件难以维护、想加一个步骤要改好几个地方。 LangGraph 解决的正是这个难题。它将“Agent 的运行过程”抽象成一张有向图,用图的语言来描述流程,让复杂的控制逻辑变得清晰、可维护、可扩展。 ### 核心抽象:图、状态、节点、边 LangGraph 的整个设计围绕四个核心概念展开。 **State(状态)** State 是贯穿整个图的共享数据结构,用 Python 的 `TypedDict` 定义: ```python from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] # 对话历史 current_plan: str # 当前计划 iteration_count: int # 已迭代次数 ``` State 有几个关键特性: 首先是不可变更新——节点函数不直接修改 state,而是返回一个“diff”(增量),LangGraph 负责把这个 diff 合并进当前 state。这个思路与 Redux 一致,能让状态变更变得可追溯。 其次是 Reducer 机制:`Annotated[list, operator.add]` 告诉 LangGraph,当多个节点同时更新 `messages` 字段时,用 `operator.add`(列表追加)而不是覆盖。这是处理并发更新冲突的方式——每个字段都可以有自己的合并策略。 ```python # 节点只需返回要更新的字段,不需要返回完整 state def my_node(state: AgentState) -> dict: return {"messages": [new_message]} # LangGraph 负责 merge ``` **Node(节点)** 节点是图中的计算单元,实质上就是一个 Python 函数: ```python def call_llm(state: AgentState) -> dict: response = model.invoke(state["messages"]) return {"messages": [response]} ``` 节点可以是任何操作:LLM 调用、工具执行、数据转换、外部 API 请求。LangGraph 对节点内部的实现没有任何约束。 LangGraph 还内置了一些常用节点,最常用的是 `ToolNode`,它会自动解析 LLM 输出的 `tool_calls`,执行对应的工具,然后把结果包装成 `ToolMessage` 写回 state: ```python from langgraph.prebuilt import ToolNode tool_node = ToolNode(tools=[search, calculator]) ``` **Edge(边)** 边定义了节点之间的流转关系,分两种: 固定边:A 执行完永远去 B。 ```python builder.add_edge("tools", "agent") ``` 条件边:根据当前 state 决定下一步去哪里,这是 LangGraph 实现分支和循环的关键。 ```python def should_continue(state: AgentState) -> str: last_msg = state["messages"][-1] if last_msg.tool_calls: return "tools" # 去执行工具 return END # 结束 builder.add_conditional_edges("agent", should_continue) ``` `should_continue` 函数接收当前 state,返回一个字符串,LangGraph 根据这个字符串决定下一个节点。这个函数就是整个 Agent 的“大脑决策”——完全由你控制,不依赖任何框架的黑魔法。 **Graph(图)** 图是上述三个元素的容器,负责编排和执行: ```python from langgraph.graph import StateGraph, END builder = StateGraph(AgentState) builder.add_node("agent", call_llm) builder.add_node("tools", tool_node) builder.set_entry_point("agent") builder.add_conditional_edges("agent", should_continue) builder.add_edge("tools", "agent") graph = builder.compile() ``` `compile()` 做的事情包括:验证图的合法性(有没有死路、节点是否都定义)、生成执行计划、挂载 checkpointer。编译后的 `graph` 对象是不可变的,可以在多线程环境下安全复用。 ### 执行机制:图是怎么跑起来的 调用 `graph.invoke(input)` 后,LangGraph 的执行流程如下: 1. 初始化 state:将 input 合并进初始 state 2. 从 entry_point 节点开始 3. 执行当前节点函数,拿到返回的 diff 4. 将 diff 通过 reducer 合并进 state(生成新 state 快照) 5. 根据出边(固定边或条件边)确定下一个节点 6. 重复步骤 3-5,直到到达 END 7. 返回最终 state 每一步执行后,LangGraph 都会生成一个新的 state 快照。如果配置了 checkpointer,这个快照会被持久化——这是多轮对话和断点恢复的基础。 这个执行模型本质上是一个事件循环,每次循环处理一个节点。循环本身(即 Agent 的“思考-行动”循环)是通过图的结构来表达的,而不是通过代码里的 `while True`。 ### Checkpoint:状态持久化的原理 Checkpoint 是 LangGraph 最被低估的特性之一。它的核心思想非常简单:把每一步执行后的完整 state 快照存储下来。 ```python from langgraph.checkpoint.memory import MemorySa ver memory = MemorySa ver() graph = builder.compile(checkpointer=memory) config = {"configurable": {"thread_id": "session_001"}} graph.invoke({"messages": [HumanMessage("你好")]}, config=config) ``` `thread_id` 是会话的唯一标识。每次 `invoke` 时,LangGraph 会: 1. 用 `thread_id` 从 checkpointer 里查找历史快照 2. 如果有,恢复到上次结束时的 state 3. 执行完成后,把新的 state 快照存进 checkpointer 这意味着多轮对话的记忆不是靠手动维护消息列表实现的,而是 checkpointer 自动帮你完成。换成 SQLite 做持久化,进程重启后对话历史依然还在: ```python from langgraph.checkpoint.sqlite import SqliteSa ver memory = SqliteSa ver.from_conn_string("./agent.db") ``` Checkpoint 还支持时间旅行——你可以获取任意历史快照,从那个时间点重新执行,这对调试复杂 Agent 非常有用。 ### Human-in-the-Loop:中断与恢复 LangGraph 原生支持在指定节点前暂停,等待人工干预后再继续: ```python graph = builder.compile( checkpointer=memory, interrupt_before=["execute_action"] # 执行这个节点前暂停 ) # 第一次调用:执行到 execute_action 前暂停 graph.invoke(input, config=config) # 人工审核,确认后继续执行 graph.invoke(None, config=config) # 传 None 表示从暂停点继续 ``` 实现原理是:到达中断节点时,LangGraph 保存当前 state 快照,然后抛出一个特殊的中断信号。下次 `invoke` 时,从快照恢复,跳过已执行的节点,从中断点继续。这个机制对需要人工审核的高风险操作场景——比如执行数据库删除、发送邮件——非常关键。 ### 与其他框架的本质区别 很多人问:LangGraph 和 LangChain 的 `AgentExecutor`、AutoGen 有什么本质区别? `AgentExecutor` 是一个封装好的黑盒,它内置了 ReAct 的执行逻辑,你只需要提供工具列表。好处是上手快,坏处是控制权在框架手里——想改循环逻辑、加条件分支、自定义状态,都很别扭。 `AutoGen` 的核心抽象是“多个 Agent 互相对话”,适合角色分工明确的多 Agent 协作场景,但对单个 Agent 内部的控制流支持有限。 而 LangGraph 的定位是控制流引擎——它不预置 Agent 的默认行为,只提供图的描述能力和执行机制。你来定义流程,它负责执行。这种设计在简单场景下比 `AgentExecutor` 啰嗦,但在复杂场景下优势明显:任何你能想到的流程,都可以用图来描述。 ### 总结 LangGraph 的设计哲学可以用一句话概括:将 Agent 的控制流从代码中解放出来,用图来表达复杂流程。 | 概念 | 作用 | |------|------| | State | 全局共享的不可变数据,reducer 处理并发更新 | | Node | 无副作用的计算单元,只负责处理和返回 diff | | Edge | 控制流的显式描述,条件边实现分支和循环 | | Checkpoint | 每步快照持久化,支持多轮对话与断点恢复 | | Interrupt | 人工介入的原语,高风险操作不可或缺 | 理解了这五个概念,LangGraph 就没有黑魔法了——它就是一个带状态管理的图执行引擎,只不过这个图恰好特别适合描述 Agent 的行为。 **参考资料** - LangGraph 官方文档 - LangGraph 源码 - LangGraph Academy
来源:https://juejin.cn/post/7649206162341478450
上一篇苹果WWDC 2026补齐AI最后短板 下一篇阿里云万小智灵感值用途与收费标准
本站内容用于信息整理与展示,如有侵权或内容问题请及时联系处理。

相关推荐

补充同频道和同主题内容,方便继续浏览更多相关内容。

同类最新

继续查看同栏目最近更新的文章。

更多
CapCut AI Docker 一键部署:镜像拉取、端口映射与数据目录配置教程
AI教程 · 2026-06-30

CapCut AI Docker 一键部署:镜像拉取、端口映射与数据目录配置教程

CapCutAI容器化部署需先确认镜像来源与授权范围,再完成环境准备、镜像拉取、端口映射、数据目录挂载和启动验证,适合本地试用、团队内网演示与轻量化AI剪辑服务管理。

CapCut AI Windows本地安装配置2026最新版含下载与环境要求
AI教程 · 2026-06-30

CapCut AI Windows本地安装配置2026最新版含下载与环境要求

CapCutAI与剪映AI在Windows端适合短视频、口播、课程和营销素材剪辑,安装前需确认系统、显卡、存储与网络条件,优先选择官方渠道下载,并完成账号、素材目录、硬件加速和导出参数配置。

Veo新手保姆级安装教程:从下载到首次运行
AI教程 · 2026-06-30

Veo新手保姆级安装教程:从下载到首次运行

Veo适合用文字生成短视频,新手应先确认官方入口、准备账号与设备环境,再按网页或应用方式完成启用。首次运行重点在提示词、参数、素材合规与结果保存,避免使用非官方安装包。

Veo本地模型运行下载路径设置与性能优化指南
AI教程 · 2026-06-30

Veo本地模型运行下载路径设置与性能优化指南

Veo本地模型部署需先确认模型来源与硬件条件,再完成下载校验、目录规划、路径配置和推理参数优化。重点关注显存占用、依赖版本、缓存位置、授权范围与常见报错处理。

Veo安装失败解决指南:常见报错与日志排查及升级回滚方案
AI教程 · 2026-06-30

Veo安装失败解决指南:常见报错与日志排查及升级回滚方案

Veo安装失败通常与系统环境、依赖版本、网络源、权限和缓存有关。排查时应先确认版本要求,再查看安装日志,按报错类型处理,并提前备份项目,确保升级与回滚可控。