企业官网建设流程全解析

LangGraph State 设计：多节点之间怎么安全地共享数据

写 LangGraph 时，最容易出问题、也最影响后续扩展性的，
不是节点逻辑，而是State 怎么定义。

State 定义错了，轻则消息被覆盖，重则并发写入丢数据，
而且这类 bug 很难复现。

这篇讲清楚 State 的两种写法、reducer 机制，以及几个高频踩坑点。

一、State 是什么

State 就是在图的节点之间流动的那份共享数据。
每个节点接收当前 State，返回"要更新的部分"，
LangGraph 把更新合并回 State，再传给下一个节点。

defnode_a(state):print(state)# 读：完整的当前状态return{"foo":1}# 写：只返回要改的字段

返回值是"增量更新"，不是"完整新状态"。
这是新手第一个误解——你不需要把没变的字段原样返回。

二、两种定义方式

2.1 TypedDict（推荐）

fromtypingimportTypedDictclassState(TypedDict):question:strdocs:list[str]answer:str

优点：轻量、可读、IDE 友好。日常首选。

2.2 Pydantic BaseModel

frompydanticimportBaseModelclassState(BaseModel):question:strdocs:list[str]=[]answer:str=""

优点：有运行时校验，类型错了立刻报错。
代价是每次更新都要做一次校验，有轻微性能开销。

什么时候用 Pydantic？状态结构复杂、字段有约束（范围、枚举）、
或者你需要序列化/反序列化状态时。其余用 TypedDict。

2.3 内置的 MessagesState

对话类 Agent 有现成的：

fromlanggraph.graphimportMessagesState# 等价于：classMessagesState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]

直接用：

fromlanggraph.graphimportStateGraph,MessagesStatedefchatbot(state:MessagesState):return{"messages":[llm.invoke(state["messages"])]}graph=StateGraph(MessagesState)graph.add_node("chatbot",chatbot)

三、核心机制：Reducer

这是 State 设计里最重要的概念。

默认情况下，节点返回{"foo": 1}会覆盖foo原来的值。
但很多场景我们需要的是追加，比如消息列表。

Reducer 就是用来改变"合并行为"的：

fromtypingimportAnnotatedfromlanggraph.graph.messageimportadd_messagesclassState(TypedDict):messages:Annotated[list,add_messages]# 追加，并支持按 id 更新/删除

Annotated[类型, reducer函数]的格式：
第一个参数是类型，第二个是"怎么合并"。

3.1 自己写 reducer

fromtypingimportAnnotated,TypedDictimportoperatorclassState(TypedDict):# 列表追加logs:Annotated[list[str],operator.add]# 取最大值max_score:Annotated[float,max]# 字典浅合并meta:Annotated[dict,lambdaa,b:{**a,**b}]# 覆盖（默认行为，不写就是覆盖）current_step:str

operator.add对列表是拼接，对数字是相加——同一个 reducer 在不同类型上行为不同，
用的时候注意。

3.2 add_messages 不只是追加

add_messages是专门给消息列表用的，它比简单追加聪明：

fromlangchain_core.messagesimportAIMessage# 已有消息 id 相同 → 更新而不是追加return{"messages":[AIMessage(content="修正后的内容",id=existing_id)]}# 删除：用 RemoveMessagefromlanggraph.graph.messageimportRemoveMessagereturn{"messages":[RemoveMessage(id=some_id)]}

消息需要带id才能被识别为"同一条"。
LangChain 的消息对象默认有 id，自己构造时要注意。

四、四个高频坑

坑 1：字段没在 State 里声明

classState(TypedDict):messages:Annotated[list,add_messages]defmy_node(state):return{"temp":123}# ❌ State 里没有 temp

会报错或静默丢弃（取决于版本）。
节点只能返回 State 里已声明的字段。想加字段就先改 State 定义。

坑 2：忘了用 Annotated，消息被覆盖

# ❌ 错：每次返回都覆盖掉全部历史，模型永远只看到最后一轮classState(TypedDict):messages:list# ✅ 对classState(TypedDict):messages:Annotated[list,add_messages]

这个错误的表现是"对话没有记忆"，而且很隐蔽——
单轮测试完全正常，多轮才暴露。

坑 3：节点返回完整 State 而不是增量

# ❌ 冗余且危险：把旧消息又追加了一遍defnode(state):return{"messages":state["messages"]+[new_msg]}# ✅ 只返回增量，reducer 负责合并defnode(state):return{"messages":[new_msg]}

第一种写法会让消息翻倍，几轮之后上下文爆炸。

坑 4：并行节点写同一个字段

graph.add_edge("start","a")graph.add_edge("start","b")# a 和 b 并行# a 返回 {"count": 1}，b 返回 {"count": 2}# 谁最后完成？不确定 → 结果不稳定

并行分支写同一个非 reducer 字段，结果是不确定的。
要么让它们写不同字段，要么给该字段配一个可交换的 reducer
（如operator.add、max）。

五、实战：一个带中间产物的 RAG 状态

fromtypingimportAnnotated,TypedDict,Optionalfromlangchain_core.messagesimportAnyMessagefromlangchain_core.documentsimportDocumentfromlanggraph.graph.messageimportadd_messagesclassRAGState(TypedDict):# 对话历史messages:Annotated[list[AnyMessage],add_messages]# 改写后的独立问题（每轮覆盖）standalone_question:str# 检索到的文档（每轮覆盖）retrieved_docs:list[Document]# 重试次数（累加）retry_count:Annotated[int,lambdaa,b:a+b]# 最终答案answer:Optional[str]defrewrite(state:RAGState):q=condense_chain.invoke({"history":state["messages"][:-1],"question":state["messages"][-1].content,})return{"standalone_question":q}defretrieve(state:RAGState):docs=retriever.invoke(state["standalone_question"])return{"retrieved_docs":docs}defgenerate(state:RAGState):ctx=format_docs(state["retrieved_docs"])ans=(prompt|llm|StrOutputParser()).invoke({"context":ctx,"question":state["standalone_question"],})return{"answer":ans}defgrade(state:RAGState):"""判断答案质量，决定要不要重试"""ifis_good(state["answer"]):return"end"ifstate["retry_count"]>=2:return"end"return"retry"defbump_retry(state:RAGState):return{"retry_count":1}

设计原则：

• 每轮会被重新计算的字段（问题、检索结果）→ 覆盖（不配 reducer）；
• 需要累积的字段（重试次数、日志）→ 配累加 reducer；
• 消息历史 → 一律用add_messages。

六、读写 State 的几个技巧

6.1 只读访问用 state[“x”]，别用 .get 掩盖问题

# 如果字段必须存在，直接取值让它报错，早发现早修q=state["standalone_question"]# 只有真正可选的字段才用 getans=state.get("answer")

6.2 在节点里更新嵌套结构

defadd_tag(state:RAGState):# ❌ 直接改侵入式修改，LangGraph 可能感知不到state["meta"]["tags"].append("new")# ✅ 返回新对象new_meta={**state["meta"],"tags":state["meta"]["tags"]+["new"]}return{"meta":new_meta}

不要就地修改（mutate）State。返回新对象，让 LangGraph 处理合并，
这样 checkpoint 才能正确记录变更。

6.3 私有中间数据不要放 State

State 会被 checkpoint 持久化。
数据库连接、临时大对象这类不该落盘的东西，用闭包或全局变量持有：

db=get_connection()# 模块级，不进 Statedefquery_node(state:RAGState):rows=db.execute(...)return{"retrieved_docs":rows}

七、小结

• State 是节点间流动的共享数据，节点返回的是增量更新，不是完整状态；
• 定义方式：TypedDict（推荐）/ Pydantic（要校验）/ 内置MessagesState（对话场景）；
• Reducer 是核心：Annotated[类型, 合并函数]决定"覆盖还是追加"；
  add_messages支持按 id 更新和删除；
• 四个坑：字段没声明、忘了 Annotated 导致消息被覆盖、
  返回完整 State 导致消息翻倍、并行节点写同一字段结果不确定；
• 设计原则：每轮重算的字段覆盖，累积字段配 reducer，消息一律 add_messages；
• 不要就地修改 State，返回新对象；不落盘的资源别放 State。

下一篇讲 Checkpointer——怎么让图的状态持久化、支持断点续跑。