Agent Memory 的架构：LangGraph 真正的工作原理

Source: Dev.to

理解 LangGraph 的 State（状态）

State 是 LangGraph 代理的执行记忆。它记录每一次输入、内部思考、工具输出以及在遍历图时的决策。每个节点接收当前的 state，执行自己的逻辑，并 仅 返回它想要更新的 state 部分。LangGraph 会根据你提供的 schema 将这些更新合并到已有的 state 中。这种确定性的合并可以追加、覆盖或组合数据，取决于 state 的定义方式。

使用 TypedDict 的简单 State 定义

from typing_extensions import TypedDict

class State(TypedDict):
    messages: list
    extra_field: int

TypedDict 定义了在图中流动的记忆结构。除了对话历史，state 还能保存工具输出、元数据、任务状态、计数器、检索到的文档等信息。

为什么 TypedDict 优于 Pydantic 或 Dataclasses

• 部分更新：节点可以返回只包含需要修改键的 dict。LangGraph 会自动合并这些部分更新。
• 性能：更新单个字段不需要重新构造整个对象，对大规模历史记录来说成本更低。
• 不可变性保证：LangGraph 依赖确定性的合并；对完整的 Pydantic 或 dataclass 实例进行变更可能破坏这些保证。
• 缓存问题：Pydantic 的内部元数据在不同实例化之间可能会变化，导致缓存行为异常。

基于这些权衡，生产环境中的大多数 LangGraph 图都使用 TypedDict 来定义 state schema。

Reducers：控制更新的合并方式

Reducers 定义如何将已有值与新值合并。其签名为：

def reducer(current_value, new_value) -> merged_value:
    ...

• 默认行为：如果未指定 reducer，新的值会覆盖旧的值。
• 自定义行为：使用 reducers 可以向列表追加、累加计数器或合并字典。

使用 Python 的 operator.add

from typing import Annotated
from operator import add
from typing_extensions import TypedDict

class State(TypedDict):
    history: Annotated[list[str], add]
    count: int

add reducer 会连接列表，因此对 history 的多次更新会被追加而不是覆盖。

LangChain 消息的内置 add_messages Reducer

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

add_messages 能智能合并 LangChain 的 HumanMessage、AIMessage 和 ToolMessage 对象，依据 ID 去重，并在流式或工具调用期间确保正确的顺序。

实际示例

from operator import add
from typing import Annotated
from typing_extensions import TypedDict

class MyState(TypedDict):
    logs: Annotated[list[str], add]
    counter: Annotated[int, add]

节点实现

def start_node(state: MyState):
    return {"logs": ["Started"], "counter": 1}

def step_node(state: MyState):
    return {"logs": ["Step done"], "counter": 2}

def finish_node(state: MyState):
    return {"logs": ["Finished"], "counter": 3}

执行流程

1. start_node 之后 – logs = ["Started"]，counter = 1
2. step_node 之后 – logs = ["Started", "Step done"]，counter = 3
3. finish_node 之后 – logs = ["Started", "Step done", "Finished"]，counter = 6

Reducers 使日志能够追加，计数器能够累加；如果没有 reducers，每次更新都会覆盖之前的状态。

编写自定义 Reducers

LangGraph 接受任何符合 reducer 签名的可调用对象。自定义 reducers 可以：

• 当列表过大时裁剪旧的对话上下文。
• 合并字典并保留已有键。
• 只累加不同的项目。
• 实现针对特定业务的复杂转换逻辑。

def unique_items(current: list, new: list) -> list:
    return list(dict.fromkeys(current + new))

class CustomState(TypedDict):
    items: Annotated[list, unique_items]

在此示例中，unique_items 确保 items 列表在每次合并后不包含重复项。