关键认知：invoke 底层就是 stream 收集完。流式不是附加功能，而是引擎的原生工作方式——invoke 只是"把流收集成最终值"的特例。

所有模式定义在一个 Literal 里 types.py:120：

# types.py:120
StreamMode = Literal[
    "values", "updates", "checkpoints", "tasks", "debug", "messages", "custom"
]
"""How the stream method should emit outputs.
- "values":   每步后 emit 整个 state（含 interrupt）
- "updates":  只 emit 节点名 + 该节点返回的增量更新
- "custom":   节点内用 StreamWriter 手动 emit 的自定义数据
- "messages": LLM 消息 token-by-token + metadata
- "checkpoints": 每次生成检查点时 emit（格式同 get_state()）
- "tasks":    任务开始/结束事件（含结果和错误）
- "debug":    emit checkpoints + tasks（调试用）
"""

不管哪种模式，塞进流的都是同一种结构——一个三元组 protocol.py:272：

# protocol.py:272
StreamChunk = tuple[tuple[str, ...], str, Any]
#              (   namespace   ,  mode ,  data)

# protocol.py:275  流本身是一个"可调用 + 带 modes 集合"的对象
class StreamProtocol:
    __slots__ = ("modes", "__call__")
    modes: set[StreamMode]
    __call__: Callable[[Self, StreamChunk], None]
    def __init__(self, __call__, modes):
        self.__call__ = __call__
        self.modes = modes

核心在 PregelLoop 的 _emit（_loop.py:1380）——所有模式都经过这道闸：

# _loop.py:1380
def _emit(self, mode, values, *args, **kwargs):
    if self.stream is None:
        return
    debug_remap = mode in ("checkpoints", "tasks") and "debug" in self.stream.modes
    if mode not in self.stream.modes and not debug_remap:
        return                                            # ★ 用户没要这个模式 → 直接不干活
    for v in values(*args, **kwargs):                     # values 是个"惰性生成"函数
        if mode in self.stream.modes:
            self.stream((self.checkpoint_ns, mode, v))    # ★ 推进流：(ns, mode, data)
        if debug_remap:
            self.stream((self.checkpoint_ns, "debug", {...}))  # L05

谁来调 _emit？以 updates 为例，在 output_writes（_loop.py:1415）里：

# _loop.py:1441 （节点写完后）
if "updates" in stream_modes:
    self._emit("updates", lambda: iter(interrupts))       # 中断也当 update 吐
if "values" in stream_modes:
    current_values = read_channels(self.channels, self.output_keys)
    ...
    self._emit("values", lambda: iter([current_values]))  # 读当前全部通道值 → 吐整个状态
# :1450 普通写（非中断）
self._emit("updates", map_output_updates, self.output_keys, [(task, writes)], cached)

回看 _emit 里的 debug_remap（_loop.py:1389）：

# _loop.py:1389
debug_remap = mode in ("checkpoints", "tasks") and "debug" in self.stream.modes
...
if debug_remap:
    self.stream((self.checkpoint_ns, "debug", {
        "step": self.step - 1 if mode == "checkpoints" else self.step,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "type": "checkpoint" if mode == "checkpoints"
                else "task_result" if "result" in v else "task",
        "payload": v,                                     # ★ 原样包住 checkpoints/tasks 数据
    }))

tasks 数据本身由 map_debug_tasks/map_debug_task_results 生成（debug.py:41、debug.py:106），checkpoints 由 map_debug_checkpoint（debug.py:144）生成——它们才是"原始数据源"，debug 只是转包。

values/updates/debug 是引擎在超步循环里主动吐的。但 custom 和 messages 数据源在节点内部，得用不同机制接。看 stream() 的准备段（main.py:2810）：

# main.py:2810  messages 模式：挂一个回调 handler
if "messages" in stream_modes:
    ns_ = config[CONF].get(CONFIG_KEY_CHECKPOINT_NS)
    messages_handler_cls = (StreamMessagesHandlerV2 if use_stream_messages_v2
                            else StreamMessagesHandler)
    run_manager.inheritable_handlers.append(
        messages_handler_cls(stream.put, subgraphs,       # ← 把 stream.put 交给 handler
            parent_ns=tuple(ns_.split(NS_SEP)) if ns_ else None))

# main.py:2841  custom 模式：给节点一个 stream_writer
if "custom" in stream_modes:
    def stream_writer(c):
        stream.put((tuple(get_config()[CONF][CONFIG_KEY_CHECKPOINT_NS].split(NS_SEP)[:-1]),
                    "custom", c))                          # ← 把 c 包成 custom chunk

你可以一次订多个模式。stream() 开头把参数归一成集合（main.py:2571）：

# main.py:2571
if isinstance(stream_mode, str):
    stream_modes = {stream_mode}          # 单个字符串 → 单元素集合
else:
    stream_modes = set(stream_mode)       # 传 list → 集合
...

✋ 10 分钟动手

# 1. 核心 emit 逻辑逐行读
sed -n '1380,1462p' libs/langgraph/langgraph/pregel/_loop.py     # _emit + output_writes
sed -n '2841,2860p' libs/langgraph/langgraph/pregel/main.py      # custom writer

# 2. 亲手对比五种模式
python - <<'PY'
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class S(TypedDict):
    n: int
g = StateGraph(S)
g.add_node("a", lambda s: {"n": s["n"]+1})
g.add_edge(START,"a"); g.add_edge("a",END)
app = g.compile()
print("== values =="); [print(c) for c in app.stream({"n":0}, stream_mode="values")]
print("== updates =="); [print(c) for c in app.stream({"n":0}, stream_mode="updates")]
print("== 多模式 =="); [print(c) for c in app.stream({"n":0}, stream_mode=["updates","debug"])]
PY