官方 docstring 里就给了一个最小 reducer 的例子（graph/state.py:167）：

# graph/state.py:167 —— StateGraph 类文档里的示例 reducer
def reducer(a: list, b: int | None) -> list:
    if b is not None:
        return a + [b]
    return a


class State(TypedDict):
    x: Annotated[list, reducer]

Annotated 是 Python 标准库 typing 提供的"给类型贴便利贴"的工具。Annotated[list, reducer] 的意思是：这个字段的类型是 list，另外还附带一条元数据 reducer。类型检查器只看第一个 list，而 reducer 这条元数据"搭便车"存进了一个叫 __metadata__ 的属性里，等 LangGraph 来取。

编译图时，LangGraph 要把你的 State 类翻译成一组 channel。入口函数是 _get_channels()（graph/state.py:1801）：

# graph/state.py:1801
def _get_channels(schema):
    if not hasattr(schema, "__annotations__"):        # ① schema 不是"带字段的类"
        return (
            {"__root__": _get_channel("__root__", schema, allow_managed=False)},
            {}, {},
        )

    type_hints = get_type_hints(schema, include_extras=True)   # ② 读出所有字段的类型注解
    all_keys = {
        name: _get_channel(name, typ)                 # ③ 每个字段各算一个 channel
        for name, typ in type_hints.items()
        if name != "__slots__"
    }
    return (
        {k: v for k, v in all_keys.items() if isinstance(v, BaseChannel)},   # 真正的通道
        {k: v for k, v in all_keys.items() if is_managed_value(v)},          # 托管值（后面章节）
        type_hints,
    )

单个字段怎么变成通道？_get_channel()（graph/state.py:1836）是一条清晰的"四级分派"链：

# graph/state.py:1836
def _get_channel(name, annotation, *, allow_managed=True):
    # 先剥掉 Required / NotRequired 外壳（TypedDict 的可选标记）
    if hasattr(annotation, "__origin__") and annotation.__origin__ in (Required, NotRequired):
        annotation = annotation.__args__[0]

    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？
        channel.key = name
        return channel
    elif channel := _is_field_binop(annotation):               # ③ 便利贴里是个 reducer 函数？
        channel.key = name
        return channel

    fallback: LastValue = LastValue(annotation)                # ④ 什么都没写 → 默认 LastValue
    fallback.key = name
    return fallback

分支③是最常用的。它怎么判断"便利贴里那个东西是不是一个 reducer"？看 _is_field_binop()（graph/state.py:1890）：

# graph/state.py:1890
def _is_field_binop(typ):
    if hasattr(typ, "__metadata__"):                 # 有便利贴吗？
        meta = typ.__metadata__
        if len(meta) >= 1 and callable(meta[-1]):     # 最后一张便利贴是"可调用的"吗？
            sig = signature(meta[-1])
            params = list(sig.parameters.values())
            if (
                sum(
                    p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD)
                    for p in params
                ) == 2                                # ★必须恰好接收 2 个位置参数
            ):
                return BinaryOperatorAggregate(typ, meta[-1])
            else:
                raise ValueError(
                    f"Invalid reducer signature. Expected (a, b) -> c. Got {sig}"
                )
    return None

顺带看分支②的判断 _is_field_channel()（graph/state.py:1862），它认的是"便利贴里直接放了通道实例或通道类"：

# graph/state.py:1862（节选）
def _is_field_channel(typ):
    if hasattr(typ, "__metadata__"):
        for item in typ.__metadata__:
            if isinstance(item, BaseChannel):          # 便利贴是"通道实例"
                return item
            elif isclass(item) and issubclass(item, BaseChannel):  # 便利贴是"通道类"
                return item(typ.__origin__ if hasattr(typ, "__origin__") else typ)
    return None

分支④兜底给的是 LastValue。它的合并逻辑（update）解释了"为什么不写 reducer 的字段是覆盖语义、而且并行写会报错"（channels/last_value.py:56）：

# channels/last_value.py:56
def update(self, values):
    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. "
                    "Use an Annotated key to handle multiple values.",
            error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE,
        )
        raise InvalidUpdateError(msg)
    self.value = values[-1]                            # 只有 1 个值 → 直接覆盖
    return True

最后看一个防御性边界。如果你传给 StateGraph 的 schema 既不是类、也不是 Annotated[...]，框架不会硬崩，而是发一条警告（graph/state.py:111）：

# graph/state.py:111
def _warn_invalid_state_schema(schema):
    if isinstance(schema, type):          # 是个类（TypedDict/dataclass/pydantic）→ OK
        return
    if typing.get_args(schema):           # 是 Annotated[...] 之类带参数的 → 也 OK
        return
    warnings.warn(
        f"Invalid state_schema: {schema}. Expected a type or Annotated[type, reducer]. "
        "Please provide a valid schema to ensure correct updates.\n ..."
    )

# 1. 读 reducer 读取的三个核心函数
sed -n '1801,1908p' libs/langgraph/langgraph/graph/state.py

# 2. 读 LastValue 的覆盖 + 并行报错逻辑
sed -n '56,67p' libs/langgraph/langgraph/channels/last_value.py

# 3. 亲手验证 Annotated 的结构
python -c "from typing import Annotated; t=Annotated[list,'x']; print(t.__origin__, t.__metadata__)"