回忆 Day 20 主循环：第一步之前有个 _first() 把输入喂进图，用的就是 map_input；每步 after_tick 里 _emit("values", map_output_values, ...) 产出流式输出，用的就是 map_output_values。今天把这两道门拆开。

进门映射 map_input（_io.py:81）非常直白：

# _io.py:81
def map_input(input_channels, chunk) -> Iterator[tuple[str, Any]]:
    """Map input chunk to a sequence of pending writes in the form (channel, value)."""
    if chunk is None:
        return                                       # 87 输入是 None → 不写任何东西（恢复场景）
    elif isinstance(input_channels, str):
        yield (input_channels, chunk)                # 89 单输入通道：整个 chunk 写进去
    else:
        if not isinstance(chunk, dict):
            raise TypeError(f"Expected chunk to be a dict, got {type(chunk).__name__}")  # 92
        for k in chunk:                              # 93 多输入通道：按 key 分发
            if k in input_channels:
                yield (k, chunk[k])                  # 95 key 是合法输入通道 → 写它
            else:
                logger.warning(f"Input channel {k} not found in {input_channels}")  # 97

出门要先会"读通道"。两个基础函数 read_channel/read_channels（_io.py:23-53）：

# _io.py:23
def read_channel(channels, chan, *, catch=True) -> Any:
    try:
        return channels[chan].get()                  # 30 调通道的 get() 取当前值
    except EmptyChannelError:                         # 31 通道还没被写过 → 空
        if catch:
            return None                               # 33 默认吞掉，返回 None
        else:
            raise                                     # 35 catch=False 时把空当错误抛

# _io.py:38
def read_channels(channels, select, *, skip_empty=True) -> dict | Any:
    if isinstance(select, str):
        return read_channel(channels, select)         # 45 单个：直接读
    else:
        values: dict[str, Any] = {}
        for k in select:                              # 48 多个：逐个读拼成 dict
            try:
                values[k] = read_channel(channels, k, catch=not skip_empty)
            except EmptyChannelError:
                pass                                  # 52 skip_empty=True 时跳过空通道
        return values

values 模式的输出映射（_io.py:100）——每步产出"整个状态现在长啥样"：

# _io.py:100
def map_output_values(output_channels, pending_writes, channels) -> Iterator[dict | Any]:
    """Map pending writes (a sequence of tuples (channel, value)) to output chunk."""
    if isinstance(output_channels, str):
        if pending_writes is True or any(
            chan == output_channels for chan, _ in pending_writes  # 108 本步是否写了这个输出通道
        ):
            yield read_channel(channels, output_channels)          # 110 写了 → 读它的全量值产出
    else:
        if pending_writes is True or {
            c for c, _ in pending_writes if c in output_channels   # 112 本步写的通道里有输出通道吗
        }:
            yield read_channels(channels, output_channels)         # 115 有 → 读全部输出通道拼 dict

updates 模式（_io.py:118）反过来——只产出"这一步哪个节点改了什么"，按节点名分组：

# _io.py:118
def map_output_updates(output_channels, tasks, cached=False) -> Iterator[dict]:
    output_tasks = [
        (t, ww) for t, ww in tasks
        if (not t.config or TAG_HIDDEN not in t.config.get("tags", EMPTY_SEQ))
        and ww[0][0] != ERROR and ww[0][0] != INTERRUPT   # 129 过滤隐藏/报错/中断的任务
    ]
    if not output_tasks:
        return                                            # 132 这步没有可输出的任务
    updated: list[tuple[str, Any]] = []
    for task, writes in output_tasks:
        rtn = next((value for chan, value in writes if chan == RETURN), MISSING)  # 135
        if rtn is not MISSING:
            updated.append((task.name, rtn))              # 137 节点用 RETURN 显式指定返回值
        elif isinstance(output_channels, str):
            updated.extend((task.name, value) for chan, value in writes if chan == output_channels)
        elif any(chan in output_channels for chan, _ in writes):   # 142 该任务写了输出通道
            ...把该任务写的输出通道打包成 {通道: 值} ...
    grouped: dict[str, Any] = {t.name: [] for t, _ in output_tasks}  # 164 按节点名分组
    for node, value in updated:
        grouped[node].append(value)
    ...
    yield grouped                                         # 174 产出 {节点名: 该节点这步的更新}

除了普通 dict，图还能接受 Command 作为输入（人在环恢复、跨节点跳转）。它由 map_command（_io.py:56）翻译成写入：

# _io.py:56
def map_command(cmd: Command) -> Iterator[tuple[str, str, Any]]:
    if cmd.graph == Command.PARENT:
        raise InvalidUpdateError("There is no parent graph")   # 59 顶层图没有父图
    if cmd.goto:                                               # 60 要跳到某节点/发 Send
        sends = cmd.goto if isinstance(cmd.goto, (tuple, list)) else [cmd.goto]
        for send in sends:
            if isinstance(send, Send):
                yield (NULL_TASK_ID, TASKS, send)              # 67 Send → 写 TASKS（Day 23 那招）
            elif isinstance(send, str):
                yield (NULL_TASK_ID, f"branch:to:{send}", START)  # 69 跳到节点名 → 写它的入边通道
    if cmd.resume is not None:
        yield (NULL_TASK_ID, RESUME, cmd.resume)               # 74 恢复值 → 写 RESUME 通道
    if cmd.update:
        for k, v in cmd._update_as_tuples():
            yield (NULL_TASK_ID, k, v)                          # 77 状态更新 → 逐字段写

✋ 10 分钟动手

# 1. 读 _io.py 全貌（就 175 行，值得通读）
sed -n '1,175p' libs/langgraph/langgraph/pregel/_io.py

# 2. 对比 values 与 updates 两种流
python -c "
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: {'count': s['count']+1})
g.add_node('b', lambda s: {'count': s['count']+10})
g.add_edge(START,'a'); g.add_edge('a','b'); g.add_edge('b',END)
app = g.compile()
print('== values =='); [print(c) for c in app.stream({'count':0}, stream_mode='values')]
print('== updates =='); [print(c) for c in app.stream({'count':0}, stream_mode='updates')]
"

# 3. 试试多写一个 schema 外字段，观察 warning
python -c "
import logging; logging.basicConfig(level=logging.WARNING)
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class S(TypedDict): x: int
g=StateGraph(S); g.add_node('n', lambda s:{'x':1}); g.add_edge(START,'n'); g.add_edge('n',END)
print(g.compile().invoke({'x':0,'ghost':9}))   # ghost 被忽略
"