建 LangGraph 的第一件事不是写节点，而是划分 State。

State 就像一块公共笔记板：prompt、历史消息、中间结果、计数器……所有节点需要共享的数据都放在这里。字段又叫 Channel。

搞懂 State 的读写规则和 Reducer 的合并逻辑，后面写复杂 Agent 才不会乱。

State 的三层视角

LangGraph 允许你为不同节点定义不同的 State 视角，但底层框架为了注入和高可用，实际是把所有 State 混合在一起的。

类型 作用
InputState 约束 invoke() 的入参，调用时只填这里定义的字段
OutputState 约束 invoke() 的返回值，result = graph.invoke(...) 只能拿到这些字段
PrivateState 内部字段，不用传给 StateGraph()，框架会自动映射

几个容易误解的点：

  • 节点 return 时，不需要严格按 OutputState 来写——只要 return 的 key 在底层混合 State 里存在，框架就会自动更新
  • PrivateState 只是声明，用到了框架会自动处理，不用手动传
  • 真正影响对外接口的是 InputState 和 OutputState

用一个例子说明 InputState 和 OutputState 的边界：

1
2
3
4
5
6
7
8
9
10
11
12
13
class InputState(TypedDict):
    messages: list          # 调用者只需要传 messages

class OutputState(TypedDict):
    final_answer: str       # 调用者只能拿到 final_answer

class InternalState(TypedDict):
    messages: list
    llm_calls: int          # 内部计数，不对外暴露
    debug_trace: list       # 调试信息，不对外暴露
    final_answer: str

graph = StateGraph(InternalState, input=InputState, output=OutputState)

调用时：

1
2
3
result = graph.invoke({"messages": [HumanMessage("你好")]})

print(result)  # {"final_answer": "..."}

中间过程的 llm_calls、debug_trace 对外不可见，但节点内部可以自由读写。这就是 Input/Output State 的价值——控制 Agent 的对外接口。

Reducer：字段怎么更新？

默认情况下，节点 return 某个字段会直接覆盖旧值。但很多场景需要合并而不是替换——比如多个子任务各自返回一份分析结果，父图要汇总成列表。

Reducer 就是决定「这个字段怎么更新」的函数：

1
2
3
4
5
def merge_analyses(left: list, right: list):
    return left + right

class State(TypedDict):
    analyses: Annotated[list, merge_analyses]

节点 return {"analyses": [new_item]} 时，框架会自动调用 merge_analyses(old, [new_item])，而不是直接覆盖。

常见坑：父图要汇总多个子图结果（比如多份 RCA 分析），但没写 Reducer → 后写的顶掉先写的，汇总就「失效」了。遇到列表聚合，第一反应应该是加 Reducer。

LangChain 内置了 operator.add 用于列表追加，简单场景直接用：

1
2
3
4
from operator import add

class State(TypedDict):
    items: Annotated[list, add]

Reducer 的工作原理

Annotated 的写法是 Annotated[字段类型, reducer函数]。节点 return 时，框架自动拿旧值和新值调用 reducer：

1
2
3
旧值 = state["analyses"]        # 比如 [A, B]
新值 = return 里的值             # 比如 [C]
结果 = reducer([A, B], [C])     # → [A, B, C]

没指定 reducer 的字段，return 会直接覆盖旧值。写图之前对每个字段想一句：这个字段是替换还是合并？ 合并就必须加 reducer。

Message：LLM 应用最常见的 State 字段

Message 在 LLM 应用里太常见了，框架单独做了优化。

追加 vs 覆盖

  • 历史消息一般追加即可
  • 人工介入修改某条消息时，需要覆盖原来的——带上消息的 id，框架会用新内容替换旧内容而不是追加

框架提供了 add_messages Reducer 来处理这两种情况。如果自己管理 Message，需要手动指定；用 MessagesState 则自动适配，开箱即用。

MessagesState 是 LangGraph 最常用的 State 基类，Agent 项目几乎都会用到：

1
2
3
4
5
6
7
8
9
10
11
from langgraph.graph import MessagesState

class AgentState(MessagesState):
    llm_calls: int          # 在 MessagesState 基础上加自定义字段

def llm_call(state: AgentState):
    response = model.invoke(state["messages"])
    return {
        "messages": [response],
        "llm_calls": state.get("llm_calls", 0) + 1,
    }

继承 MessagesState 后，messages 字段自动带 add_messages reducer——追加、覆盖都帮你处理好了。

人工介入：覆盖某条消息

Human-in-the-loop 场景下，用户可能修改 AI 的某条回复再让 Agent 继续。带上消息的 id 就能覆盖而不是追加：

1
return {"messages": [AIMessage(content="修改后的内容", id="msg_123")]}
1
2
3
{"messages": [HumanMessage(content="你好")]}

{"messages": [{"type": "human", "content": "你好"}]}

流式传输的安全问题

流式返回时，框架为了方便会暴露所有用过的 State 字段——这意味着敏感中间数据可能在传输过程中泄露。

两种防护手段：

1
2
3
stream_events(..., output_keys=["symptoms", "final_report"])

graph.stream(..., stream_mode="updates")

生产环境如果 Agent 处理的是用户隐私或内部数据，这一点必须检查。

两种 stream 模式对比

模式 行为 适用场景
stream_mode="values" 每步返回 State 全量快照 调试、UI 展示完整状态
stream_mode="updates" 每步只返回本次节点的增量 生产环境、减少数据暴露

推荐生产环境默认用 updates，调试时切换 values 看全量。

Node 与 State 的关系

Node 是一个同步或异步的 Python 函数，接收 config、State、Runtime 等参数。

几个实操注意点：

1. Checkpoint 重试要幂等

用了 checkpointer 后，节点可能被执行多次。如果节点里有写数据库、发请求等副作用，必须保证幂等，否则重试会出脏数据。

2. @task 和 interrupt() 的 replay 机制

一旦在 node 里用了 @task 或 interrupt()，恢复执行不是从断点那一行继续，而是从头 replay + 用缓存填结果。

所以代码顺序必须稳定，否则 replay 时会拿错缓存。

3. END 节点

所有执行路径的最终节点都应该是 END。漏了 END，图可能跑不完或行为不确定。

节点缓存

LangGraph 支持在编译时给节点加缓存，参数由两部分组成：

  • key_func：组织缓存键的字符串（默认字符串相等即命中）
  • 持续时间：缓存过期时间

框架没有提供自定义命中策略的入口——输入字符串一样就算命中。如果需要更精细的缓存控制，得在节点内部自己实现。

实战：SRE Agent 里的 State 设计

用一个真实场景来理解 State 和 Reducer 的配合——SRE 故障诊断 Agent：

  1. Supervisor 节点拆出多个故障主题 → 写入 subjects: list[str]
  2. Send 并行派多个 RCA Worker → 每个返回一份 rca_analysis
  3. 父图 Reducer 把多份分析合并到 rca_analyses_list

如果第 3 步没写 Reducer，后返回的 Worker 会直接覆盖前面的结果，汇总就「失效」了——这是实际项目里最常见的 State 设计失误。

1
2
3
4
5
6
7
def merge_rca_analyses(left: list, right: list):
    return left + right

class ParentState(TypedDict):
    subjects: list[str]
    rca_analyses_list: Annotated[list, merge_rca_analyses]
    final_report: str

实战：设计 State 的检查清单

开始写图之前，先回答这几个问题：

  1. 哪些数据要在节点间共享？→ 放进 State
  2. 哪些字段需要合并而不是覆盖？→ 加 Reducer
  3. 哪些字段只给内部用？→ PrivateState
  4. 对外 invoke() 只需要暴露什么？→ OutputState
  5. 流式输出会暴露哪些字段？→ 检查敏感数据
  6. 有没有 checkpoint 重试？→ 副作用操作要幂等