【LangGraph】最新版技术解析：有状态多智能体图执行引擎的架构原理与工程实践

LangGraph最新版技术解析：有状态多智能体图执行引擎的架构原理与工程实践

一、引言

亲爱的朋友们，创作不容易，若对您有帮助的话，请点赞收藏加关注哦，您的关注是我持续创作的动力，谢谢大家！有问题请私信或联系邮箱：jasonai.fn@gmail.com

2024 年初，当绝大多数 Agent 框架还在用while True循环包裹 ReAct 时，LangChain 团队悄然发布了 LangGraph。它的核心主张只有一句话：用有向图替代顺序循环，把控制流还给工程师。

不同于 AutoGPT 式的"让 LLM 自由规划一切"，LangGraph 的设计哲学更接近状态机：你声明节点、定义转移条件、设定持久化策略，LLM 只负责在每个节点做推理，而不是决定整个程序的走向。这一取舍让 LangGraph 在生产可靠性上远超同期竞品，也因此在短短一年内从 LangChain 的附属模块升格为独立的核心产品线。

本文以 LangGraphv0.2.x（2025 Q2 最新稳定版）为基准，覆盖图执行模型、状态管理、持久化与检查点、Human-in-the-loop、多智能体模式、流式输出与 LangGraph Platform 七大核心模块，给出可落地的工程解析。

二、核心概念：图执行模型全景

2.1 从 AgentExecutor 到 StateGraph

LangChain 旧版AgentExecutor的本质是一个死循环，控制流完全托付给 LLM 的输出格式：

# AgentExecutor 伪代码 —— 一切都是隐式的whileTrue:action=llm.predict(messages)# LLM 决定下一步ifaction=="Final Answer":breakresult=tool_executor.run(action)# 执行工具messages.append(result)# 追加结果

这个结构有三个致命弱点：无法中断恢复、无法条件分支、无法并发执行。LangGraph 用显式图结构解决了这三个问题：

┌─────────────────────────────────────────────────────────────┐ │ StateGraph 全景 │ │ │ │ ┌──────────┐ 条件边（路由函数） ┌──────────┐ │ │ │ agent │ ──────────────────► │ tools │ │ │ │ node │ │ node │ │ │ └──────────┘ ◄────────────────── └──────────┘ │ │ │ 固定边（回流） │ │ │ 条件边 │ │ ▼ │ │ END / 其他节点 │ │ │ │ State（TypedDict）在节点间流转，Checkpoint 在每步后持久化 │ └─────────────────────────────────────────────────────────────┘

2.2 三元素：State、Node、Edge

LangGraph 的所有构建都围绕三个基本概念展开：

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 决策"""response=llm_with_tools.invoke(state["messages"])return{"messages":[response]}# ③ 构建图builder=StateGraph(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 / store）graph=builder.compile(checkpointer=MemorySaver(),# 内存 Checkpointinterrupt_before=["tools"],# 工具执行前暂停)

3.2 编译产物的能力矩阵

compile()返回一个CompiledGraph，它实现了与 LCEL 相同的Runnable接口：

四、持久化与检查点（Checkpointer）

4.1 Checkpoint 机制

LangGraph 在每个节点执行完成后自动保存一次 Checkpoint，包含完整 State 快照。这带来了两项关键能力：

• 断点续跑：进程崩溃或人工中断后，从上次 Checkpoint 恢复
• 时间旅行：回放或分叉到历史任意节点的状态

执行流与 Checkpoint 关系： START → [agent] → checkpoint_1 → [tools] → checkpoint_2 → [agent] → ... ↑ ↑ 可回放/分叉 可回放/分叉

4.2 三种 Checkpointer 实现

# 生产环境：PostgresSaverfromlanggraph.checkpoint.postgresimportPostgresSaverwithPostgresSaver.from_conn_string("postgresql://...")ascheckpointer:checkpointer.setup()# 建表（首次运行）graph=builder.compile(checkpointer=checkpointer)# config 中的 thread_id 是会话隔离键config={"configurable":{"thread_id":"user-123-session-456"}}result=graph.invoke({"messages":[HumanMessage("你好")]},config)

五、Human-in-the-loop：可控智能体的工程基础

5.1 interrupt_before / interrupt_after

这是 LangGraph 在生产场景最核心的差异化能力——在任意节点前后注入人工确认点，而不需要修改节点代码本身：

# 在 tools 节点执行前暂停，等待人工审批graph=builder.compile(checkpointer=checkpointer,interrupt_before=["tools"])config={"configurable":{"thread_id":"t-001"}}# 第一次运行：agent 决策后暂停在 tools 入口snapshot=graph.invoke(input,config)# snapshot["__interrupt__"] 包含待执行的工具调用信息# 人工审核：获取当前状态state=graph.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 Interrupt）

v0.2引入了在节点函数内部动态触发中断的能力，适合基于运行时条件决定是否需要人工介入：

fromlanggraph.typesimportinterruptdefsensitive_tool_node(state:AgentState)->dict:tool_call=state["messages"][-1].tool_calls[0]# 运行时判断：高风险操作才触发中断iftool_call["name"]inHIGH_RISK_TOOLS:human_decision=interrupt({"question":f"即将执行高风险操作：{tool_call}，是否确认？","tool_call":tool_call})ifhuman_decision!="approve":return{"messages":[ToolMessage("操作已取消",tool_call_id=tool_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_graph=build_coder_subgraph().compile()# 嵌入父图（子图作为节点）supervisor_builder=StateGraph(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 网状多智能体（Swarm）

langgraph-swarm（2025 Q1 发布）提供了智能体之间互相 handoff 的原语，适合动态协作场景：

fromlanggraph_swarmimportcreate_handoff_tool,create_swarm# 每个 Agent 可以把控制权交给另一个 Agenttransfer_to_researcher=create_handoff_tool(agent_name="researcher",description="当需要搜索最新信息时，转交给研究员")coder_agent=create_react_agent(llm,tools=[write_code,transfer_to_researcher])researcher_agent=create_react_agent(llm,tools=[search_web,transfer_to_coder])swarm=create_swarm([coder_agent,researcher_agent],default_active_agent="coder")app=swarm.compile(checkpointer=MemorySaver())

七、流式输出：细粒度事件系统

7.1 三种流式模式对比

7.2 astream_events 事件类型

asyncforeventingraph.astream_events(input,config,version="v2"):kind=event["event"]ifkind=="on_chat_model_stream":# LLM 实时 tokenchunk=event["data"]["chunk"].contentprint(chunk,end="",flush=True)elifkind=="on_tool_start":# 工具开始执行print(f"\n[调用工具]:{event['name']}({event['data']['input']})")elifkind=="on_chain_end"andevent["name"]=="agent":# Agent 节点执行完毕print(f"\n[节点完成]: agent")

八、LangGraph Platform：从本地到生产

v0.2同步发布了LangGraph Platform（含 Cloud 和自托管版本），将 LangGraph 从开发工具提升为完整的 Agent 运行时：

┌────────────────────────────────────────────────────────────┐ │ LangGraph Platform │ ├────────────────────────────────────────────────────────────┤ │ LangGraph Server（REST 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）。

九、与竞品横向对比

AutoGen 在研究场景（尤其是代码生成对话）有独特优势，其UserProxyAgent+AssistantAgent的对话模式极其直观。但它的 Actor 模型缺乏结构化状态，调试需要解析消息历史，生产化改造成本高。CrewAI 的角色概念对非工程背景用户最友好，但在状态持久化和流式输出上的缺失是生产环境的硬伤。

十、总结

LangGraph 代表了 AI Agent 工程化的一个方向：把控制流的显式表达权还给工程师，把局部推理权交给 LLM。它不试图让 LLM 规划一切，而是在确定性工程结构和不确定性 AI 推理之间找到了一条严谨的分界线。随着v0.3持续演进多智能体协作原语和 Platform 的稳定性提升，LangGraph 有望成为未来两年 Production-grade Agent 开发的事实标准。

参考资料：

1. LangGraph 官方文档 v0.2 — langchain-ai.github.io/langgraph
2. LangGraph Platform 文档
3. LangGraph Swarm — GitHub
4. Human-in-the-loop 设计模式 — LangGraph Docs
5. Multi-agent 架构综述 — LangGraph Docs