如果你用 LangGraph 构建过多节点的 Agent，一定接触过这样的状态定义：

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
    messages: Annotated[list, add_messages]
    current_tool: str
    step_count: int

Annotated[list, add_messages] 这个写法乍看有点奇怪——为什么不直接用 list[AnyMessage]？背后的 add_messages 是什么意思？这篇文章就来拆解 LangGraph 状态系统的核心设计：Channel。

为什么不用共享可变字典

最直觉的做法是给每个图维护一个全局字典，节点执行完后把输出写进去。这在单节点顺序执行时没问题，但 LangGraph 的核心价值之一是并发——多个节点可以在同一个超步（superstep）内并行运行。一旦并发，共享可变字典至少带来三个问题。

竞态问题。假设 tool_a 和 tool_b 并行执行，都往 state["result"] 写入，最后谁写谁赢，取决于线程调度顺序，结果不可预期。你可以加锁，但这等于串行化了并发，得不偿失。

序列化问题。LangGraph 的核心特性之一是 checkpoint——每个超步执行完后将状态持久化到存储，以便恢复和重放。如果状态是一个普通 dict，checkpoint 层只能存一个"最终值的快照"。当两个节点并发写入同一个 key 时，checkpoint 不知道这两次写入该如何合并——是取最后一个？还是拼接？还是求和？这个语义必须由某处来定义。

调度问题。Pregel 式执行引擎在每个超步结束后需要决定下一轮触发哪些节点：只有"在本超步有新写入"的 channel 才会触发订阅了它的节点。普通 dict 没有版本信息，引擎无法判断某个 key 在这一轮是否真的被更新了。

这三个问题加在一起，指向同一个结论：状态中的每个字段不能只是一个裸值，它需要封装自己的合并逻辑、版本追踪和序列化/反序列化能力。这就是引入 Channel 的原因。

BaseChannel 接口

BaseChannel 定义在 langgraph/channels/base.py，是一个泛型抽象基类，三个类型参数分别是 Value（读取时的类型）、Update（写入时的类型）和 Checkpoint（持久化时的类型）。接口中最关键的两个方法是 update() 和 checkpoint()。

update() 的签名是 update(self, values: Sequence[Update]) -> bool。注意它接收的是一个列表，而不是单个值。这是设计上最重要的一个决策：同一超步内可能有多个节点同时写入同一个 channel，框架在超步结束时收集所有写入，一次性调用 update()，由 channel 自身决定如何合并这批写入。返回值 bool 表示本次调用是否真正改变了 channel 的内部状态——引擎靠这个决定是否触发下游节点。

checkpoint() 返回一个可序列化的表示，from_checkpoint() 从这个表示重建 channel 实例。两者合起来构成了 channel 的持久化协议。

版本追踪隐含在 update() 返回的布尔值里：每次 update() 返回 True，引擎就认为这个 channel 的版本发生了变化，会触发订阅它的节点在下一超步执行。版本不比较值本身，只看"有没有被更新"，这让不同类型的 channel 可以自由定义"更新"的语义。

五种 Channel 深度解析

下图展示了六种 Channel 的继承关系和各自的核心语义：

classDiagram
    class BaseChannel {
        +update(values: list) bool
        +get() Value
        +checkpoint() Any
        +from_checkpoint(data)
        +version : int|str|float
    }
    class LastValue {
        单写覆盖语义
        多写直接抛异常
    }
    class BinaryOperatorAggregate {
        reducer(a, b) → a
        多写折叠合并
        e.g. add_messages
    }
    class DeltaChannel {
        增量存储 diff
        父链重建全量值
        快照截断父链
    }
    class NamedBarrierValue {
        等待全部指定写入
        fan-in 同步原语
    }
    class EphemeralValue {
        超步后自动清空
        不写入 checkpoint
    }
    class UntrackedValue {
        写入不递增版本
        不触发节点调度
    }
    BaseChannel <|-- LastValue
    BaseChannel <|-- BinaryOperatorAggregate
    BaseChannel <|-- DeltaChannel
    BaseChannel <|-- NamedBarrierValue
    BaseChannel <|-- EphemeralValue
    BaseChannel <|-- UntrackedValue

LastValue：最简单的单值槽

LastValue 是默认 channel 类型，语义直白：每次写入覆盖旧值，最多只接受一次写入。

# last_value.py 核心逻辑
def update(self, values: Sequence[Value]) -> bool:
    if len(values) == 0:
        return False
    if len(values) != 1:
        raise InvalidUpdateError(
            f"At key '{self.key}': Can receive only one value per step. "
            "Use an Annotated key to handle multiple values."
        )
    self.value = values[-1]
    return True