两种手段互补：

• debug 流（debug.py）：运行时观测。每步产出两类事件——task（这步要跑谁）和 task_result/checkpoint（这步跑完的结果和存档）。
• 画图（_draw.py）：静态观测。graph.get_graph().draw_mermaid() 把节点和边画出来，一眼看清结构。

debug 的第一类事件由 map_debug_tasks（debug.py:41）产出——在任务执行之前发（Day 20 tick 里 _emit("tasks", ...)）：

# debug.py:41
def map_debug_tasks(tasks) -> Iterator[TaskPayload]:
    """Produce "task" events for stream_mode=debug."""
    for task in tasks:
        if task.config is not None and TAG_HIDDEN in task.config.get("tags", []):
            continue                                   # 44 隐藏任务不吐
        payload: TaskPayload = {
            "id": task.id,                             # 48 任务唯一 id
            "name": task.name,                         # 49 节点名
            "input": task.input,                       # 50 喂给它的输入（读通道读出来的）
            "triggers": task.triggers,                 # 51 是被哪些通道触发的
        }
        if task.config is not None:                    # 60 附上用户级 metadata（过滤内部键）
            md = {k: v for k, v in (task.config.get("metadata") or {}).items()
                  if k not in EXCLUDED_METADATA_KEYS}
            ...
            if md:
                payload["metadata"] = md
        yield payload

第二类事件由 map_debug_checkpoint（debug.py:144）产出——在 after_tick 存档时发，携带这步的完整快照：

# debug.py:144
def map_debug_checkpoint(config, channels, stream_channels, metadata, tasks,
                         pending_writes, parent_config, output_keys) -> Iterator[CheckpointPayload]:
    ...
    yield {
        "config": rm_pregel_keys(patch_checkpoint_map(config, metadata)),   # 177 本步 checkpoint 定位
        "parent_config": ...,                          # 178 上一个 checkpoint（用于时间旅行链）
        "values": read_channels(channels, stream_channels),   # 179 存档时的完整状态
        "metadata": metadata,                          # 180 step/source 等元信息
        "next": [t.name for t in tasks],               # 181 下一步会跑谁
        "tasks": [                                     # 182 每个任务的结果/错误/中断
            {"id": t.id, "name": t.name, "error": t.error, "state": t.state}
            if t.error else
            {"id": t.id, "name": t.name, "result": t.result,
             "interrupts": tuple(asdict(i) for i in t.interrupts), "state": t.state}
            if t.result else
            {"id": t.id, "name": t.name, "interrupts": ..., "state": t.state}
            for t in tasks_w_writes(tasks, pending_writes, task_states, output_keys)
        ],
    }

debug 流不是独立通道，而是 _emit（Day 20，_loop.py:1380）把 checkpoints/tasks 事件重新包装成 debug：

# _loop.py:1380
def _emit(self, mode, values, *args, **kwargs) -> None:
    if self.stream is None:
        return
    debug_remap = mode in ("checkpoints", "tasks") and "debug" in self.stream.modes  # 1389
    if mode not in self.stream.modes and not debug_remap:
        return                                         # 1390 既不是订阅的模式、也不需重映射 → 不发
    for v in values(*args, **kwargs):                  # 1392 调 map_debug_* 生成事件
        if mode in self.stream.modes:
            self.stream((self.checkpoint_ns, mode, v))  # 1394 原模式直接发
        if debug_remap:                                # 1396 若订阅了 debug，再包一层发一份
            self.stream((self.checkpoint_ns, "debug", {
                "step": self.step - 1 if mode == "checkpoints" else self.step,  # 1402
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "type": "checkpoint" if mode == "checkpoints" else "task_result" ...
            }))

一个小而精的细节：debug 展示任务写入时，同一通道被写多次要特殊呈现（map_task_result_writes，debug.py:83）：

# debug.py:83
def map_task_result_writes(writes) -> dict[str, Any]:
    """单次写记为 {channel: write}；多次写记为 {channel: {'$writes': [w1, w2, ...]}}"""
    result: dict[str, Any] = {}
    for channel, value in writes:
        existing = result.get(channel)
        if existing is not None:                       # 93 这个通道之前已经写过
            channel_writes = (existing["$writes"]      # 94 已是聚合形式就取列表
                              if is_multiple_channel_write(existing) else [existing])
            channel_writes.append(value)               # 99 追加本次写入
            result[channel] = {"$writes": channel_writes}   # 100 聚合成 $writes 列表
        else:
            result[channel] = value                    # 102 首次写：直接记值

而任务跑完后的 task_result 事件（debug.py:106）就用它来打包结果里的写入：

# debug.py:106
def map_debug_task_results(task_tup, stream_keys) -> Iterator[TaskResultPayload]:
    """Produce "task_result" events for stream_mode=debug."""
    task, writes = task_tup
    yield {
        "id": task.id, "name": task.name,
        "error": next((w[1] for w in writes if w[0] == ERROR), None),   # 118 有错取错
        "result": map_task_result_writes(                                # 119 用 L05 那个聚合
            [w for w in writes if w[0] in stream_channels_list or w[0] == RETURN]),
        "interrupts": [asdict(v) for w in writes if w[0] == INTERRUPT    # 122 中断信息
                       for v in (w[1] if isinstance(w[1], Sequence) else [w[1]])],
    }

换到画图。draw_graph（_draw.py:42）最惊艳的地方：它不去读图的静态结构，而是拿空 checkpoint 真的跑一遍 Pregel 循环，看每步谁触发谁，反推出边（_draw.py:88-118）：

# _draw.py:88
input_writes = list(map_input(input_channels, {}))    # 89 造一批"空输入"写入（复用 Day 24！）
updated_channels = apply_writes(                       # 90 空跑 apply_writes（复用 Day 21！）
    checkpoint, channels,
    [PregelTaskWrites((), INPUT, input_writes, [])],
    get_next_version, trigger_to_nodes,
)
tasks = prepare_next_tasks(                            # 100 空跑 prepare_next_tasks（复用 Day 21！）
    checkpoint, [], nodes, channels, managed, config,
    step, -1, for_execution=True, ...,
    updated_channels=updated_channels,
)
start_tasks = tasks
for step in range(step, limit):                        # 118 真的循环模拟超步
    if not tasks:
        break                                          # 119 没任务了=图跑完=画完了
    ...记录这步 tasks 是被哪些通道触发的 → 连边...

✋ 10 分钟动手

# 1. 读 debug 事件映射
sed -n '41,71p' libs/langgraph/langgraph/pregel/debug.py
sed -n '144,206p' libs/langgraph/langgraph/pregel/debug.py

# 2. 读画图"空跑发现边"那段
sed -n '88,120p' libs/langgraph/langgraph/pregel/_draw.py

# 3. 看 debug 流的真实事件 + 画 Mermaid
python -c "
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_node('b', lambda s:{'n':s['n']*2})
g.add_edge(START,'a'); g.add_edge('a','b'); g.add_edge('b',END)
app = g.compile()
for ev in app.stream({'n':1}, stream_mode='debug'):
    print(ev['type'], '| step', ev.get('step'), '|', ev['payload'].get('name') or ev['payload'].get('next'))
print(app.get_graph().draw_mermaid())   # 打印 Mermaid 源码
"