zwhy
2026 年5 月 12 日 12:25
1
from typing import TypedDict, Annotated
import operator
from langgraph.graph.state import StateGraph, START, END
def remain_min(old: int, new: int) → int:
print("reducer remain_min is called", f"old: {old}, new: {new}")
if old is None:
return new
if new is None:
return old
return min(old, new)
class State(TypedDict):
min_score: Annotated[int, remain_min]
builder = StateGraph(State)
builder.add_edge(START, END)
graph =builder.compile()
initial_state = {“min_score”: 60}
print(graph.invoke(initial_state)) 问ai都说只有return时才会调用reducer进行更新参数,但是我在invoke传入自定义state时也调用reducer更新了参数,似乎min_score被默认初始化为了0然后我传入的被当成了更新,导致min_score一直是0
admin
2026 年5 月 17 日 01:36
2
你的观察是对的,AI 给你的解释错了。
LangGraph 里 reducer 的触发规则是:任何对 channel 的写入都会过 reducer ,包括 invoke 时传入的 initial input。input dict 本质上就是第一次"写入",并不会绕过 reducer 直接赋值。
至于为什么 old 是 0 而不是 None:带 reducer 的字段在底层是 BinaryOperatorAggregate channel,初始值用的是注解类型的零值——int 就是 0,list 就是 [],dict 就是 {}。所以你的流程实际上是:
channel 初始化 → min_score = 0
invoke 传入 {"min_score": 60} → 触发 reducer,old=0, new=60
min(0, 60) = 0 → 最终就是 0
所以 reducer 调用本身没问题,问题是 int 的零值刚好是 min 运算的吸收元,导致永远卡在 0。
几种改法,看你的语义需求:
方法 1:reducer 里识别"未初始化"状态
把字段类型改成 int | None,然后 reducer 里把 0 也当 sentinel 处理。但 int | None 的零值是什么取决于 LangGraph 版本,不太可靠。
方法 2:用一个不会冲突的 sentinel,比如 float('inf')
def remain_min(old, new):
if old is None: return new
if new is None: return old
return min(old, new)
class State(TypedDict):
min_score: Annotated[float, remain_min] # 用 float
然后初始化时显式传 float('inf'),或者:
方法 3(推荐):reducer 里用业务语义判断
如果你的 min_score 是分数(0–100),那 0 本来就是合法值,sentinel 走不通。这种情况干脆把字段包一层:
def remain_min(old, new):
if not old: # 0 / None / 空 都视为未初始化
return new
if new is None:
return old
return min(old, new)
但要确认业务上 min_score=0 是不是有效值。如果是有效值(比如真的有人得 0 分),这种写法就有 bug。
方法 4:用 langgraph.channels.LastValue 风格 + 自己管理
直接不用 reducer,在 node 内部维护 min 逻辑。如果你只是想保留运行期间的最小值,其实在 node 里 state["min_score"] = min(state["min_score"], new_score) 也行。reducer 更适合多个 node 并发写入需要 merge 的场景(比如 fan-out/fan-in)。