它的类头 _messages.py:49：

# _messages.py:49
class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
    """A callback handler that implements stream_mode=messages.
    Collects messages from:
    (1) chat model stream events; and
    (2) node outputs.
    """
    run_inline = True
    """We want this callback to run in the main thread to avoid order/locking issues."""

构造函数 _messages.py:60，家当很少：

# _messages.py:60
def __init__(self, stream, subgraphs, *, parent_ns=None):
    self.stream = stream                 # ← D57 传进来的 stream.put，往流里推数据
    self.subgraphs = subgraphs           # 是否也收子图的消息
    self.metadata: dict[UUID, Meta] = {} # run_id → (命名空间, 元数据)
    self.seen: set[int | str] = set()    # 已经吐过的 message id（去重用）
    self.parent_ns = parent_ns           # 本 handler 创建时所在的命名空间

模型一开始调用，先触发这个（_messages.py:130）：

# _messages.py:130
def on_chat_model_start(self, serialized, messages, *, run_id, ..., tags=None, metadata=None, **kwargs):
    if metadata and (not tags or (TAG_NOSTREAM not in tags)):     # 没被标记"不要流"
        ns = tuple(metadata["langgraph_checkpoint_ns"].split(NS_SEP))[:-1]  # 算命名空间
        if not self.subgraphs and len(ns) > 0 and ns != self.parent_ns:
            return                                                # 不收子图 → 跳过
        if (filtered_tags := filter_to_user_tags(tags)) is not None:
            metadata["tags"] = filtered_tags
        self.metadata[run_id] = (ns, metadata)                    # ★ 登记 run_id → (ns, meta)

今天的绝对核心——每来一个 token 触发（_messages.py:151）：

# _messages.py:151
def on_llm_new_token(self, token, *, chunk=None, run_id, ..., **kwargs):
    if not isinstance(chunk, ChatGenerationChunk):
        return                                       # 不是聊天模型的 chunk → 忽略
    if meta := self.metadata.get(run_id):            # ★ 查户口：登记过才处理
        self._emit(meta, chunk.message)              # 把这一小片消息推进流

推流动作 _emit（_messages.py:97）：

# _messages.py:97
def _emit(self, meta, message, *, dedupe=False):
    if dedupe and message.id in self.seen:
        return                                       # 去重（L05）
    else:
        if message.id is None:
            message.id = str(uuid4())                # 没 id 就补一个
        self.seen.add(message.id)
        self.stream((meta[0], "messages", (message, meta[1])))  # ★ (ns, "messages", (消息, 元数据))

靠 seen 集合 + dedupe=True。看 on_llm_end（_messages.py:166）和节点结束 on_chain_end（_messages.py:223）：

# _messages.py:166  模型结束：完整消息 dedupe=True
def on_llm_end(self, response, *, run_id, **kwargs):
    if meta := self.metadata.get(run_id):
        if response.generations and response.generations[0]:
            gen = response.generations[0][0]
            if isinstance(gen, ChatGeneration):
                self._emit(meta, gen.message, dedupe=True)   # ★ 去重吐完整消息
    self.metadata.pop(run_id, None)                          # 用完清户口

# _messages.py:223  节点结束：节点输出里的消息也 dedupe
def on_chain_end(self, response, *, run_id, **kwargs):
    if meta := self.metadata.pop(run_id, None):
        ... self._find_and_emit_messages(meta, response)     # 内部 _emit(dedupe=True)

节点可能返回各种形状：一条消息、消息列表、或 {"messages":[...]} 状态。_find_and_emit_messages（_messages.py:106）负责挖出所有消息：

# _messages.py:106
def _find_and_emit_messages(self, meta, response):
    if isinstance(response, BaseMessage):
        self._emit(meta, response, dedupe=True)              # 直接是一条消息
    elif isinstance(response, Sequence):
        for value in response:
            if isinstance(value, BaseMessage):
                self._emit(meta, value, dedupe=True)         # 消息列表
    else:
        for value in _state_values(response):                # 状态 dict：遍历各字段值
            if isinstance(value, BaseMessage):
                self._emit(meta, value, dedupe=True)
            elif isinstance(value, Sequence):
                for item in value:
                    if isinstance(item, BaseMessage):
                        self._emit(meta, item, dedupe=True)  # 字段值是消息列表

新版模型支持"内容块（content-block）"事件（如推理块、工具调用块分开流）。StreamMessagesHandlerV2（_messages.py:259）用 on_stream_event（_messages.py:373）转发：

# _messages.py:373
def on_stream_event(self, event, *, run_id, ..., **kwargs):
    if meta := self.metadata.get(run_id):
        if event.get("event") == "message-start":
            self._streamed_run_ids.add(run_id)
            msg_id = event.get("message_id")
            if msg_id:
                self.seen.add(msg_id)                # 提前记 id，让 on_chain_end 去重
        v2_meta = {**meta[1], "run_id": str(run_id)}
        self.stream((meta[0], "messages", (event, v2_meta)))  # 把事件本身推进流

✋ 10 分钟动手

# 1. 核心回调逐行读
sed -n '97,165p'  libs/langgraph/langgraph/pregel/_messages.py    # _emit + start + token
sed -n '166,257p' libs/langgraph/langgraph/pregel/_messages.py    # end + 去重 + 节点消息

# 2. 亲手看逐 token（需真实模型 key）
python - <<'PY'
from langgraph.prebuilt import create_react_agent
agent = create_react_agent("openai:gpt-4o-mini", [])
for msg, meta in agent.stream(
        {"messages":[("user","用一句话介绍北京")]},
        stream_mode="messages"):
    print(repr(msg.content), "| node=", meta.get("langgraph_node"))  # 一片一片打印
PY