LangGraph:状态图编排
你将学会什么
Section titled “你将学会什么”- 用
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 langgraphfrom typing import TypedDict, Annotatedimport operatorfrom 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") # 入口哨兵 → llmbuilder.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 循环的标准写法。
StateGraph(State)之后为什么必须compile()?- 节点函数的返回值应该是什么?
Annotated[list, operator.add]的作用是什么?
参考答案
- StateGraph 只是图的声明(Blueprint),compile() 才把它变成可执行的图。
- 返回要更新的字段增量(dict),不是整个状态;未返回的字段保持原样。
- 它是 reducer:多个节点对同一字段的更新会追加合并,防止后写的覆盖先写的消息。
- LangGraph 把工作流建模成 StateGraph:节点 + 边 + 条件边
- State 用 TypedDict 声明,节点返回字段增量,reducer 控制合并方式
- 文档以 docs.langchain.com 为准;compile → invoke 是标准执行链