Skip to content

核心概念总览

上一篇你跑通了第一个程序,对 State / Node / Edge 有了直观印象。本篇用一张大图把 LangGraph 所有核心概念串起来,建立全局地图——后续每一章都是对这张图的深入展开。

一、全景概念图

mermaid
graph TB
    subgraph 构图阶段
        SG[StateGraph 状态图容器]
        S[State 状态]
        N1[Node 节点]
        N2[Node 节点]
        E1[Edge 普通边]
        CE[ConditionalEdge 条件边]
        R[Reducer 归约器]
    end

    subgraph 运行阶段
        CG[CompiledGraph 编译产物]
        CP[Checkpointer 检查点]
        CFG[Config 配置 thread_id 等]
        ST[Streaming 流式输出]
        HITL[Interrupt 人机交互]
        SUB[Subgraph 子图]
    end

    S --> SG
    N1 -.注册.-> SG
    N2 -.注册.-> SG
    R -.修饰字段.-> S
    SG --> E1
    SG --> CE
    SG -->|compile| CG
    CG --> CP
    CG --> CFG
    CG --> ST
    CG -.可选.-> HITL
    CG -.可选.-> SUB

读图方法:上半部分"构图阶段"是你搭图时做的事;下半部分"运行阶段"是图跑起来后能用的能力。

二、核心概念速览

概念一句话详见
State 状态在节点间共享的数据容器,整个图的"工作内存"状态 State
Node 节点处理逻辑的单元,本质是普通 Python 函数节点 Node
Edge 边节点之间的连线,决定执行顺序边 Edge
ConditionalEdge 条件边根据状态动态选下一节点条件边 ConditionalEdge
StateGraph 状态图把节点和边装起来的容器StateGraph 状态图
Reducer 归约器决定字段如何被多个节点合并更新Reducer 归约器
编译与运行compile 生成可执行版本,invoke/stream 跑图编译与运行

三、概念逐个讲解(2-3 句话版)

State 状态

图执行过程中在节点之间流动的共享数据。用 TypedDict 或 Pydantic 定义。状态让节点解耦:节点 A 不需要知道节点 B 的存在,只要读/写状态即可。

Node 节点

普通 Python 函数,签名 (state: State) -> dict。返回的 dict 按 reducer 合并进状态。节点内可以调 LLM、调工具、读写文件、做计算——干任何事。

Edge 边

节点之间的连线 add_edge(source, target)。边本身不传数据——数据通过 state 传递,边只决定"执行完 source 后下一步去哪"。

ConditionalEdge 条件边

路由函数 router(state) -> str 根据当前状态返回一个字符串(下一节点名),实现动态分支。这是构建循环(如 ReAct)的关键。

StateGraph 状态图

核心容器 StateGraph(State),三个大方法 add_node / add_edge / add_conditional_edges 把图搭出来。

Reducer 归约器

字段更新策略。默认是"覆盖",但消息列表需要"追加",所以用 Annotated[list, add_messages]。reducer 决定了状态如何进化。

编译与运行

compile() 校验并生成可执行版本;invoke 一次性同步跑,stream 流式逐步返回。配合 config={"thread_id": ...} 和 checkpointer 可以让图"有记忆"。

四、进阶特性(后续模块深入)

mermaid
mindmap
  root((进阶能力))
    持久化
      MemorySaver 内存
      SQLite 检查点
      Postgres 生产
    人机交互 HITL
      interrupt 中断
      审核后继续
      Command 恢复
    流式
      stream_mode values
      stream_mode updates
      stream_mode messages
    子图
      嵌套图
      模块化复用
    多智能体
      Supervisor
      Swarm
    Send 多路分发
      并行扇出

这些在 04-进阶特性 模块展开,本篇只做总览。

五、LangGraph(Python) ↔ LangGraph4j(Java) 对照

如果你从 LangGraph4j 迁移过来,下面这张表能帮你快速对号入座:

概念LangGraph (Python)LangGraph4j (Java)
状态类型TypedDict / Pydantic v2Record<String,Object> / 自定义类
字段合并策略Annotated[list, add_messages]@Reducer 注解 + 类
消息状态MessagesState(内置)MessagesState(内置)
节点普通 Python 函数AgentNode 接口 / lambda
节点签名(state: State) -> dictapply(state) -> Map<String,Object>
图容器StateGraph(State)StateGraph.of(State.class)
添加节点add_node("name", fn)addNode("name", fn)
添加边add_edge("a", "b")addEdge("a", "b")
条件边add_conditional_edges(src, router, path_map)addConditionalEdges(src, router, path_map)
入口/出口START / END(常量)START / END(常量)
编译graph.compile()graph.compile()
编译参数compile(checkpointer=..., interrupt_before=...)compile(checkpointer, interrupt_before)
运行app.invoke(input)app.invoke(input)
流式app.stream(input) / astreamapp.stream(input)
检查点MemorySaver / SQLite / PostgresMemorySaver
预置 ReActcreate_react_agentcreateReActAgent
工具@tool 装饰器@Tool 注解
中断interrupt(value)interrupt(value)
恢复app.invoke(Command(resume=...))app.invoke(Command(resume=...))
多路分发Send(node, data)Send(node, data)

迁移提示:

  1. API 几乎一一对应,主要差异是 Python 函数一等公民、Java 要 lambda/接口。
  2. 类型系统不同:Python 用 TypedDict(鸭子类型),Java 要严格类定义。
  3. 生态不同:Python 版本因 LangChain 庞大生态,工具/向量库选择更多。
  4. 执行模型相同:都是基于状态快照 + reducer 合并的图执行。
  5. 运行时校验:Python 默认无校验(除非用 Pydantic),Java 因静态类型天然更严格。
  6. 包导入路径:Python 从 langgraph.graph / langgraph.prebuilt / langgraph.checkpoint.memory 引入;Java 从对应的 org.langgraph4j.* 包引入。

六、学习顺序建议

mermaid
graph LR
    A[State 状态] --> B[Node 节点]
    B --> C[Edge 边]
    C --> D[ConditionalEdge 条件边]
    D --> E[StateGraph 状态图]
    E --> F[Reducer 归约器]
    F --> G[编译与运行]
    G --> H[进阶: Checkpointer/HITL/Streaming/Subgraph]

推荐顺序:

  1. 先理解 State——其他概念都围绕它展开。
  2. 再看 Node——它读写 State。
  3. Edge 把 Node 连起来。
  4. ConditionalEdge 让图能分叉和循环。
  5. StateGraph 把上面都装进容器。
  6. Reducer 决定字段如何进化。
  7. 编译与运行 让图真正跑起来。
  8. 进阶特性(持久化、HITL、流式、子图)建立在以上基础上。

提示:如果跳着读,至少先把 State 和 Node 两篇看完,否则后面的代码会看不懂字段从哪来、节点怎么写。

七、一个小例子串所有概念

python
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 1. State(带 reducer)
class State(TypedDict):
    topic: str
    messages: Annotated[list, add_messages]  # 消息累加

# 2. Node
def think(state: State) -> dict:
    return {"messages": [("user", f"关于 {state['topic']} 的思考")]}

def respond(state: State) -> dict:
    return {"messages": [("assistant", "好主意!")]}

# 3. 路由:决定下一节点
def router(state: State) -> str:
    if len(state["messages"]) < 4:
        return "think"   # 消息少就继续思考
    return "END"          # 够了就结束

# 4. StateGraph 装配 + 5. ConditionalEdge
g = StateGraph(State)
g.add_node("think", think)
g.add_node("respond", respond)
g.add_edge(START, "think")
g.add_conditional_edges("think", router, {"think": "respond", "END": END})
g.add_edge("respond", "think")  # 形成循环

# 6. 编译
app = g.compile()

# 7. 运行
for ev in app.stream({"topic": "学 LangGraph", "messages": []}):
    print(ev)

这个例子虽然简单,但已经涵盖了 State / Node / Edge / ConditionalEdge / StateGraph / Reducer / 编译运行 全部七个核心概念。

八、小结

  • LangGraph 七大核心概念:State、Node、Edge、ConditionalEdge、StateGraph、Reducer、编译与运行。
  • 上半部分是"构图",下半部分是"运行"。
  • 与 LangGraph4j 概念一一对应,迁移友好。
  • 建议按 State → Node → Edge → 条件边 → StateGraph → Reducer → 运行 的顺序学。
  • 一段 30 行代码就能串起所有核心概念。

接下来进入核心概念详解,从最基础的 状态 State 开始。