LastValue:覆盖写通道
昨天读了"合同"BaseChannel。今天读第一个、也是最常用的"员工"——LastValue:只保留最后写入的那一个值,新值覆盖旧值。你不写 reducer 的每个普通字段,编译后默认就是它。我们会看到它怎么用 MISSING 哨兵表示"空"、为什么"一步只准收一个值否则报错",以及它靠 finish() 变身的兄弟 LastValueAfterFinish。
你没写 reducer 的字段,默认就是 LastValue
class State(TypedDict): user_id: str,从没提过 channel,它背后到底是什么?
答案是 LastValue。编译时若一个字段没有任何 Annotated reducer,LangGraph 就兜底给它一个 LastValue 通道(graph/state.py:1857):# graph/state.py:1857 —— 字段解析的最后一步:没有 channel、没有 reducer 就兜底
fallback: LastValue = LastValue(annotation)
fallback.key = name
return fallback
所以 LastValue 是整个框架里出场率最高的通道。看它的类定义(channels/last_value.py:20-25):
class LastValue(Generic[Value], BaseChannel[Value, Value, Value]): # last_value.py:20
"""Stores the last value received, can receive at most one value per step."""
__slots__ = ("value",) # last_value.py:23
value: Value | Any
BaseChannel[Value, Value, Value]三个泛型全一样!写进来什么类型、读出去、存档就是什么类型。这是 LastValue "简单"的直接体现——它不做任何"碎片→整体"的变换。对照昨天 Topic 的三者不同,这里三者相同。__slots__ = ("value",)在父类的 key/typ 之外,只加一个属性 value——存"最后那个值"。子类自己声明 slots,延续昨天说的省内存约定。can receive at most one value per stepdocstring 就点明了铁规矩:每步最多收一个值。这是 L03/L04 要展开的重点。用 MISSING 哨兵区分"空"和"值是 None"
昨天讲过:通道要严格区分"从没写过"和"写了个 None"。LastValue 用 MISSING 哨兵实现这点。看构造和读取(last_value.py:27-29, 69-75):
def __init__(self, typ: Any, key: str = "") -> None: # last_value.py:27
super().__init__(typ, key)
self.value = MISSING # ← 初始状态:空。不是 None,是 MISSING
def get(self) -> Value: # last_value.py:69
if self.value is MISSING:
raise EmptyChannelError() # ← 从没写过 → 抛异常,不返回 None
return self.value
def is_available(self) -> bool: # last_value.py:74
return self.value is not MISSING # ← 重写:直接比哨兵,不走 try/except
self.value = MISSING刚建好的通道,value 是那个全局唯一哨兵对象 MISSING = object()(_internal/_typing.py:45),表示"这里还没值"。get 里 is MISSING → 抛异常读到哨兵就抛 EmptyChannelError。注意用 is 不用 ==——哨兵是靠"身份唯一"识别的,不是靠值相等。is_available 重写父类默认要 try get() 抓异常(慢);LastValue 重写成一句 is not MISSING——高频调度路径上这点提速很值。正是昨天 base 里 docstring 建议"子类重写"的落地。None 是合法业务值(你可能真想存 user=None 表示"未登录")。若拿 None 当"空",就分不清"没值"和"值为空"。哨兵 MISSING = object() 是一个不可能被业务用到的独一无二对象,拿它当"空"标记,永不与任何真实值撞车。这就是"哨兵值(sentinel)"模式。LastValue(str) → value=MISSING → is_available()==False、get() 抛异常。写入
update([None]) 后 → value=None → is_available()==True、get() 返回 None。看到区别没?同样"看起来是空",前者是"根本没值"、后者是"明确写了 None",两种状态被清清楚楚分开。
update:覆盖写的三行核心
LastValue 的 update 是全框架最短的 reducer 之一,但每一行都有讲究(last_value.py:56-67):
def update(self, values: Sequence[Value]) -> bool: # last_value.py:56
if len(values) == 0:
return False # ① 空序列:没人写,我没变
if len(values) != 1: # ② 一步收到 >1 个值 → 违规
msg = create_error_message(
message=f"At key '{self.key}': Can receive only one value per step. "
f"Use an Annotated key to handle multiple values.",
error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE,
)
raise InvalidUpdateError(msg)
self.value = values[-1] # ③ 正好一个:覆盖写进去
return True
① len==0 → return False呼应昨天的边界:引擎每步会对所有通道调 update,没人写时传空序列。LastValue 遇空什么都不做、返回 False(我没变化,别推进版本号)。② len != 1 → 抛异常核心规矩:一步只准收一个值。收到两个及以上直接抛 InvalidUpdateError,还贴心提示"想收多个值请用 Annotated reducer"。③ self.value = values[-1]正常路径:把唯一那个值存进去,覆盖旧的。返回 True(我变了 → 推进版本号 → 唤醒下游)。用 values[-1] 而非 values[0] 是习惯写法,长度已保证为 1。self.value = 新值,旧值被 GC 回收。对比明天的累加通道 self.value = self.operator(self.value, 新值)(要和旧值运算),你就秒懂"覆盖 vs 累加"的代码差异其实只有一行。为什么"一步多写"要报错,而不是随便留一个
answer 字段,为什么不干脆留最后一个、非要报错?values 顺序没有意义。如果 LastValue 悄悄"留 values[-1]",那结果就取决于不可控的执行时序——同样的图、同样的输入,今天留 A 的、明天留 B 的,成了不可复现的玄学 bug。与其埋雷,不如当场大声报错:告诉你"这个字段被并发写了,你得明确指定合并规则(reducer)"。Send 做 map-reduce 扇出(Day 16),多个并行分支都写回同一个没配 reducer 的字段,就会撞上这个 InvalidUpdateError。解法正是错误消息里那句提示:把该字段改成 Annotated[list, operator.add] 之类,换成一个"能合并多值"的通道(明天的 BinaryOperatorAggregate)。所以这个报错不是刁难,是在逼你为并发写想清楚合并语义。存档、还原与 copy 的高效重写
LastValue 的存档三件套简单直接(last_value.py:44-54, 77-78):
def copy(self) -> Self: # last_value.py:44 —— 重写父类的两步式 copy
empty = self.__class__(self.typ, self.key)
empty.value = self.value
return empty
def from_checkpoint(self, checkpoint: Value) -> Self: # last_value.py:50
empty = self.__class__(self.typ, self.key)
if checkpoint is not MISSING: # ← 存档是 MISSING 说明当时是空的
empty.value = checkpoint
return empty
def checkpoint(self) -> Value: # last_value.py:77
return self.value # ← 直接返回值(可能是 MISSING)
checkpoint 重写父类默认是"try get()、空则 MISSING"。LastValue 干脆直接 return self.value——反正空时 value 本来就是 MISSING,省掉一次异常处理。from_checkpoint造个新通道,只有存档不是 MISSING 时才把值填进去。存档是 MISSING → 新通道保持空。这就是"从存档精确复活",空的还原成空、有值的还原成有值。copy 重写父类 copy 默认是 from_checkpoint(checkpoint()) 两步;LastValue 重写成"新建 + 直接拷 value 字段"一步到位——正是昨天说的"子类可重写成更高效实现"。它的兄弟:LastValueAfterFinish(憋到收尾才放值)
同一个文件里还有个变种 LastValueAfterFinish(last_value.py:81)——它演示了昨天那对"冷门钩子" finish()/consume() 怎么用。docstring 一句话:"存最后的值,但只在 finish() 后才对外可见;一旦被读走就清空"。看它多出来的字段和四个关键方法:
__slots__ = ("value", "finished") # last_value.py:87 —— 比 LastValue 多一个 finished 标志
def update(self, values): # last_value.py:122
if len(values) == 0:
return False
self.finished = False # ← 一有新写入,就退回"未收尾"
self.value = values[-1]
return True
def finish(self) -> bool: # last_value.py:138
if not self.finished and self.value is not MISSING:
self.finished = True # ← 收尾信号:把值"解锁"
return True
return False
def get(self) -> Value: # last_value.py:145
if self.value is MISSING or not self.finished:
raise EmptyChannelError() # ← 没收尾之前,一律视为"空",读不到!
return self.value
def consume(self) -> bool: # last_value.py:130
if self.finished:
self.finished = False
self.value = MISSING # ← 被读走后清空,实现"一次性"
return True
return False
多一个 finished 标志核心区别:值存进来后不马上可见,要等 finished=True。它是个"上锁的信箱"。get 里 or not self.finished只要还没 finish,get 就抛空异常——值在里面但对外装作没有。这就是"延迟到收尾才可见"。finish() 解锁引擎判定图快跑完时调它(昨天 _algo.py:338),把 finished 置真,值这才真正可读。consume() 清空值被读走后,consume 把它清回 MISSING——一次性语义,读完即焚,不会被重复消费。update 重置 finished只要来了新写入就 finished=False——重新上锁,等下一次 finish 再放。保证放出去的永远是"收尾那一刻的终值"。谁在用 LastValue + 小结
除了"无 reducer 字段的兜底"(L01),LastValue 还有个隐藏大客户:条件边的路由通道。看编译期给分支建的通道(graph/state.py:1512-1516 附近):
# graph/state.py:1514 附近 —— 分支/路由用的临时通道
LastValueAfterFinish(Any)
if ... # 收尾语义的分支
else EphemeralValue(Any, guard=False) # 普通分支用瞬时通道(Day 31)
| 场景 | 用哪种 | 为什么 |
|---|---|---|
| 普通字段(无 reducer) | LastValue | 覆盖写,最朴素,够用 |
| 并发写同一字段 | ❌ LastValue 会报错 | 改用累加/Topic 通道(Day 29/30) |
| 只想收尾时暴露终值 | LastValueAfterFinish | 过程中捂住,finish 才可见 |
👶 小白:既然 LastValue 这么简单,为什么不干脆内置成"字典赋值",还搞个类?
👨🏫 老师:因为要和别的通道共享同一套接口(昨天的 BaseChannel)。引擎的 apply_writes 只认 update/get/checkpoint,不认字典。把 LastValue 也做成通道,引擎就能一视同仁地对待所有字段——不用为"简单字段"和"累加字段"写两套代码。多态的价值就在这:加一种新通道,引擎一行都不用改。
🧠 今日小结自测
- 什么字段默认用 LastValue?(没配任何 Annotated reducer 的字段,state.py:1857 兜底)
- 它三个泛型为什么全相同?(写=读=存,不做碎片到整体的变换)
- 怎么区分"空"和"值为 None"?(value 初始为 MISSING 哨兵,get 读到 MISSING 抛 EmptyChannelError)
- 为什么一步收到多个值要报错而不是留最后一个?(顺序不保证,隐式留一个会导致不可复现,逼你显式配 reducer)
- LastValue 为什么重写 copy/checkpoint/is_available?(这些是热路径,重写成直接操作字段省开销)
- LastValueAfterFinish 比 LastValue 多什么?(多 finished 锁:写完捂住、finish 解锁、consume 读完即焚)
✋ 10 分钟动手
# 1. 通读 LastValue 与它的兄弟(整文件 152 行)
sed -n '1,152p' libs/langgraph/langgraph/channels/last_value.py
# 2. 看编译时"无 reducer 字段兜底 LastValue"
sed -n '1855,1862p' libs/langgraph/langgraph/graph/state.py
# 3. 亲手触发"一步多写"报错:用 Send 扇出多个分支写同一个无 reducer 字段,看 InvalidUpdateError
# 4. 对比:给该字段加 Annotated[list, operator.add] 后报错消失
operator.add 就能实现 messages 越攒越多)。我们会深挖它怎么处理 Overwrite 重置、为什么 reducer 必须满足结合律,再看进阶的 DeltaChannel——内存里存完整值、存档却只写哨兵靠"回放历史写入"重建,为超长运行省存储。