🎯 StateGraph 基础 ★
欢迎进入第三阶段!LangGraph 把 Agent 的控制流显式建模成图。本章从最小可运行的 StateGraph 开始——5 行代码定义一个图,掌握节点、边、状态三大核心概念,告别 AgentExecutor 的隐式循环。
本章目标
- 理解 StateGraph 的"状态图"心智模型——节点读写共享状态
- 掌握
add_node/add_edge/compile三大核心 API - 理解
START/END虚拟节点的作用 - 认识到节点签名是
State → Partial<State>,返回的是"更新"而非"全量" - 跑通你的第一个状态图
LC8 章结尾我们看到了 AgentExecutor 的局限:状态管理简单、无法中断、控制流固定。LangGraph 的思路是把控制流显式建模成"图"——这样状态、中断、并行、循环都能精确控制。本章是整个第三阶段的地基。
状态图心智模型
StateGraph 的核心思想用一句话概括:多个节点(函数)通过读写一个共享的"状态"来协作。
核心规则:节点签名 State → Partial<State>
每个节点是一个函数:接收完整状态,返回状态的"部分更新"。注意是部分——你不需要返回整个状态,只返回要改的字段:
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 —— 添加节点
graph.add_node("node_name", function)
# "node_name": 节点唯一标识
# function: State -> dict 的函数(也可以是 Runnable)
2. add_edge —— 添加边
graph.add_edge("node_a", "node_b")
# 执行完 node_a 后,必定去 node_b
# 特殊节点:START(图的起点)、END(图的终点)
3. compile —— 编译成可执行图
compiled = graph.compile()
# 编译后返回 CompiledStateGraph(它继承自 Pregel!LG5 详解)
# 此时才能调用 invoke() / stream() / ainvoke() 等
初学者最常犯的错:StateGraph(...) 之后直接调 .invoke()。必须先 .compile()!StateGraph 只是构建器,编译后的 CompiledStateGraph 才是可执行对象(它继承自 Pregel 引擎,LG5 会讲)。
START 与 END
每个图都有两个虚拟节点,定义在 constants.py:28:
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"。
# 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}
可视化:用 get_graph().draw 看图结构
# 编译后可以查看图的结构
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 会讲更灵活的控制):
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 结束
把 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 结束 |
小结
- StateGraph = 节点(函数)通过读写共享状态协作的图。节点签名
State → Partial<State>。 - 三大 API:
add_node(加节点)、add_edge(加边)、compile(编译成可执行图)。 START/END是虚拟起止点。StateGraph 必须先 compile 才能 invoke。- 条件边 + 边指回前节点 = 循环,这是 LangGraph 表达 Agent 循环的方式。
- 相比 AgentExecutor,StateGraph 把控制流显式化、可视化、可任意编排。
本章节点只返回了简单字段,但"多个节点写同一字段时怎么合并"还没解决。下一章 LG2 · 状态与 Reducer:学习 Annotated[T, reducer] 和 add_messages——这是让 messages 列表正确累积的关键,也是 LC7 章说的"用 messages 承载状态"的底层机制。