LangGraph 状态管理进阶：Overwrite、输入输出模式与私有状态！

文章目录

• • 先说结论
  • 5.1 使用 Overwrite 绕过 reducer
  • • 为什么需要 Overwrite
    • 代码示例
    • 运行结果说明
    • 适用场景
    • 实战提醒
  • 5.2 使用独立的输入 / 输出模式
  • • 为什么要区分输入、输出和内部状态
    • 参数含义
    • 代码示例
    • 结果说明
    • 这个设计的价值
    • 实战建议
  • 5.3 在节点间传递私有状态
  • • 一个典型例子
    • 代码示例
    • 一次执行后，你会看到什么
    • 这种设计适合什么场景
    • 这里最容易误解的一点
  • 三个能力放在一起，怎么理解最清楚
  • • 1. `Overwrite`
    • 2. `input_schema` / `output_schema` / `state_schema`
    • 3. 节点间私有状态
  • 实战里怎么选
  • • 场景 1：字段需要继续累积
    • 场景 2：字段需要整段替换
    • 场景 3：对外接口和内部状态不一致
    • 场景 4：中间敏感数据只允许局部流转
  • 常见坑点
  • • 坑 1：以为所有状态更新都应该走 append
    • 坑 2：把内部中间字段直接暴露给输出
    • 坑 3：把“最终不输出”和“节点不可见”混为一谈
    • 坑 4：状态设计过早贪大
  • 总结

这一篇专门讲 LangGraph 里三个很容易被忽略、但一旦做复杂工作流就绕不过去的能力：Overwrite、独立的输入/输出模式，以及节点间私有状态传递。

先说结论

如果已经会写最基础的 StateGraph，接下来最该补的不是更多节点，而是状态管理能力。

因为真实项目里，问题通常不是“图不会跑”，而是：

• 某个字段到底该追加还是覆盖？
• 用户输入、内部状态、最终输出能不能分开？
• 中间节点的敏感数据，怎么在节点之间传递但不暴露给外部？

这三个问题，刚好对应本文的三个主题：

5.1 使用 Overwrite 绕过 reducer

在 LangGraph 中，reducer 用来控制状态更新的处理方式。

默认情况下，每个状态键都可以配置自己的 reducer 函数，用于决定多个节点更新同一个字段时应该如何合并。

最常见的例子就是消息列表：

• 正常对话时，新消息通常会追加到 messages 中
• 但某些场景下，我们并不想追加，而是希望直接覆盖旧值

这时就需要 Overwrite。

为什么需要 Overwrite

可以想象一个聊天系统：

• 平时的消息流转，都是不断往消息列表后面追加内容
• 但如果用户点击“清空上下文并重新开始”，这时就不应该继续 append
• 你要的不是“合并”，而是“整段替换”

也就是说：

• reducer 负责“怎么合”
• Overwrite 负责“这次别合了，直接替换”

代码示例

from langgraph.graph import StateGraph, START, END
from langgraph.types import Overwrite
from typing_extensions import Annotated, TypedDict
import operator

class State(TypedDict):
messages: Annotated[list, operator.add]

def add_message(state: State):
return {"messages": ["first message"]}

def replace_messages(state: State):
# 绕过 reducer，直接替换整个 messages 列表
return {"messages": Overwrite(["replacement message"])}

builder = StateGraph(State)
builder.add_node("add_message", add_message)
builder.add_node("replace_messages", replace_messages)

builder.add_edge(START, "add_message")
builder.add_edge("add_message", "replace_messages")
builder.add_edge("replace_messages", END)

graph = builder.compile()

result = graph.invoke({"messages": []})
print(result)

运行结果说明

如果没有 Overwrite，最终的 messages 往往会变成：

["first message", "replacement message"]

但用了 Overwrite 之后，最终结果会变成：

["replacement message"]

也就是直接覆盖原状态，而不是走 operator.add 的追加逻辑。

适用场景

• 清空聊天记录并重新开始
• 重置工作流上下文
• 覆盖缓存字段
• 重新生成某一段中间结果
• 异常恢复时替换错误状态

实战提醒

Overwrite 很好用，但不要滥用。

因为一旦使用覆盖语义，意味着你跳过了原本的状态合并规则。适合“明确要重置”的场景，不适合日常状态流转。

一句话总结：

默认用 reducer 合并，只有在确实要“整段替换”时再用 Overwrite。

5.2 使用独立的输入 / 输出模式

默认情况下，LangGraph 使用单一状态模式，也就是：

• 输入是这份状态
• 节点内部流转是这份状态
• 最终输出还是这份状态

这在小 Demo 里很方便，但真实业务里往往不够优雅。

因为我们通常希望：

• 输入模式：只暴露用户需要传入的字段
• 输出模式：只返回用户真正关心的结果
• 内部状态：保存节点间流转所需的完整信息

换句话说，内部状态不一定应该完全暴露给外部。

为什么要区分输入、输出和内部状态

以一个问答系统为例：

• 输入：用户的问题 question
• 输出：模型答案 answer
• 内部：可能还会有 retrieved_docs、tool_calls、rewrite_count 等中间字段

这些内部字段对系统很重要，但对调用方并不友好，也没必要都返回。

这时就可以使用 StateGraph 的这几个参数：

• input_schema
• output_schema
• state_schema

参数含义

代码示例

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# 1. 输入模式：只接收 question
class InputState(TypedDict):
question: str

# 2. 输出模式：只返回 answer
class OutputState(TypedDict):
answer: str

# 3. 内部完整状态：同时包含输入和输出
class OverallState(InputState, OutputState):
pass

def answer_node(state: InputState):
"""理解用户问题并生成答案"""
return {
"question": state["question"],
"answer": f"Answer to: {state['question']}",
}

builder = StateGraph(
OverallState,
input_schema=InputState,
output_schema=OutputState,
)

builder.add_node("answer_node", answer_node)
builder.add_edge(START, "answer_node")
builder.add_edge("answer_node", END)

graph = builder.compile()

result = graph.invoke({"question": "What is LangGraph?"})
print(result)

结果说明

输出会是：

{"answer": "Answer to: What is LangGraph?"}

注意一点：

• question 虽然存在于内部状态中
• 但因为 output_schema=OutputState
• 所以它不会出现在最终返回结果里

这个设计的价值

这类模式在工程里非常有用，尤其适合下面几类场景：

• API 开发：输入输出契约清晰
• 微服务：避免内部字段直接外泄
• 数据管道：明确每一步的边界
• Agent 系统：内部状态可以复杂，外部接口保持简洁

实战建议

如果图已经开始出现这些字段：

• debug_info
• intermediate_result
• trace_id
• tool_result
• sensitive_data

那就说明很可能已经不适合继续用“单一状态模式”了。

这个时候，尽快拆出：

• 对外输入模式
• 对外输出模式
• 内部完整状态

后面维护成本会低很多。

5.3 在节点间传递私有状态

在复杂图里，经常会有一种需求：

• 某些中间数据对后续节点很重要
• 但这些数据不应该出现在最终输出里
• 甚至不应该被所有节点看到

这就是“私有状态”的问题。

一个典型例子

假设要做一个报告生成流程：

这里有个核心要求：

• 节点 1 和 节点 2 之间需要共享敏感数据
• 但 节点 3 不应该看到原始敏感数据
• 最终输出更不应该包含这些内容

这时就可以通过节点特定输入 / 输出类型来实现局部传递。

代码示例

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# 公共状态：最终对外可见
class OverallState(TypedDict):
final_result: str

# 节点1的私有输出
class Node1Output(TypedDict):
sensitive_data: str

# 节点2需要的输入
class Node2Input(TypedDict):
sensitive_data: str

def node_1(state: OverallState) –> Node1Output:
"""第一步：获取包含敏感信息的原始数据"""
private_data = "这是敏感信息"
print("Node1: 获取到敏感数据，但不会暴露给最终输出")
return {"sensitive_data": private_data}

def node_2(state: Node2Input) –> OverallState:
"""第二步：处理敏感数据，输出清洗结果"""
print(f"Node2: 正在处理敏感数据: {state['sensitive_data']}")
return {"final_result": "清理后的处理结果"}

def node_3(state: OverallState) –> OverallState:
"""第三步：只消费公开结果，不接触敏感信息"""
print(f"Node3: 只看到最终结果: {state['final_result']}")
return {"final_result": state["final_result"] + " – 完成"}

builder = StateGraph(OverallState)

# 这里顺序执行三个节点
builder.add_sequence([node_1, node_2, node_3])
builder.add_edge(START, "node_1")

graph = builder.compile()

response = graph.invoke({"final_result": "initial"})
print(response)

一次执行后，你会看到什么

Node1: 获取到敏感数据，但不会暴露给最终输出
Node2: 正在处理敏感数据: 这是敏感信息
Node3: 只看到最终结果: 清理后的处理结果

最终输出: {'final_result': '清理后的处理结果 – 完成'}

从这个过程可以看出：

• sensitive_data 在节点 1 和节点 2 之间传递成功
• 它没有进入最终输出
• 节点 3 只看到清洗后的结果

这就是私有状态传递的价值。

这种设计适合什么场景

• 数据处理：中间步骤的临时字段
• 认证流程：token、验证码、密钥片段
• 金融或订单系统：原始明细与脱敏结果分离
• Agent 工具链：工具原始返回和最终摘要分离
• 错误处理：内部错误详情与对外错误消息分离

这里最容易误解的一点

很多人会以为：

“只要字段不在最终 output_schema 中，就算私有状态了。”

这不完全对。

因为“最终不输出”和“只在特定节点间传递”是两个层次的问题：

• output_schema 解决的是“对外暴露什么”
• 节点专属输入输出类型，解决的是“哪些节点能看到什么”

也就是说：

• 不返回给外部，不等于不在内部传播
• 真正的私有状态，需要你在节点间显式控制可见范围

三个能力放在一起，怎么理解最清楚

你可以把这三者理解成三个不同层面的控制能力：

1. Overwrite

解决的是：

状态更新时，到底是合并还是覆盖？

2. input_schema / output_schema / state_schema

解决的是：

图的输入边界、输出边界、内部状态边界怎么划分？

3. 节点间私有状态

解决的是：

某些中间数据应该在谁和谁之间流动，不应该被谁看到？

如果把它们组合起来，你对 LangGraph 的状态管理就已经不再停留在“能跑”的层面，而是进入“可控、可维护、可上线”的层面。

实战里怎么选

如果你在写 LangGraph 工作流，可以按这个判断顺序来：

场景 1：字段需要继续累积

选 reducer，比如：

• operator.add
• add_messages

适用于：

• 消息历史
• 日志列表
• 检索结果集合

场景 2：字段需要整段替换

选 Overwrite

适用于：

• 重置上下文
• 覆盖缓存
• 清空后重新开始

场景 3：对外接口和内部状态不一致

拆分：

• input_schema
• output_schema
• state_schema

适用于：

• API 服务
• 复杂 Agent 编排
• 内部字段较多的工作流

场景 4：中间敏感数据只允许局部流转

使用节点特定的输入 / 输出状态定义

适用于：

• 敏感字段处理
• 脱敏链路
• 内部审核与对外结果分离

常见坑点

坑 1：以为所有状态更新都应该走 append

并不是。

有些字段天然就是“覆盖型”，这时继续 append 只会让状态越来越脏。

坑 2：把内部中间字段直接暴露给输出

这在 Demo 里看起来无所谓，但在正式接口里会带来两个问题：

• 返回结构混乱
• 泄露内部实现细节

坑 3：把“最终不输出”和“节点不可见”混为一谈

前者是输出层控制。

后者是节点级状态可见性控制。

这两个不是一回事。

坑 4：状态设计过早贪大

很多人一开始就把所有字段都堆进一个大 TypedDict 里，后面越改越难。

更好的方式是：

• 先定义最小公共状态
• 再按节点职责拆出局部状态
• 最后用输入 / 输出模式划边界

总结

LangGraph 的强大，不只体现在“能把节点连起来”，更体现在它对状态流转的控制力。

这篇讲的三个点，本质上对应三个非常关键的问题：

• Overwrite：控制更新方式
• 输入 / 输出模式：控制对外边界
• 私有状态：控制内部可见性

如果想把 LangGraph 从 Demo 写到真实项目，这三个能力迟早都要补上。

最后用一句话收尾：

图结构决定流程，状态设计决定系统质量。