LangGraph 完整核心概念概述LangGraph 是面向 Agent 的状态图编排框架底层基于 Google Pregel BSP批量同步并行模型支持循环、分支、多 Agent、断点续跑和普通 DAGDirected 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: stroperator.add简单列表拼接适合日志、trajectory 轨迹记录add_messagesLangGraph 内置处理 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(contentresult)]}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(checkpointerxxx)编译阶段做的事情校验图结构检查是否存在孤立节点内部创建Channels通道对应 state 每个字段注册节点、边、条件路由逻辑如果传入checkpointer开启状态持久化、断点续跑。compile 之后才是真正可执行的 Runnable 对象。5. Checkpointer检查点生产级重要组件每一个超级步superstep执行完成自动保存一份完整 state 快照支持中断、人工介入、断点续跑内存 / 文件 / Redis 都可以做存储Agent 执行失败可以从快照恢复重新跑不需要从头执行。6. SuperStep超级步底层执行单元LangGraphLangGraph 底层 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 通道触发STARTSuperStep #1执行supervisor节点读取当前完整 stateLLM 分析用户任务输出next_agent写入 state节点返回增量更新合并到全局 state条件边读取next_agent路由到对应子 Agent 节点SuperStep #2执行选中的子 Agentlighthouse/jmeter读取 state执行工具调用执行性能测试将工具输出追加写入messages子 Agent 执行完成固定边跳转回supervisorSuperStep #3再次运行 supervisor读取刚刚子 Agent 返回的结果LLM 再次判断是否还需要调用其他工具 / 子 Agent如果还需要干活继续路由对应子节点如果任务全部完成next_agentfinish条件边路由到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 ChainLangGraph流程只能线性 DAG不支持循环支持循环、分支、条件跳转适合 Agent 思考循环状态外部手动维护记忆内置 State 状态管理、归约合并中断恢复不支持Checkpointer 支持断点续跑多 Agent很难实现调度Supervisor 原生支持多智能体编排六、生产级 Agent 标准链路定义State →编写各个Node函数 →add_node注册节点 →add_edge/add_conditional_edges编排流程 →compile(带checkpointer) →invoke/stream执行 →采集trajectory轨迹 →Eval‑Harness做自动化评测