LangGraph 完整核心概念 + 运行流程-开发者社区

LangGraph 完整核心概念概述

LangGraph 是面向 Agent 的状态图编排框架，底层基于 Google Pregel BSP（批量同步并行）模型，支持循环、分支、多 Agent、断点续跑；和普通 DAG（Directed Acyclic Graph：有向无环图）最大区别：允许图里面有环（循环），完美适配 Agent「思考 - 调用工具 - 再思考」循环逻辑。

一、核心概念

1. State（状态）【全局共享数据总线】LangGraph

整个图唯一共享的数据容器，所有节点只读 state，返回增量更新，禁止直接修改入参 state 对象。

定义：TypedDict/ Pydantic BaseModel；
普通字段：节点返回字典，直接覆盖原值；
Annotated + reducer（归约器）：定义字段如何合并新旧状态，这是编写 Agent 需要重点掌握的点。

from typing import TypedDict, Annotated import operator from langchain_core.messages import BaseMessage class AgentState(TypedDict): # operator.add：列表拼接，新消息追加，不覆盖旧列表 messages: Annotated[list[BaseMessage], operator.add] next_agent: str report_content: str

operator.add：简单列表拼接；适合日志、trajectory 轨迹记录；
add_messages：LangGraph 内置，处理 Message 对象，支持按 id 覆盖更新，对话消息优先用它LangChain。

如果不加 Annotated 归约器：节点返回{"messages":[new_msg]}会直接把历史消息全部覆盖丢失。

2. Node（节点）【计算单元】LangGraph

本质就是普通 Python 函数。

入参：接收当前完整state；
执行逻辑：可以调用 LLM、调用工具（lighthouse/jmeter）、数据库、业务逻辑；
返回值：返回字典，只写需要更新的字段增量，不要返回完整 state；
分类：LLM 节点、工具节点、Supervisor 主管调度节点。

def lighthouse_agent(state: AgentState): # 读取state msgs = state["messages"] # 执行业务逻辑 result = run_lighthouse_test(msgs[-1].content) # 返回增量更新，不是完整state return {"messages": [AIMessage(content=result)]}

3. Edge（边）【控制流转，节点之间的通路】

分为 2 大类：普通边、条件边。

普通边 add_edge (A,B)：A 执行完，固定流转到 B；
条件边 add_conditional_edges：根据当前 state，动态决定下一个跑哪个节点。

graph.add_conditional_edges( "supervisor", lambda s: s["next_agent"], { "lighthouse_agent": "lighthouse_agent", "jmeter_agent": "jmeter_agent", "analyst_agent": "analyst_agent", "finish": END } )

supervisor 节点执行完成，读取 state 里面next_agent字段的值，映射到对应节点，如果返回finish，直接走到 END 结束图。

START：虚拟入口节点，图执行起点，没有业务逻辑；
END：虚拟终止节点，走到 END，整个 graph.invoke () 结束，返回最终 state。

4. compile ()【编译】

graph = builder.compile(checkpointer=xxx)

编译阶段做的事情：

校验图结构，检查是否存在孤立节点；
内部创建Channels通道，对应 state 每个字段；
注册节点、边、条件路由逻辑；
如果传入checkpointer，开启状态持久化、断点续跑。

compile 之后，才是真正可执行的 Runnable 对象。

5. Checkpointer（检查点）生产级重要组件

每一个超级步（superstep）执行完成，自动保存一份完整 state 快照；
支持中断、人工介入、断点续跑；
内存 / 文件 / Redis 都可以做存储；
Agent 执行失败，可以从快照恢复，重新跑，不需要从头执行。

6. SuperStep（超级步）底层执行单元LangGraph

LangGraph 底层 BSP 模型核心概念：

关于bsp的详细信息可以参考文章：

一文带大家理解LangGraph 的底层模型——BSP （Bulk‑Synchronous Parallel）-CSDN博客

一个 SuperStep = 一轮执行周期；
同一 SuperStep 内满足条件的节点可以并行执行；
全部节点执行完毕，统一合并状态更新，写入 Channels；
再进入下一轮 SuperStep，判断下一批要运行的节点；
循环往复，直到没有任何节点被触发，执行终止。

通俗理解：一轮一轮跑，同一轮多个节点可以并行；全部跑完统一合并状态，不会出现并发读写冲突。

7. Trajectory（轨迹）

完整图执行历史：每一轮 superstep 的 state 快照、节点输入输出、工具调用记录。

就是我们做 Agent 测试，Eval‑Harness 要采集的核心日志。

二、完整运行流程

举例:Supervisor 多 Agent （性能测试）架构

用户输入 → START → supervisor主管节点 ↓（条件边路由） ├─ lighthouse_agent （子Agent节点） ├─ jmeter_agent （子Agent节点） └─ analyst_agent （子Agent节点） ↓子节点执行完毕，都会回到 supervisor主管 循环往复，直到主管判断任务完成 → END → 返回最终state

分步拆解 invoke 执行完整链路

调用 graph.invoke (input_dict)，传入初始输入；
框架把输入写入 Channels 通道，触发START；
SuperStep #1：执行supervisor节点
- 读取当前完整 state；
- LLM 分析用户任务，输出next_agent写入 state；
- 节点返回增量更新，合并到全局 state；
- 条件边读取next_agent，路由到对应子 Agent 节点；
SuperStep #2：执行选中的子 Agent（lighthouse/jmeter）
- 读取 state，执行工具调用，执行性能测试；
- 将工具输出追加写入messages；
- 子 Agent 执行完成，固定边跳转回supervisor；
SuperStep #3：再次运行 supervisor
- 读取刚刚子 Agent 返回的结果；
- LLM 再次判断：是否还需要调用其他工具 / 子 Agent；
- 如果还需要干活：继续路由对应子节点；
- 如果任务全部完成：next_agent="finish"，条件边路由到END；
到达 END，整个图停止执行，返回完整最终 state；
如果配置 checkpointer，每一步都会保存状态快照。

关键点：子 Agent 跑完，强制回到主管，由主管做下一次决策，这就是 Supervisor 模式。

三、关键误区（写代码高频踩坑）

Node 函数不要直接修改传入的 state 对象

# 错误 def node(state): state["messages"].append(AIMessage("xxx")) return state # 正确：返回增量字典，由框架做状态合并 def node(state): return {"messages": [AIMessage("xxx")]}

忘记Annotated[..., operator.add]，messages 列表被覆盖丢失历史对话。
compile () 之前没有定义 START、END 路径，编译报错。
条件边路由函数返回的字符串，必须和 add_node 注册节点名称完全一致，大小写敏感。
不要把compile()写在循环内部，compile 只执行一次。

四、invoke () vs stream ()

graph.invoke(input)：同步阻塞，全部跑完返回最终完整 state；适合自动化 Eval‑Harness 测试；
graph.stream(input)：迭代器，每一个 SuperStep 执行完成就返回当前更新片段，可以实时流式输出，看中间执行轨迹。

五、和普通 LangChain Chain 区别

LangChain Chain	LangGraph
流程	只能线性 DAG，不支持循环	支持循环、分支、条件跳转；适合 Agent 思考循环
状态	外部手动维护记忆	内置 State 状态管理、归约合并
中断恢复	不支持	Checkpointer 支持断点续跑
多 Agent	很难实现调度	Supervisor 原生支持多智能体编排