StateGraph:状态管理核心

面试官:「LangGraph 的状态是怎么管理的?」

你:「用一个 State 字典,在节点间流转。」

面试官:「多个节点同时更新同一个字段怎么办?」

你:「用 Reducer 来控制怎么合并... 具体怎么写我得查一下。」

🤔 这一章,把状态管理讲透。

2.1 State 设计模式

State 是 LangGraph 的核心。本章深入探讨如何设计高效、可维护的状态结构。

2.1.1 基础 State 定义

python
from typing import TypedDict class BasicState(TypedDict): messages: list[str] count: int result: str | None

2.1.2 带 Reducer 的 State

python
from typing import TypedDict, Annotated from operator import add, itemgetter class AdvancedState(TypedDict): # messages 会自动累加 messages: Annotated[list, add] # count 会替换(非累加) count: int # 使用 itemgetter 只取最新的 latest_msg: Annotated[str, itemgetter(-1)]
图 2-1:不同 Reducer 的行为
messages: [add] ["hi"] → ["hi", "hello"] count: int(默认替换) 5 → 6(直接替换) last: [itemgetter(-1)] ["hi", "bye"] → "bye"
图 2-2:Reducer 合并机制 — 每次节点执行后的状态更新
Step 0 messages=[] count=0 result=None Step 1 → Node A add → ["hi"] count=0 (不变) result=None Step 2 → Node B add → ["hi","bye"] count=1 (替换) result=None Step 3 → Final messages=["hi","bye"] count=1 result=None

2.2 自定义 Reducer

除了内置的 additemgetter,你还可以定义自己的 Reducer。

python
from typing import TypedDict, Annotated # 自定义 Reducer:只保留最近的 N 条消息 def keep_last_n_messages(existing, new, n=5): combined = existing + new return combined[-n:] class ChatState(TypedDict): messages: Annotated[list, keep_last_n_messages] session_id: str
python
# 自定义 Reducer:合并字典 def merge_dicts(existing: dict, new: dict) -> dict: return {**existing, **new} class AgentState(TypedDict): context: Annotated[dict, merge_dicts] step: int
面试要点

Reducer 本质上是一个函数(existing, new) -> result。它接收旧值和新值,返回合并后的结果。

2.3 条件状态更新

有时候需要根据当前状态有条件地更新某些字段:

python
from typing import TypedDict, Literal class RouterState(TypedDict): next_node: Literal["node_a", "node_b", "END"] confidence: float # 条件更新:根据 confidence 决定下一个节点 def route_based_on_confidence(state: RouterState) -> RouterState: if state["confidence"] > 0.8: return {"next_node": "node_a"} elif state["confidence"] > 0.5: return {"next_node": "node_b"} else: return {"next_node": "END"}

2.4 状态模式实战

2.4.1 ReAct Agent 状态

python
from typing import TypedDict, Annotated from operator import add class ReActState(TypedDict): messages: Annotated[list, add] # 对话历史 observations: Annotated[list, add] # 工具执行结果 reasoning: Annotated[list, add] # 推理过程 tool_calls: list[str] # 调用的工具列表

2.4.2 对话 Agent 状态

python
from typing import TypedDict, Annotated from operator import add class ConversationState(TypedDict): messages: Annotated[list, add] # 完整对话历史 user_profile: dict # 用户画像(会合并) conversation_turns: int # 对话轮次 intent: str | None # 识别的用户意图

2.5 本章小结

  • State 使用 TypedDict 定义,支持任意嵌套结构
  • Annotated[type, reducer] 指定如何合并更新
  • 内置 Reducer:add(累加)、itemgetter(取特定位置)
  • 自定义 Reducer:(existing, new) -> result
  • 合理的状态设计是构建复杂工作流的基础
下章预告

第 3 章我们将学习 Nodes 和 Edges 的进阶用法:LLM 节点、工具节点、条件边、以及如何组织代码结构。