LastValue:覆盖式写入的默认通道

源码版本1.2.9

职责

LastValue 是 BaseChannel 最简单的子类:它存一个值,每次写就把旧值覆盖掉(LastValue class:20-21)。它就是「state 字段不加任何 Annotated 标注时默认用的通道」,你在 StateGraph 里写 class State(TypedDict): count: int,编译期看到的 count 字段背后就是一个 LastValue(int)(fallback LastValue:1857-1859)。

它的关键约束是「每步只能写一个值」。update 拿到长度大于 1 的序列时不合并也不覆盖,而是直接抛 InvalidUpdateError(update:56-67),提示用户该用 Annotated[int, reducer] 表达「这一步里多个节点同时写要怎么 fold」。这个约束不是技术限制,而是语义选择:标量状态本来就没有合理的合并方式,与其默默丢值,不如显式报错。

另外这个文件还放了一个变体 LastValueAfterFinish(LastValueAfterFinish:81-152):行为像 LastValue,但写进去的值要等整个 Pregel 运行 finish() 之后才对外可见,见一次就清空。它用于「中断信号」「结束信号」这类只在运行末尾生效的内部通道。

设计动机

• 覆盖是标量状态唯一合理的默认:整数、字符串、布尔这些值没有「合并」语义。count: int 这一类字段,新值来了直接覆盖旧值是最符合直觉的行为,所以 StateGraph 在 fallback:1857 里就把没标注 reducer 的字段编译成 LastValue。
• 拒绝静默丢值:如果同一超步里两个节点同时写 count,LastValue 不会随便挑一个赢,而是抛 INVALID_CONCURRENT_GRAPH_UPDATE(InvalidUpdateError:60-64)。错误信息明确提示「Use an Annotated key to handle multiple values」,把用户推向正确的解法——加一个 reducer,见 /channels/topic-binop。
• MISSING 而不是 None:初始值用 langgraph._internal._typing.MISSING(value = MISSING:29),这样 None 是合法的存值。get() 在 MISSING 时抛 EmptyChannelError(get:69-72)。这套和 BaseChannel 默认实现保持一致,但 is_available 在这里被重写成一行 self.value is not MISSING(is_available:74-75),避免每次都走 try/except。
• copy 比 from_checkpoint 便宜:虽然父类默认 copy = from_checkpoint(checkpoint()),这里重写了 copy(copy:44-48),直接共享同一个 value 引用——因为标量本来也不需要深拷贝。from_checkpoint(from_checkpoint:50-54)则按 MISSING 分支决定要不要赋值。
• LastValueAfterFinish 的延迟可见性:有些控制信号(比如 Command 的 goto)在写的时候不能立刻被 prepare_next_tasks 看到,否则会在本步内自我触发形成环。LastValueAfterFinish 用一个 finished 标志位(finished flag:93-95),get() 只在 finished=True 时返回(get gated:145-148),finish() 由 Pregel 在收尾时调(finish:138-143),consume() 在被消费一次后清空(consume:130-136)。

关键文件

• LastValue class:20-25 — 类定义,泛型 BaseChannel[Value, Value, Value],即存值类型、写值类型、checkpoint 类型都是同一个 Value。
• __init__:27-29 — 构造时 value = MISSING,只有被写过才不是 MISSING。
• ValueType / UpdateType:34-42 — 两者都返回 self.typ,即声明里给的那个类型。
• copy:44-48 — 直接复用 value 引用,标量不需要深拷贝。
• from_checkpoint:50-54 — 从 checkpoint 值重建实例,MISSING 表示从空开始。
• update:56-67 — 核心约束:空序列返回 False,长度大于 1 抛 InvalidUpdateError,否则存最后一个值并返回 True。
• get / is_available:69-75 — 读接口,MISSING 抛 EmptyChannelError;is_available 重写成一行判断。
• LastValueAfterFinish:81-95 — 变体:加 finished 标志,延迟到 finish() 后可见。
• LastValueAfterFinish.update:122-128 — 写时顺手把 finished 复位,意味着新写会撤掉上一次的可见性。
• consume / finish / get:130-148 — 三个钩子配合实现「见一次就清空」的语义。
• fallback LastValue:1857-1859 — StateGraph 编译时遇到没标注 reducer 的字段就退回 LastValue(annotation)。

数据流

LastValue.update 的全部逻辑就这几行(update:56-67):

def update(self, values: Sequence[Value]) -> bool:
    if len(values) == 0:
        return False
    if len(values) != 1:
        msg = create_error_message(
            message=f"At key '{self.key}': Can receive only one value per step. Use an Annotated key to handle multiple values.",
            error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE,
        )
        raise InvalidUpdateError(msg)

    self.value = values[-1]
    return True

注意第三分支写的是 values[-1] 而不是 values[0]——因为前面已经判了 len != 1,这一行只会走到 len == 1 的情况,[-1] 和 [0] 等价。空序列返回 False 是关键:它配合 apply_writes 里对未变更通道的 update(EMPTY_SEQ) 调用(EMPTY_SEQ update:329),让 LastValue 在被空调用时正确返回「未改动」。

StateGraph 编译时怎么把没标注的字段映射成 LastValue,这段在 field_to_channel:1840-1859:

if manager := _is_field_managed_value(name, annotation):
    if allow_managed:
        return manager
    else:
        raise ValueError(f"This {annotation} not allowed in this position")
elif channel := _is_field_channel(annotation):
    channel.key = name
    return channel
elif channel := _is_field_binop(annotation):
    channel.key = name
    return channel

fallback: LastValue = LastValue(annotation)
fallback.key = name
return fallback

按顺序尝试:managed value(如 MessagesState 里的特殊管理)→ 显式 BaseChannel 实例 → Annotated[..., reducer](BinaryOperatorAggregate,见 /channels/topic-binop)→ 都不命中,退回 LastValue。

整个读写时序如下:

边界与失败

• 多值并发写直接报错:multi-value error:59-64 抛 InvalidUpdateError,错误码 INVALID_CONCURRENT_GRAPH_UPDATE。这是初学者最常见的坑:两个节点同时往同一个标量字段写。修法是改成 Annotated[int, lambda a, b: a + b] 这种带 reducer 的字段。
• 空写返回 False:empty seq:57-58 返回 False,和 BaseChannel 契约一致,apply_writes 据此跳过 channel_versions 更新。
• __eq__ 不看值只看类型:__eq__:31-32 返回 isinstance(value, LastValue),意味着两个 LastValue 实例无论存什么都被认为相等。这是给图编译期去重用的,不要拿它做值比较。
• MISSING 与 None:None 是合法存值;只有 MISSING 才表示「从未写过」(get / is_available:70-75)。如果你的业务字段允许 None,放心写,通道不会把它当空。
• LastValueAfterFinish 的新写会撤掉可见性:LAF.update:122-128 一开头 self.finished = False。也就是说如果 finish() 之后又来一个写,通道会重新进入「未公开」状态,要等下一次 finish() 才能读。
• LastValueAfterFinish.consume 在 finished 为假时是 no-op:见 LAF.consume:130-136。这意味着「写完但还没 finish」的值不会被清掉,只有在被消费过一次后才清空——这正是它「结束信号」的语义所需。

小结

LastValue 是 LangGraph 状态的默认形状——存一个标量,覆盖式更新,每步单值约束在写多值时报错而不是默默丢值。理解它就理解了所有没标注 reducer 的字段在运行时是怎么变化的。要看并发写怎么正确合并,继续读 /channels/topic-binop;通道接口的定义回 /channels/base-channel;具体什么时候被 update 见 /pregel/algo。

对照官方资料:LangGraph 文档 · README。