跳转到内容

LangGraph:状态图编排

  • StateGraph + TypedDict 定义一个全图共享的状态
  • add_node / add_edge 把流程画成图,并 compile() 编译
  • 理解节点只返回「字段增量」和 reducer 的累加语义

LangGraph 是 LangChain 团队的 Agent 编排引擎,2025-10-22 发布 1.0 后成为生产环境的事实标准。它的核心思想是把工作流画成一张「有向状态图」:节点(Node)是步骤,边(Edge)是流转,条件边(Conditional Edge)让 LLM 决定下一步走哪条路——就像程序员的「流程图 + 存档点」。官方文档现在在 docs.langchain.com(旧域名只剩跳转页)。所有节点共享一个 State,用 TypedDict 声明,先看一个能跑的最小例子:

# 安装: pip install -U langgraph
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
# ① State 是全图共享的 TypedDict;Annotated[list, operator.add] 是 reducer:
# 多个节点对 messages 的更新会「追加」而不是互相覆盖。
class State(TypedDict):
messages: Annotated[list, operator.add]
llm_calls: int
# ② 节点 = 普通函数:读 state,返回「要更新的字段」。
def llm_node(state: State) -> dict:
return {"messages": [{"role": "ai", "content": "你好,我是图里的一个节点!"}],
"llm_calls": state.get("llm_calls", 0) + 1}
# ③ 画图:定义状态 → 加节点 → 连边 → 编译
builder = StateGraph(State)
builder.add_node("llm", llm_node)
builder.add_edge(START, "llm") # 入口哨兵 → llm
builder.add_edge("llm", END) # llm → 出口哨兵
graph = builder.compile() # 编译是必需的!
# ④ 运行:invoke 传入初始状态
result = graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})
print(result["messages"]) # 用户消息 + AI 回复
print(result["llm_calls"]) # 1

三个新手易错点:忘记 compile() 就直接 invoke()(StateGraph 只是声明,不执行逻辑);节点返回的是「字段增量」而不是整个状态,没返回的字段保持原样;messages 字段不加 reducer 时,后一个节点会覆盖前一个节点的消息。更复杂的流程(LLM 决定是否调工具)会用 add_conditional_edges("llm_call", should_continue, ["tool_node", END]),路由函数返回「节点名或 END」字符串——这是生产里 ReAct 循环的标准写法。

  1. StateGraph(State) 之后为什么必须 compile()
  2. 节点函数的返回值应该是什么?
  3. Annotated[list, operator.add] 的作用是什么?
参考答案
  1. StateGraph 只是图的声明(Blueprint),compile() 才把它变成可执行的图。
  2. 返回要更新的字段增量(dict),不是整个状态;未返回的字段保持原样。
  3. 它是 reducer:多个节点对同一字段的更新会追加合并,防止后写的覆盖先写的消息。
  • LangGraph 把工作流建模成 StateGraph:节点 + 边 + 条件边
  • State 用 TypedDict 声明,节点返回字段增量,reducer 控制合并方式
  • 文档以 docs.langchain.com 为准;compile → invoke 是标准执行链
LangGraph 小测
x
1 / 3

StateGraph 构建完成后,执行前必须做什么?