🎯 StateGraph 基础

欢迎进入第三阶段!LangGraph 把 Agent 的控制流显式建模成图。本章从最小可运行的 StateGraph 开始——5 行代码定义一个图,掌握节点、边、状态三大核心概念,告别 AgentExecutor 的隐式循环。

本章目标

  • 理解 StateGraph 的"状态图"心智模型——节点读写共享状态
  • 掌握 add_node / add_edge / compile 三大核心 API
  • 理解 START / END 虚拟节点的作用
  • 认识到节点签名是 State → Partial<State>,返回的是"更新"而非"全量"
  • 跑通你的第一个状态图
📚 为什么需要 LangGraph

LC8 章结尾我们看到了 AgentExecutor 的局限:状态管理简单、无法中断、控制流固定。LangGraph 的思路是把控制流显式建模成"图"——这样状态、中断、并行、循环都能精确控制。本章是整个第三阶段的地基

状态图心智模型

StateGraph 的核心思想用一句话概括:多个节点(函数)通过读写一个共享的"状态"来协作

START 节点 A 读状态 → 写部分状态 节点 B 读状态 → 写部分状态 共享状态 State messages: [...] count: 3 END
图 LG1.1 · 节点通过读写共享状态协作;START/END 是虚拟起止点

核心规则:节点签名 State → Partial<State>

每个节点是一个函数:接收完整状态,返回状态的"部分更新"。注意是部分——你不需要返回整个状态,只返回要改的字段:

📄 langgraph/graph/state.py:130 · StateGraph 类定义(节选) python
class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
    """节点通过读写共享状态来通信的图。

    每个节点的签名是 State -> Partial<State>。
    每个状态键可以可选地标注一个 reducer 函数,
    用于聚合来自多个节点的值。reducer 签名:(Value, Value) -> Value。

    ⚠️ StateGraph 只是构建器,不能直接执行!
    必须先调用 .compile() 创建可执行图。
    """
    ...

# 节点函数示例:只返回要更新的字段
def my_node(state: State) -> dict:
    return {"count": state["count"] + 1}   # 只返回 count,不返回 messages
    # 返回的 dict 会和原 state 合并(下一章讲 reducer 决定怎么合并)

三大核心 API

1. add_node —— 添加节点

📄 state.py:662 · add_node python
graph.add_node("node_name", function)
# "node_name": 节点唯一标识
# function: State -> dict 的函数(也可以是 Runnable)

2. add_edge —— 添加边

📄 state.py:915 · add_edge python
graph.add_edge("node_a", "node_b")
# 执行完 node_a 后,必定去 node_b
# 特殊节点:START(图的起点)、END(图的终点)

3. compile —— 编译成可执行图

📄 state.py:1164 · compile python
compiled = graph.compile()
# 编译后返回 CompiledStateGraph(它继承自 Pregel!LG5 详解)
# 此时才能调用 invoke() / stream() / ainvoke() 等
⚠️ 易错点:StateGraph 不能直接 invoke

初学者最常犯的错:StateGraph(...) 之后直接调 .invoke()。必须先 .compile()!StateGraph 只是构建器,编译后的 CompiledStateGraph 才是可执行对象(它继承自 Pregel 引擎,LG5 会讲)。

START 与 END

每个图都有两个虚拟节点,定义在 constants.py:28

📄 langgraph/constants.py:28 · 虚拟起止点 python
START = sys.intern("__start__")   # 图的虚拟起点
END = sys.intern("__end__")       # 图的虚拟终点(到达即结束)

# 用法:
graph.add_edge(START, "first_node")   # 指定入口节点
graph.add_edge("last_node", END)      # 指定何时结束

可运行代码

🚀 你的第一个状态图

pip install langgraph。下面这个例子是 LangGraph 的"Hello World"。

📄 lg1_state_graph.py · 最小状态图 python
# pip install langgraph
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

# ============ ① 定义状态(TypedDict)============
class State(TypedDict):
    messages: list[str]
    count: int

# ============ ② 定义节点函数(State -> 部分状态)============
def greet(state: State) -> dict:
    """节点1:打招呼,追加一条消息。"""
    msg = f"你好!这是第 {state['count']} 次问候"
    return {"messages": state["messages"] + [msg]}  # 返回部分更新

def increment(state: State) -> dict:
    """节点2:计数器加1。"""
    return {"count": state["count"] + 1}

# ============ ③ 构建图 ============
graph = StateGraph(State)       # 用 State schema 创建构建器
graph.add_node("greet", greet)  # 添加节点
graph.add_node("increment", increment)

# 添加边(控制流)
graph.add_edge(START, "greet")          # 入口 → greet
graph.add_edge("greet", "increment")    # greet → increment
graph.add_edge("increment", END)        # increment → 结束

# ============ ④ 编译 + 执行 ============
app = graph.compile()   # 关键!编译后才可执行

result = app.invoke({"messages": [], "count": 1})
print(result)
# → {'messages': ['你好!这是第 1 次问候'], 'count': 2}
START greet +messages increment count+1 END
图 LG1.2 · 上例的图结构:START → greet → increment → END

可视化:用 get_graph().draw 看图结构

📄 查看图的结构(调试利器) python
# 编译后可以查看图的结构
print(app.get_graph().draw_ascii())
# 输出 ASCII 图:
#   +-----------+
#   | __start__ |
#   +-----------+
#          |
#          v
#     +-------+
#     | greet |
#     +-------+
#          |
#          v
#    +-----------+
#    | increment |
#    +-----------+
#          |
#          v
#    +---------+
#    | __end__ |
#    +---------+

# 还能画 mermaid 图(贴到 markdown 里渲染)
# print(app.get_graph().draw_mermaid())

图的循环:把 LangGraph 当 Agent 用

真正的 Agent 需要循环(思考-行动-观察)。虽然 add_edge 是单向的,但边可以指回之前的节点,形成环。这是 LangGraph 表达 Agent 循环的方式(条件边 LG3 会讲更灵活的控制):

📄 lg1_cycle.py · 带循环的图(模拟计数器跑 N 步) python
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    count: int
    limit: int

def step(state: State) -> dict:
    print(f"  执行第 {state['count']} 步")
    return {"count": state["count"] + 1}

def should_continue(state: State) -> str:
    """路由函数:决定下一步去哪。"""
    if state["count"] < state["limit"]:
        return "step"     # 还没到上限,继续循环
    return END            # 到上限,结束

graph = StateGraph(State)
graph.add_node("step", step)
graph.add_edge(START, "step")
# 条件边(LG3 详解):step 执行后,由 should_continue 决定去 step 还是 END
graph.add_conditional_edges("step", should_continue)

app = graph.compile()
app.invoke({"count": 0, "limit": 3})
# 输出:
#   执行第 0 步
#   执行第 1 步
#   执行第 2 步
# 循环 3 次后 count=3 >= limit=3,路由到 END 结束
💡 这就是 Agent 循环的 LangGraph 版本

step 换成"调用 LLM",把 should_continue 换成"判断有无 tool_calls"——这就是 LG8 章的 create_react_agent 的核心结构!LangGraph 用"图 + 条件边"实现了 LC8 的 AgentExecutor 循环,但更灵活

对比 AgentExecutor

维度AgentExecutor(LC8)StateGraph(本章)
控制流隐式 while 循环显式图,可视化
状态intermediate_steps 列表任意 TypedDict,可多字段
路由固定(Action/Finish 二态)条件边,任意分支
循环单一线性循环可多环、可嵌套子图
调试靠 verbose 打印draw_ascii 可视化 + 时间旅行

与生产实践对照

LangGraph 概念OpenCode 对应
State 共享状态Session 的消息历史 + 元数据(持久化在 DB)
节点(State → 部分状态)runLoop 内每步的处理逻辑(调 LLM/执行工具)
条件边路由runLoop 的 continue/stop/compact 三态判断
compile() 编译无显式编译(运行时直接循环)
add_edge START/END用户输入启动 / assistant 无 tool_calls 结束

小结

下一章

本章节点只返回了简单字段,但"多个节点写同一字段时怎么合并"还没解决。下一章 LG2 · 状态与 Reducer:学习 Annotated[T, reducer]add_messages——这是让 messages 列表正确累积的关键,也是 LC7 章说的"用 messages 承载状态"的底层机制。