文章目录LangGraph最新版技术解析有状态多智能体图执行引擎的架构原理与工程实践一、引言二、核心概念图执行模型全景2.1 从 AgentExecutor 到 StateGraph2.2 三元素State、Node、Edge三、图的构建与编译3.1 完整构建流程3.2 编译产物的能力矩阵四、持久化与检查点Checkpointer4.1 Checkpoint 机制4.2 三种 Checkpointer 实现五、Human-in-the-loop可控智能体的工程基础5.1 interrupt_before / interrupt_after5.2 动态断点Dynamic Interrupt六、多智能体模式6.1 Supervisor 模式6.2 子图Subgraph6.3 网状多智能体Swarm七、流式输出细粒度事件系统7.1 三种流式模式对比7.2 astream_events 事件类型八、LangGraph Platform从本地到生产九、与竞品横向对比十、总结LangGraph最新版技术解析有状态多智能体图执行引擎的架构原理与工程实践一、引言亲爱的朋友们创作不容易若对您有帮助的话请点赞收藏加关注哦您的关注是我持续创作的动力谢谢大家有问题请私信或联系邮箱jasonai.fngmail.com2024 年初当绝大多数 Agent 框架还在用while True循环包裹 ReAct 时LangChain 团队悄然发布了 LangGraph。它的核心主张只有一句话用有向图替代顺序循环把控制流还给工程师。不同于 AutoGPT 式的让 LLM 自由规划一切LangGraph 的设计哲学更接近状态机你声明节点、定义转移条件、设定持久化策略LLM 只负责在每个节点做推理而不是决定整个程序的走向。这一取舍让 LangGraph 在生产可靠性上远超同期竞品也因此在短短一年内从 LangChain 的附属模块升格为独立的核心产品线。本文以 LangGraphv0.2.x2025 Q2 最新稳定版为基准覆盖图执行模型、状态管理、持久化与检查点、Human-in-the-loop、多智能体模式、流式输出与 LangGraph Platform 七大核心模块给出可落地的工程解析。二、核心概念图执行模型全景2.1 从 AgentExecutor 到 StateGraphLangChain 旧版AgentExecutor的本质是一个死循环控制流完全托付给 LLM 的输出格式# AgentExecutor 伪代码 —— 一切都是隐式的whileTrue:actionllm.predict(messages)# LLM 决定下一步ifactionFinal Answer:breakresulttool_executor.run(action)# 执行工具messages.append(result)# 追加结果这个结构有三个致命弱点无法中断恢复、无法条件分支、无法并发执行。LangGraph 用显式图结构解决了这三个问题┌─────────────────────────────────────────────────────────────┐ │ StateGraph 全景 │ │ │ │ ┌──────────┐ 条件边路由函数 ┌──────────┐ │ │ │ agent │ ──────────────────► │ tools │ │ │ │ node │ │ node │ │ │ └──────────┘ ◄────────────────── └──────────┘ │ │ │ 固定边回流 │ │ │ 条件边 │ │ ▼ │ │ END / 其他节点 │ │ │ │ StateTypedDict在节点间流转Checkpoint 在每步后持久化 │ └─────────────────────────────────────────────────────────────┘2.2 三元素State、Node、EdgeLangGraph 的所有构建都围绕三个基本概念展开概念类型职责StateTypedDict Reducer贯穿全图的共享数据容器节点读写它NodeCallable[[State], dict]执行单元接收 State 返回局部更新Edge固定边 / 条件边声明节点间的转移逻辑Reducer是 State 的核心设计——它规定了当多个节点并发写同一个字段时如何合并fromtypingimportTypedDict,AnnotatedimportoperatorclassAgentState(TypedDict):# operator.add 作为 reducer新值追加而不是覆盖messages:Annotated[list,operator.add]# 无 reducer 注解新值直接覆盖current_plan:str# 自定义 reducer保留最大值best_score:Annotated[float,max]三、图的构建与编译3.1 完整构建流程fromlanggraph.graphimportStateGraph,ENDfromlanggraph.prebuiltimportToolNode,tools_condition# ① 定义 StateclassMyState(TypedDict):messages:Annotated[list,operator.add]user_id:str# ② 定义 Node 函数defcall_model(state:MyState)-dict:Agent 节点调用 LLM 决策responsellm_with_tools.invoke(state[messages])return{messages:[response]}# ③ 构建图builderStateGraph(MyState)builder.add_node(agent,call_model)builder.add_node(tools,ToolNode(tools))# 预制工具节点# ④ 设置入口builder.set_entry_point(agent)# ⑤ 添加条件边路由函数builder.add_conditional_edges(agent,tools_condition,# 预制有 tool_calls → tools否则 → END)# ⑥ 添加固定边builder.add_edge(tools,agent)# ⑦ 编译可附加 checkpointer / storegraphbuilder.compile(checkpointerMemorySaver(),# 内存 Checkpointinterrupt_before[tools],# 工具执行前暂停)3.2 编译产物的能力矩阵compile()返回一个CompiledGraph它实现了与 LCEL 相同的Runnable接口方法行为graph.invoke(input, config)同步运行至终止节点或中断点graph.stream(input, config)逐节点输出 State 快照graph.astream_events(input, config, versionv2)异步输出细粒度事件含 LLM token 流graph.get_state(config)读取当前 Checkpoint 状态graph.update_state(config, values)外部写入状态Human-in-loop 核心四、持久化与检查点Checkpointer4.1 Checkpoint 机制LangGraph 在每个节点执行完成后自动保存一次 Checkpoint包含完整 State 快照。这带来了两项关键能力断点续跑进程崩溃或人工中断后从上次 Checkpoint 恢复时间旅行回放或分叉到历史任意节点的状态执行流与 Checkpoint 关系 START → [agent] → checkpoint_1 → [tools] → checkpoint_2 → [agent] → ... ↑ ↑ 可回放/分叉 可回放/分叉4.2 三种 Checkpointer 实现Checkpointer适用场景安装MemorySaver本地开发、单进程测试内置无需安装SqliteSaver单机持久化轻量生产pip install langgraph-checkpoint-sqlitePostgresSaver多进程/分布式生产环境pip install langgraph-checkpoint-postgres# 生产环境PostgresSaverfromlanggraph.checkpoint.postgresimportPostgresSaverwithPostgresSaver.from_conn_string(postgresql://...)ascheckpointer:checkpointer.setup()# 建表首次运行graphbuilder.compile(checkpointercheckpointer)# config 中的 thread_id 是会话隔离键config{configurable:{thread_id:user-123-session-456}}resultgraph.invoke({messages:[HumanMessage(你好)]},config)五、Human-in-the-loop可控智能体的工程基础5.1 interrupt_before / interrupt_after这是 LangGraph 在生产场景最核心的差异化能力——在任意节点前后注入人工确认点而不需要修改节点代码本身# 在 tools 节点执行前暂停等待人工审批graphbuilder.compile(checkpointercheckpointer,interrupt_before[tools])config{configurable:{thread_id:t-001}}# 第一次运行agent 决策后暂停在 tools 入口snapshotgraph.invoke(input,config)# snapshot[__interrupt__] 包含待执行的工具调用信息# 人工审核获取当前状态stategraph.get_state(config)print(state.next)# (tools,) ← 下一个待执行节点print(state.values)# 完整 State 快照# 人工修改可以改写工具调用参数graph.update_state(config,{messages:[HumanMessage(改用更安全的参数)]})# 恢复执行传入 None 表示从 Checkpoint 继续graph.invoke(None,config)5.2 动态断点Dynamic Interruptv0.2引入了在节点函数内部动态触发中断的能力适合基于运行时条件决定是否需要人工介入fromlanggraph.typesimportinterruptdefsensitive_tool_node(state:AgentState)-dict:tool_callstate[messages][-1].tool_calls[0]# 运行时判断高风险操作才触发中断iftool_call[name]inHIGH_RISK_TOOLS:human_decisioninterrupt({question:f即将执行高风险操作{tool_call}是否确认,tool_call:tool_call})ifhuman_decision!approve:return{messages:[ToolMessage(操作已取消,tool_call_idtool_call[id])]}# 低风险工具直接执行returntool_node.invoke(state)六、多智能体模式6.1 Supervisor 模式一个 Supervisor Agent 负责将任务分发给专属 Worker Agent适合任务明确分工的场景┌─────────────────────────────────────────────────────┐ │ Supervisor Graph │ │ │ │ ┌──────────────────┐ │ │ │ supervisor │ │ │ │ (路由决策节点) │ │ │ └──────┬───────────┘ │ │ ┌───────┼───────────┐ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ coder │ │researcher│ │ writer │ │ │ │ (子图) │ │ (子图) │ │ (子图) │ │ │ └──────────┘ └──────────┘ └──────────┘ │ │ └───────┴───────────┘ │ │ ▼ 结果回流 │ │ supervisor → END │ └─────────────────────────────────────────────────────┘6.2 子图Subgraph每个 Worker 本身就是一个独立编译的CompiledGraph作为节点嵌入父图# 定义 Worker 子图coder_graphbuild_coder_subgraph().compile()# 嵌入父图子图作为节点supervisor_builderStateGraph(SupervisorState)supervisor_builder.add_node(coder,coder_graph)# 子图即节点supervisor_builder.add_node(researcher,researcher_graph)supervisor_builder.add_node(supervisor,supervisor_node)子图有独立的 State Schema父子图通过输入/输出 Schema 映射通信实现状态隔离。6.3 网状多智能体Swarmlanggraph-swarm2025 Q1 发布提供了智能体之间互相 handoff 的原语适合动态协作场景fromlanggraph_swarmimportcreate_handoff_tool,create_swarm# 每个 Agent 可以把控制权交给另一个 Agenttransfer_to_researchercreate_handoff_tool(agent_nameresearcher,description当需要搜索最新信息时转交给研究员)coder_agentcreate_react_agent(llm,tools[write_code,transfer_to_researcher])researcher_agentcreate_react_agent(llm,tools[search_web,transfer_to_coder])swarmcreate_swarm([coder_agent,researcher_agent],default_active_agentcoder)appswarm.compile(checkpointerMemorySaver())七、流式输出细粒度事件系统7.1 三种流式模式对比模式API输出粒度适用场景节点级graph.stream(input, stream_modevalues)每节点后的完整 State调试、状态监控增量更新graph.stream(input, stream_modeupdates)每节点后的 State 变化量增量 UI 更新事件级graph.astream_events(input, versionv2)LLM token 工具调用 节点生命周期实时聊天 UI7.2 astream_events 事件类型asyncforeventingraph.astream_events(input,config,versionv2):kindevent[event]ifkindon_chat_model_stream:# LLM 实时 tokenchunkevent[data][chunk].contentprint(chunk,end,flushTrue)elifkindon_tool_start:# 工具开始执行print(f\n[调用工具]:{event[name]}({event[data][input]}))elifkindon_chain_endandevent[name]agent:# Agent 节点执行完毕print(f\n[节点完成]: agent)八、LangGraph Platform从本地到生产v0.2同步发布了LangGraph Platform含 Cloud 和自托管版本将 LangGraph 从开发工具提升为完整的 Agent 运行时┌────────────────────────────────────────────────────────────┐ │ LangGraph Platform │ ├────────────────────────────────────────────────────────────┤ │ LangGraph ServerREST API │ │ ├── POST /runs → 创建并执行 Graph Run │ │ ├── GET /runs/{id} → 查询执行状态 │ │ ├── POST /runs/stream → SSE 流式输出 │ │ └── GET /threads/{id}/history → 完整 Checkpoint 历史 │ ├────────────────────────────────────────────────────────────┤ │ 后台任务队列Cron / Event-triggered Runs │ │ 持久化存储PostgreSQL Checkpointer Long-term Store │ │ 水平扩展多 Worker 共享 Checkpoint DB │ └────────────────────────────────────────────────────────────┘本地开发时langgraph dev命令一键启动等价的本地 Server并自动打开LangGraph Studio可视化调试 IDE。九、与竞品横向对比维度LangGraph 0.2AutoGen 0.4CrewAI 0.9Llama Agents执行模型显式有向图Actor 消息传递顺序/层次任务微服务事件总线状态管理TypedDict Reducer无显式 State无无持久化✅ Postgres/SQLite❌❌部分Human-in-loop✅ interrupt 原语✅ UserProxy❌❌流式输出✅ token 事件级部分❌部分子图/嵌套✅ 原生支持❌❌❌可视化调试✅ LangGraph Studio❌❌❌生产部署✅ Platform/Cloud手动手动手动学习曲线中图概念高低高AutoGen 在研究场景尤其是代码生成对话有独特优势其UserProxyAgentAssistantAgent的对话模式极其直观。但它的 Actor 模型缺乏结构化状态调试需要解析消息历史生产化改造成本高。CrewAI 的角色概念对非工程背景用户最友好但在状态持久化和流式输出上的缺失是生产环境的硬伤。十、总结维度核心要点编程模型StateGraph StateTypedDictReducer Node函数 Edge固定/条件持久化Checkpoint 在每节点后自动快照支持断点续跑与时间旅行可控性interrupt_before/after 动态interrupt()实现任意粒度人工介入多智能体Supervisor/子图/Swarm 三种模式覆盖分工→协作→动态 handoff 全场景可观测性astream_events细粒度事件 LangGraph Studio 可视化调试生产部署LangGraph Platform 提供 REST API、任务队列、水平扩展一体化LangGraph 代表了 AI Agent 工程化的一个方向把控制流的显式表达权还给工程师把局部推理权交给 LLM。它不试图让 LLM 规划一切而是在确定性工程结构和不确定性 AI 推理之间找到了一条严谨的分界线。随着v0.3持续演进多智能体协作原语和 Platform 的稳定性提升LangGraph 有望成为未来两年 Production-grade Agent 开发的事实标准。参考资料LangGraph 官方文档 v0.2 — langchain-ai.github.io/langgraphLangGraph Platform 文档LangGraph Swarm — GitHubHuman-in-the-loop 设计模式 — LangGraph DocsMulti-agent 架构综述 — LangGraph Docs