LangGraph 源码深潜：Streaming、Events 与 Observability 机制解剖

1
def compile(
2
    self,
3
    checkpointer: Checkpointer = None,
4
    *,
5
    cache: BaseCache | None = None,
6
    store: BaseStore | None = None,
7
    interrupt_before: All | list[str] | None = None,
8
    interrupt_after: All | list[str] | None = None,
9
    debug: bool = False,
10
    name: str | None = None,
11
) -> CompiledStateGraph:
12
    ...

1
业务代码
2
  ↓
3
builder.compile()
4
  ↓
5
StateGraph.compile
6
  ↓
7
CompiledStateGraph / Pregel
8
  ↓
9
graph.stream / graph.astream

1
def compile(self, **options):
2
    # 1. 校验图结构
3
    self.validate()
4

5
    # 2. 构造 CompiledStateGraph
6
    graph = CompiledStateGraph(
7
        nodes={},
8
        channels=self.channels,
9
        input_channels=self.input_channels,
10
        output_channels=self.output_channels,
11
        stream_channels=self.stream_channels,
12
        checkpointer=options.get("checkpointer"),
13
        store=options.get("store"),
14
        debug=options.get("debug"),
15
        name=options.get("name"),
16
    )
17

18
    # 3. 附加节点
19
    for node_name, node_spec in self.nodes.items():
20
        graph.attach_node(node_name, node_spec)
21

22
    # 4. 附加边和分支
23
    for edge in self.edges:
24
        graph.attach_edge(edge)
25

26
    for branch in self.branches:
27
        graph.attach_branch(branch)
28

29
    # 5. 返回可执行图
30
    return graph.validate()

1
StateGraph builder
2
  ↓
3
compile
4
  ↓
5
CompiledStateGraph
6
  ↓
7
stream capable runtime

1
class RunnableConfig(TypedDict, total=False):
2
    tags: list[str]
3
    metadata: dict[str, Any]
4
    callbacks: Callbacks
5
    run_name: str
6
    configurable: dict[str, Any]
7
    run_id: UUID
8
    max_concurrency: int
9
    recursion_limit: int

1
业务代码
2
  ↓
3
graph.stream(input, config={"tags": ..., "metadata": ..., "callbacks": ...})
4
  ↓
5
Pregel runtime
6
  ↓
7
节点内 Runnable 子调用
8
  ↓
9
ChatModel / Tool / Chain
10
  ↓
11
callbacks / tracers

1
def ensure_config(config):
2
    # 1. 从父调用 ContextVar 中读取 inherited config
3
    parent_config = get_current_runnable_config()
4

5
    # 2. 创建默认 config
6
    merged = {
7
        "tags": [],
8
        "metadata": {},
9
        "callbacks": None,
10
        "configurable": {},
11
        "recursion_limit": DEFAULT_RECURSION_LIMIT,
12
    }
13

14
    # 3. 合并父 config
15
    if parent_config:
16
        merged["tags"].extend(parent_config.get("tags", []))
17
        merged["metadata"].update(parent_config.get("metadata", {}))
18
        merged["callbacks"] = merge_callbacks(
19
            parent_config.get("callbacks"),
20
            merged.get("callbacks"),
21
        )
22

23
    # 4. 合并当前 config
24
    if config:
25
        merged["tags"].extend(config.get("tags", []))
26
        merged["metadata"].update(config.get("metadata", {}))
27
        merged["callbacks"] = merge_callbacks(
28
            merged.get("callbacks"),
29
            config.get("callbacks"),
30
        )
31
        merged["configurable"].update(config.get("configurable", {}))
32

33
    return merged

1
config
2
  ↓
3
ensure_config
4
  ↓
5
callbacks / tags / metadata
6
  ↓
7
传递给子调用
8
  ↓
9
trace 可过滤、可分组

1
def stream(
2
    self,
3
    input: InputT | Command | None,
4
    config: RunnableConfig | None = None,
5
    *,
6
    context: ContextT | None = None,
7
    stream_mode: StreamMode | Sequence[StreamMode] | None = None,
8
    print_mode: StreamMode | Sequence[StreamMode] = (),
9
    output_keys: str | Sequence[str] | None = None,
10
    interrupt_before: All | Sequence[str] | None = None,
11
    interrupt_after: All | Sequence[str] | None = None,
12
    durability: Durability | None = None,
13
    subgraphs: bool = False,
14
    debug: bool | None = None,
15
    version: Literal["v1", "v2"] | None = None,
16
    **kwargs: Any,
17
) -> Iterator[dict[str, Any] | Any]:
18
    ...

1
业务代码
2
  ↓
3
graph.stream(input, stream_mode="updates")
4
  ↓
5
Pregel.stream
6
  ↓
7
PregelLoop
8
  ↓
9
PregelRunner
10
  ↓
11
StreamProtocol
12
  ↓
13
yield chunk

1
def stream(input_value, config=None, *, stream_mode=None, version=None, **kwargs):
2
    # 1. 规范化 config
3
    config = ensure_config(config)
4

5
    # 2. 规范化 stream_mode
6
    modes = normalize_stream_modes(
7
        stream_mode or self.default_stream_mode
8
    )
9

10
    # 3. 初始化 callback manager / run manager
11
    callback_manager = get_callback_manager_for_config(config)
12
    run_manager = callback_manager.on_chain_start(
13
        serialized=self,
14
        inputs=input_value,
15
        name=config.get("run_name") or self.get_name(),
16
    )
17

18
    # 4. 初始化 stream protocol
19
    stream_protocol = StreamProtocol(
20
        modes=modes,
21
        version=version or "v2",
22
        subgraphs=kwargs.get("subgraphs", False),
23
        print_modes=normalize_print_modes(kwargs.get("print_mode", ())),
24
    )
25

26
    # 5. 初始化 Pregel loop
27
    with PregelLoop(
28
        graph=self,
29
        input=input_value,
30
        config=config,
31
        stream=stream_protocol,
32
        run_manager=run_manager,
33
        interrupt_before=kwargs.get("interrupt_before"),
34
        interrupt_after=kwargs.get("interrupt_after"),
35
        durability=kwargs.get("durability"),
36
    ) as loop:
37
        # 6. 在 loop 中逐步执行 task
38
        while loop.tick():
39
            for task in loop.ready_tasks:
40
                runner.submit(task)
41

42
            # 7. 读取 runtime 产生的 stream chunks
43
            while stream_protocol.has_chunks():
44
                chunk = stream_protocol.pop_chunk()
45
                yield format_chunk(
46
                    chunk,
47
                    version=stream_protocol.version,
48
                )
49

50
    # 8. 成功结束
51
    run_manager.on_chain_end(loop.output)

1
stream()
2
  ↓
3
init config / callbacks / stream protocol
4
  ↓
5
Pregel loop
6
  ↓
7
node execution
8
  ↓
9
stream_protocol receives events
10
  ↓
11
yield chunks
12
  ↓
13
END

1
def invoke(
2
    self,
3
    input: InputT | Command | None,
4
    config: RunnableConfig | None = None,
5
    *,
6
    context: ContextT | None = None,
7
    stream_mode: StreamMode = "values",
8
    print_mode: StreamMode | Sequence[StreamMode] = (),
9
    output_keys: str | Sequence[str] | None = None,
10
    interrupt_before: All | Sequence[str] | None = None,
11
    interrupt_after: All | Sequence[str] | None = None,
12
    durability: Durability | None = None,
13
    **kwargs: Any,
14
) -> dict[str, Any] | Any:
15
    ...

1
graph.invoke(input)
2
  ↓
3
graph.stream(input, stream_mode="values")
4
  ↓
5
遍历 chunks
6
  ↓
7
保存最后一个 output
8
  ↓
9
return latest

1
def invoke(input_value, config=None, **kwargs):
2
    latest_output = None
3

4
    for chunk in self.stream(
5
        input_value,
6
        config=config,
7
        stream_mode=kwargs.pop("stream_mode", "values"),
8
        **kwargs,
9
    ):
10
        # 1. 根据 output mode 识别真正的 state/output
11
        if is_output_chunk(chunk):
12
            latest_output = extract_output(chunk)
13

14
        # 2. 忽略纯 debug / custom / messages chunks
15
        else:
16
            continue
17

18
    return latest_output

1
invoke
2
  ↓
3
stream execution
4
  ↓
5
consume all chunks
6
  ↓
7
return latest output

1
graph.stream(input, stream_mode="updates", version="v2")

1
node returns {"plan": "..."}
2
  ↓
3
Pregel apply writes
4
  ↓
5
updates stream formatter
6
  ↓
7
yield {"type": "updates", "data": {"node": {"plan": "..."}}}

1
def emit_updates(task_result, stream_protocol):
2
    # 1. 取得节点名和节点返回值
3
    node_name = task_result.name
4
    update = task_result.writes
5

6
    # 2. 过滤不可作为 state update 的内部写入
7
    public_update = filter_state_update_keys(update)
8

9
    if not public_update:
10
        return
11

12
    # 3. 生成 updates stream part
13
    stream_protocol.write(
14
        mode="updates",
15
        data={
16
            node_name: public_update
17
        },
18
        namespace=task_result.namespace,
19
        metadata=task_result.metadata,
20
    )

1
node update
2
  ↓
3
filter public keys
4
  ↓
5
yield updates chunk

1
graph.stream(input, stream_mode="values", version="v2")

1
node updates
2
  ↓
3
reducers apply writes
4
  ↓
5
state channels updated
6
  ↓
7
read output channels
8
  ↓
9
yield full state snapshot

1
def emit_values(loop, stream_protocol):
2
    # 1. 当前 super-step 的 writes 已应用到 channels
3
    channels = loop.channels
4

5
    # 2. 根据 output_keys / stream_channels 读取可见 state
6
    values = read_channels(
7
        channels=channels,
8
        keys=loop.output_keys,
9
    )
10

11
    # 3. 生成 values stream part
12
    stream_protocol.write(
13
        mode="values",
14
        data=values,
15
        namespace=loop.namespace,
16
        metadata=loop.metadata,
17
    )

1
apply writes
2
  ↓
3
read state
4
  ↓
5
yield values

1
graph.stream(input, stream_mode="messages", version="v2")

1
(message_chunk, metadata)

1
node calls ChatModel.invoke / stream
2
  ↓
3
ChatModel emits AIMessageChunk
4
  ↓
5
LangChain callbacks / stream hooks
6
  ↓
7
LangGraph messages stream
8
  ↓
9
yield (message_chunk, metadata)

1
def node(state, config):
2
    # 1. 节点内部调用模型
3
    response = model.invoke(
4
        [{"role": "user", "content": state["user_request"]}],
5
        config=config,
6
    )
7

8
    # 2. 节点最终仍然返回 state update
9
    return {"answer": response.content}
10

11

12
def on_chat_model_stream(message_chunk, run_metadata):
13
    # 3. ChatModel 每产生一个 chunk，callback / stream hook 捕获
14
    metadata = {
15
        "langgraph_node": run_metadata.node_name,
16
        "tags": run_metadata.tags,
17
        "run_id": run_metadata.run_id,
18
        "model_name": run_metadata.model_name,
19
    }
20

21
    # 4. 写入 messages stream
22
    stream_protocol.write(
23
        mode="messages",
24
        data=(message_chunk, metadata),
25
    )

1
metadata["langgraph_node"]
2
metadata["tags"]

1
ChatModel streaming
2
  ↓
3
AIMessageChunk
4
  ↓
5
metadata attached
6
  ↓
7
messages chunk

1
def get_stream_writer() -> StreamWriter:
2
    ...

1
writer = get_stream_writer()
2
writer({"status": "searching", "progress": 30})

1
node / tool
2
  ↓
3
get_stream_writer()
4
  ↓
5
current runtime context
6
  ↓
7
writer(data)
8
  ↓
9
custom stream chunk

1
def get_stream_writer():
2
    # 1. 从当前运行上下文读取 writer
3
    writer = current_runtime_context.get("stream_writer")
4

5
    # 2. 如果当前不是 stream 执行，可能返回 no-op writer 或抛错
6
    if writer is None:
7
        return noop_writer_or_raise()
8

9
    return writer
10

11

12
def node(state):
13
    writer = get_stream_writer()
14

15
    # 3. 发出自定义进度
16
    writer({
17
        "type": "progress",
18
        "stage": "search_food",
19
        "message": "正在检索美食推荐",
20
        "progress": 30,
21
    })
22

23
    # 4. 执行业务逻辑
24
    notes = search_food(state["user_request"])
25

26
    writer({
27
        "type": "progress",
28
        "stage": "search_food",
29
        "message": "美食检索完成",
30
        "progress": 100,
31
    })
32

33
    # 5. 返回真正的 state update
34
    return {"research_notes": notes}

1
get_stream_writer()
2
  ↓
3
writer(data)
4
  ↓
5
custom chunk
6
  ↓
7
node return update

1
stream = graph.stream_events(input, version="v3")

1
graph.stream_events(input)
2
  ↓
3
底层 Pregel raw events
4
  ↓
5
event router
6
  ↓
7
stream transformers
8
  ↓
9
stream.messages / stream.values / stream.output / stream.interrupts

1
def stream_events(input_value, config=None, version="v3"):
2
    # 1. 创建底层 raw event stream
3
    raw_stream = self.stream(
4
        input_value,
5
        config=config,
6
        stream_mode=[
7
            "values",
8
            "updates",
9
            "messages",
10
            "custom",
11
            "tasks",
12
            "checkpoints",
13
        ],
14
        version="v2",
15
    )
16

17
    # 2. 创建 event router
18
    router = EventRouter()
19

20
    # 3. 注册内置 transformers
21
    router.add_transformer("messages", MessagesTransformer())
22
    router.add_transformer("values", ValuesTransformer())
23
    router.add_transformer("subgraphs", SubgraphsTransformer())
24
    router.add_transformer("interrupts", InterruptsTransformer())
25
    router.add_transformer("output", OutputTransformer())
26

27
    # 4. 将 raw events 路由到 projections
28
    for event in raw_stream:
29
        router.route(event)
30

31
    # 5. 返回可并发消费的 run stream object
32
    return router.run_stream

1
raw Pregel events
2
  ↓
3
event router
4
  ↓
5
transformers
6
  ↓
7
typed projections

1
on_chain_start(...)
2
on_chain_end(...)
3
on_chain_error(...)
4
on_llm_start(...)
5
on_llm_new_token(...)
6
on_llm_end(...)
7
on_tool_start(...)
8
on_tool_end(...)

1
Runnable / Graph / Model / Tool
2
  ↓
3
CallbackManager
4
  ↓
5
CallbackHandler / Tracer
6
  ↓
7
Console / LangSmith / Custom sink

1
def runnable_invoke(input_value, config):
2
    config = ensure_config(config)
3
    callback_manager = get_callback_manager_for_config(config)
4

5
    run_manager = callback_manager.on_chain_start(
6
        serialized=self,
7
        inputs=input_value,
8
        tags=config.get("tags"),
9
        metadata=config.get("metadata"),
10
    )
11

12
    try:
13
        output = self._invoke(input_value, config)
14

15
    except Exception as exc:
16
        run_manager.on_chain_error(exc)
17
        raise
18

19
    else:
20
        run_manager.on_chain_end(output)
21
        return output
22

23

24
def chat_model_stream(messages, config):
25
    run_manager = callback_manager.on_llm_start(
26
        serialized=model,
27
        prompts_or_messages=messages,
28
    )
29

30
    for token in provider_stream(messages):
31
        chunk = AIMessageChunk(content=token)
32
        run_manager.on_llm_new_token(token, chunk=chunk)
33
        yield chunk
34

35
    run_manager.on_llm_end(final_generation)

1
messages stream
2
  面向实时 UI / token 展示。
3

4
LangSmith trace
5
  面向调试、评估、监控和历史记录。

1
Runnable start
2
  ↓
3
sub-runs start/end
4
  ↓
5
LLM token events
6
  ↓
7
tool events
8
  ↓
9
Runnable end
10
  ↓
11
trace tree