一个 PregelNode 本质是三段 Runnable 串起来的链条：

先认识它的核心字段（_read.py:97-126）：

# _read.py:97
class PregelNode:
    channels: str | list[str]      # 102 读哪些通道当输入（str=单值；list=拼成 dict）
    triggers: list[str]            # 107 哪些通道被写就触发我（Day 21 _triggers 用的就是它）
    mapper: Callable | None        # 111 读到输入后先转换一下（可选）
    writers: list[Runnable]        # 114 跑完后执行的写器们（把输出写回通道）
    bound: Runnable[Any, Any]      # 118 节点的主逻辑 = 你的函数
    retry_policy: Sequence[RetryPolicy] | None   # 122 重试策略（Day 25）
    cache_policy: CachePolicy | None             # 125 缓存策略（Day 51）

node 是个 cached_property，把 bound 和 writers 拼成一个可执行的 Runnable（_read.py:221-234）：

# _read.py:221
@cached_property
def node(self) -> Runnable[Any, Any] | None:
    """Get a runnable that combines `bound` and `writers`."""
    writers = self.flat_writers                 # 224 先做写器合并优化（L04）
    if self.bound is DEFAULT_BOUND and not writers:
        return None                             # 226 既没逻辑也没写器 → 空节点，不跑
    elif self.bound is DEFAULT_BOUND and len(writers) == 1:
        return writers[0]                       # 228 纯写：就一个写器，直接用
    elif self.bound is DEFAULT_BOUND:
        return RunnableSeq(*writers)            # 230 纯写：多个写器串起来
    elif writers:
        return RunnableSeq(self.bound, *writers)  # 232 常态：你的函数 + 写器们，串成链
    else:
        return self.bound                       # 234 只有逻辑没写器：就跑你的函数

组装前有个小优化 flat_writers（_read.py:204-219）：把连续的 ChannelWrite 合并：

# _read.py:204
@cached_property
def flat_writers(self) -> list[Runnable]:
    """Get writers with optimizations applied. Dedupes consecutive ChannelWrites."""
    writers = self.writers.copy()
    while (len(writers) > 1
           and isinstance(writers[-1], ChannelWrite)
           and isinstance(writers[-2], ChannelWrite)):   # 211 末尾两个都是 ChannelWrite
        writers[-2] = ChannelWrite(                      # 215 合并成一个
            writes=writers[-2].writes + writers[-1].writes,   # 216 写入项拼起来
        )
        writers.pop()                                    # 218 去掉被合并的那个
    return writers

读这一段由 ChannelRead（_read.py:25）完成。它自己不直接碰 channel，而是从 config 里取一个"读函数"（_read.py:73-91）：

# _read.py:73
@staticmethod
def do_read(config, *, select, fresh=False, mapper=None) -> Any:
    try:
        read: READ_TYPE = config[CONF][CONFIG_KEY_READ]   # 82 从 config 取"读函数"
    except KeyError:
        raise RuntimeError("Not configured with a read function ...")  # 84 不在 Pregel 上下文里
    if mapper:
        return mapper(read(select, fresh))                # 89 读完再转换一下
    else:
        return read(select, fresh)                        # 91 直接读

对称地，ChannelWrite（_write.py:46）负责写。它的 do_write 先校验再调 config 里的发送函数（_write.py:105-126）：

# _write.py:105
@staticmethod
def do_write(config, writes, allow_passthrough=True) -> None:
    for w in writes:                                # 112 逐条校验
        if isinstance(w, ChannelWriteEntry):
            if w.channel == TASKS:                  # 114 禁止直接写保留通道 TASKS
                raise InvalidUpdateError("Cannot write to the reserved channel TASKS")
            if w.value is PASSTHROUGH and not allow_passthrough:
                raise InvalidUpdateError("PASSTHROUGH value must be replaced")  # 119
        ...
    write: TYPE_SEND = config[CONF][CONFIG_KEY_SEND]   # 125 从 config 取"发送函数"
    write(_assemble_writes(writes))                    # 126 组装成 (通道,值) 元组列表后发出

写入的最后一步 _assemble_writes（_write.py:172）把各种写入项标准化成 (通道, 值) 元组。Send 在这里被特殊处理：

# _write.py:172
def _assemble_writes(writes) -> list[tuple[str, Any]]:
    tuples: list[tuple[str, Any]] = []
    for w in writes:
        if isinstance(w, Send):
            tuples.append((TASKS, w))               # 179 Send → 写进 TASKS 通道！
        elif isinstance(w, ChannelWriteTupleEntry):
            if ww := w.mapper(w.value):             # 181 一个映射产出多条写入
                tuples.extend(ww)
        elif isinstance(w, ChannelWriteEntry):
            value = w.mapper(w.value) if w.mapper is not None else w.value
            if value is SKIP_WRITE:                 # 185 mapper 返回 SKIP_WRITE → 跳过不写
                continue
            if w.skip_none and value is None:       # 187 skip_none 且值为 None → 跳过
                continue
            tuples.append((w.channel, value))       # 189 普通写：(通道名, 值)
        else:
            raise ValueError(f"Invalid write entry: {w}")
    return tuples

✋ 10 分钟动手

# 1. 读 PregelNode 字段 + node 组装
sed -n '97,234p' libs/langgraph/langgraph/pregel/_read.py

# 2. 读 ChannelWrite.do_write（校验 + 发送）
sed -n '105,126p' libs/langgraph/langgraph/pregel/_write.py

# 3. 读 _assemble_writes（Send→TASKS 那行）
sed -n '172,192p' libs/langgraph/langgraph/pregel/_write.py

# 4. 亲手触发 unknown channel 警告（故意拼错字段）
python -c "
import logging; logging.basicConfig(level=logging.WARNING)
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class S(TypedDict): count: int
g = StateGraph(S)
g.add_node('a', lambda s: {'cont': 1})   # 拼错 count→cont
g.add_edge(START,'a'); g.add_edge('a',END)
print(g.compile().invoke({'count':0}))    # 观察 warning + count 没变
"