D03 见过 __init__ 一上来就调 _add_schema(state_schema)。它的完整逻辑：

    def _add_schema(self, schema, /, allow_managed=True):
        if schema not in self.schemas:                       # 同一个 schema 只解析一次
            _warn_invalid_state_schema(schema)               # ① 校验是不是合法 schema
            channels, managed, type_hints = _get_channels(schema)   # ② 核心翻译
            if managed and not allow_managed:                # ③ input/output 不许有托管字段
                raise ValueError(f"Invalid managed channels ...")
            self.schemas[schema] = {**channels, **managed}   # ④ 缓存这个 schema 的解析结果
            for key, channel in channels.items():
                if key in self.channels:
                    if self.channels[key] != channel:
                        if isinstance(channel, LastValue):
                            pass                             # 同名但新的是 LastValue → 容忍
                        else:
                            raise ValueError(f"Channel '{key}' already exists ...")
                else:
                    self.channels[key] = channel             # ⑤ 汇总进全局 channels

翻译机的主函数。它先处理"没有字段的裸类型"，再逐字段调 _get_channel：

def _get_channels(schema):
    if not hasattr(schema, "__annotations__"):               # 没字段（如裸 int/list）
        return ({"__root__": _get_channel("__root__", schema, allow_managed=False)}, {}, {})

    type_hints = get_type_hints(schema, include_extras=True)  # ★带上 Annotated 的元数据
    all_keys = {
        name: _get_channel(name, typ)                        # 每个字段翻译成一个通道
        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,
    )

单个字段的通道类型，靠这个"四选一"函数拍板：

def _get_channel(name, annotation, *, allow_managed=True):
    # 先剥掉 Required / NotRequired 外壳
    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(...)
    elif channel := _is_field_channel(annotation):             # ② 标注里直接给了通道对象吗？
        channel.key = name
        return channel
    elif channel := _is_field_binop(annotation):               # ③ 标注里给了 reducer 函数吗？
        channel.key = name
        return channel

    fallback = LastValue(annotation)                           # ④ 都不是 → 默认覆盖写
    fallback.key = name
    return fallback

最精妙的一段——它怎么判断 Annotated 最后那个东西是不是一个合法 reducer？答案是数它接受几个位置参数：

def _is_field_binop(typ):
    if hasattr(typ, "__metadata__"):            # 是 Annotated 才有 __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（直接给通道对象/类的情况）：

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

State schema 不止 TypedDict。建图时有个校验函数会提醒你别传错东西：

def _warn_invalid_state_schema(schema):
    if isinstance(schema, type):        # 是个类 → OK
        return
    if typing.get_args(schema):         # 是 Annotated[...] 之类带参数的 → OK
        return
    warnings.warn(
        f"Invalid state_schema: {schema}. Expected a type or Annotated[type, reducer]. ...")

✋ 10 分钟动手

# 1. 看解析入口 _add_schema
sed -n '342,372p' libs/langgraph/langgraph/graph/state.py

# 2. 看主翻译 _get_channels（注意 include_extras=True）
sed -n '1801,1821p' libs/langgraph/langgraph/graph/state.py

# 3. 看四选一 _get_channel + reducer 嗅探
sed -n '1836,1908p' libs/langgraph/langgraph/graph/state.py

# 4. 亲眼看不同字段变成不同通道
python3 -c "
from langgraph.graph import StateGraph
from typing_extensions import TypedDict, Annotated
import operator
class S(TypedDict):
    x: int
    msgs: Annotated[list, operator.add]
g = StateGraph(S)
for k,v in g.channels.items(): print(k, '->', type(v).__name__)
"