> ## Documentation Index
> Fetch the complete documentation index at: https://langchain.idochub.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# 流式传输

<警告>
  **Alpha 版本提示：** 本文档涵盖的是 **v1-alpha** 版本。内容尚不完整，且可能随时更改。

  如需查阅最新稳定版本，请参阅当前的 [LangGraph Python 文档](https://langchain-ai.github.io/langgraph/) 或 [LangGraph JavaScript 文档](https://langchain-ai.github.io/langgraphjs/)。
</警告>

LangGraph 实现了一套流式传输系统，用于实时推送更新。流式传输对于提升基于大语言模型（LLM）构建的应用程序的响应速度至关重要。通过在完整响应尚未生成完毕时就逐步显示输出内容，流式传输显著改善了用户体验（UX），尤其在应对大语言模型固有的延迟时效果尤为明显。

LangGraph 流式传输支持的功能：

* <Icon icon="share-nodes" size={16} /> [**流式传输图状态**](#stream-graph-state) — 使用 `updates` 和 `values` 模式获取状态更新/值。
* <Icon icon="square-poll-horizontal" size={16} /> [**流式传输子图输出**](#stream-subgraph-outputs) — 同时包含父图和任何嵌套子图的输出。
* <Icon icon="square-binary" size={16} /> [**流式传输 LLM 令牌**](#messages) — 从任意位置捕获令牌流：节点内部、子图或工具中。
* <Icon icon="table" size={16} /> [**流式传输自定义数据**](#stream-custom-data) — 从工具函数中直接发送自定义更新或进度信号。
* <Icon icon="layer-plus" size={16} /> [**使用多种流式传输模式**](#stream-multiple-modes) — 可选择 `values`（完整状态）、`updates`（状态增量）、`messages`（LLM 令牌 + 元数据）、`custom`（任意用户数据）或 `debug`（详细追踪信息）。

## 支持的流式传输模式

将以下一个或多个流式传输模式作为列表传递给 [`stream()`](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.CompiledStateGraph.stream) 或 [`astream()`](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.CompiledStateGraph.astream) 方法：

| 模式         | 描述                                                             |
| ---------- | -------------------------------------------------------------- |
| `values`   | 在图的每一步执行后，流式传输状态的完整值。                                          |
| `updates`  | 在图的每一步执行后，流式传输状态的更新内容。如果同一步骤中进行了多次更新（例如，运行了多个节点），这些更新将被分别流式传输。 |
| `custom`   | 从图节点内部流式传输自定义数据。                                               |
| `messages` | 从调用 LLM 的任何图节点中流式传输二元组（LLM 令牌, 元数据）。                           |
| `debug`    | 在图执行过程中尽可能多地流式传输信息。                                            |

## 基本用法示例

LangGraph 图暴露了 [`.stream()`](https://langchain-ai.github.io/langgraph/reference/pregel/#langgraph.pregel.Pregel.stream)（同步）和 [`.astream()`](https://langchain-ai.github.io/langgraph/reference/pregel/#langgraph.pregel.Pregel.astream)（异步）方法，以迭代器形式生成流式输出。

```python theme={null}
for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)
```

<Accordion title="扩展示例：流式传输更新">
  ```python {highlight={24,26}} theme={null}
  from typing import TypedDict
  from langgraph.graph import StateGraph, START, END

  class State(TypedDict):
      topic: str
      joke: str

  def refine_topic(state: State):
      return {"topic": state["topic"] + " and cats"}

  def generate_joke(state: State):
      return {"joke": f"This is a joke about {state['topic']}"}

  graph = (
      StateGraph(State)
      .add_node(refine_topic)
      .add_node(generate_joke)
      .add_edge(START, "refine_topic")
      .add_edge("refine_topic", "generate_joke")
      .add_edge("generate_joke", END)
      .compile()
  )

  for chunk in graph.stream( # (1)!
      {"topic": "ice cream"},
      stream_mode="updates", # (2)!
  ):
      print(chunk)
  ```

  1. `stream()` 方法返回一个迭代器，用于生成流式输出。
  2. 设置 `stream_mode="updates"` 以仅流式传输每个节点执行后对图状态的更新。也可使用其他流式传输模式。详情请参阅[支持的流式传输模式](#supported-stream-modes)。

  ```output theme={null}
  {'refineTopic': {'topic': 'ice cream and cats'}}
  {'generateJoke': {'joke': 'This is a joke about ice cream and cats'}}
  ```
</Accordion>

## 流式传输多种模式

你可以将 `stream_mode` 参数设置为一个列表，以同时流式传输多种模式。

流式输出将是 `(mode, chunk)` 形式的元组，其中 `mode` 是流式传输模式的名称，`chunk` 是该模式流式传输的数据。

```python theme={null}
for mode, chunk in graph.stream(inputs, stream_mode=["updates", "custom"]):
    print(chunk)
```

## 流式传输图状态

使用 `updates` 和 `values` 流式传输模式在图执行时流式传输其状态。

* `updates` 在图的每一步执行后流式传输状态的**更新**。
* `values` 在图的每一步执行后流式传输状态的**完整值**。

```python theme={null}
from typing import TypedDict
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
  topic: str
  joke: str


def refine_topic(state: State):
    return {"topic": state["topic"] + " and cats"}


def generate_joke(state: State):
    return {"joke": f"This is a joke about {state['topic']}"}

graph = (
  StateGraph(State)
  .add_node(refine_topic)
  .add_node(generate_joke)
  .add_edge(START, "refine_topic")
  .add_edge("refine_topic", "generate_joke")
  .add_edge("generate_joke", END)
  .compile()
)
```

<Tabs>
  <Tab title="updates">
    使用此模式仅流式传输每个步骤后节点返回的**状态更新**。流式输出包含节点名称及其更新内容。

    ```python {highlight={3}} theme={null}
    for chunk in graph.stream(
        {"topic": "ice cream"},
        stream_mode="updates",
    ):
        print(chunk)
    ```
  </Tab>

  <Tab title="values">
    使用此模式在每个步骤后流式传输图的**完整状态**。

    ```python {highlight={3}} theme={null}
    for chunk in graph.stream(
        {"topic": "ice cream"},
        stream_mode="values",
    ):
        print(chunk)
    ```
  </Tab>
</Tabs>

## 流式传输子图输出

为了在流式输出中包含[子图](/oss/python/subgraphs)的输出，你可以在父图的 `.stream()` 方法中设置 `subgraphs=True`。这将同时流式传输父图和任何子图的输出。

输出将以 `(namespace, data)` 元组的形式流式传输，其中 `namespace` 是一个元组，表示调用子图的节点路径，例如 `("parent_node:<task_id>", "child_node:<task_id>")`。

```python {highlight={3}} theme={null}
for chunk in graph.stream(
    {"foo": "foo"},
    subgraphs=True, # (1)!
    stream_mode="updates",
):
    print(chunk)
```

1. 设置 `subgraphs=True` 以流式传输子图的输出。

<Accordion title="扩展示例：从子图流式传输">
  ```python {highlight={39}} theme={null}
  from langgraph.graph import START, StateGraph
  from typing import TypedDict

  # 定义子图
  class SubgraphState(TypedDict):
      foo: str  # 注意此键与父图状态共享
      bar: str

  def subgraph_node_1(state: SubgraphState):
      return {"bar": "bar"}

  def subgraph_node_2(state: SubgraphState):
      return {"foo": state["foo"] + state["bar"]}

  subgraph_builder = StateGraph(SubgraphState)
  subgraph_builder.add_node(subgraph_node_1)
  subgraph_builder.add_node(subgraph_node_2)
  subgraph_builder.add_edge(START, "subgraph_node_1")
  subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
  subgraph = subgraph_builder.compile()

  # 定义父图
  class ParentState(TypedDict):
      foo: str

  def node_1(state: ParentState):
      return {"foo": "hi! " + state["foo"]}

  builder = StateGraph(ParentState)
  builder.add_node("node_1", node_1)
  builder.add_node("node_2", subgraph)
  builder.add_edge(START, "node_1")
  builder.add_edge("node_1", "node_2")
  graph = builder.compile()

  for chunk in graph.stream(
      {"foo": "foo"},
      stream_mode="updates",
      subgraphs=True, # (1)!
  ):
      print(chunk)
  ```

  1. 设置 `subgraphs=True` 以流式传输子图的输出。

  ```
  ((), {'node_1': {'foo': 'hi! foo'}})
  (('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',), {'subgraph_node_1': {'bar': 'bar'}})
  (('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',), {'subgraph_node_2': {'foo': 'hi! foobar'}})
  ((), {'node_2': {'foo': 'hi! foobar'}})
  ```

  **注意** 我们不仅接收到了节点更新，还接收到了命名空间，它告诉我们当前流式传输来自哪个图（或子图）。
</Accordion>

<a id="debug" />

### 调试

使用 `debug` 流式传输模式在图执行过程中尽可能多地流式传输信息。流式输出包括节点名称和完整状态。

```python {highlight={3}} theme={null}
for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="debug",
):
    print(chunk)
```

<a id="messages" />

## LLM 令牌

使用 `messages` 流式传输模式，从图的任何部分（包括节点、工具、子图或任务）**逐令牌**流式传输大语言模型（LLM）的输出。

[`messages` 模式](#supported-stream-modes) 的流式输出是一个元组 `(message_chunk, metadata)`，其中：

* `message_chunk`：来自 LLM 的令牌或消息片段。
* `metadata`：包含有关图节点和 LLM 调用详细信息的字典。

> 如果你的 LLM 没有 LangChain 集成，可以改用 `custom` 模式流式传输其输出。详情请参阅[与任意 LLM 一起使用](#use-with-any-llm)。

<Warning>
  **Python \< 3.11 的异步需要手动配置**
  在 Python \< 3.11 中使用异步代码时，必须显式地将 `RunnableConfig` 传递给 `ainvoke()` 以启用正确的流式传输。详情请参阅[异步与 Python \< 3.11](#async) 或升级到 Python 3.11+。
</Warning>

```python {highlight={17,33}} theme={null}
from dataclasses import dataclass

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START


@dataclass
class MyState:
    topic: str
    joke: str = ""


llm = init_chat_model(model="openai:gpt-4o-mini")

def call_model(state: MyState):
    """调用 LLM 生成关于某个主题的笑话"""
    llm_response = llm.invoke( # (1)!
        [
            {"role": "user", "content": f"Generate a joke about {state.topic}"}
        ]
    )
    return {"joke": llm_response.content}

graph = (
    StateGraph(MyState)
    .add_node(call_model)
    .add_edge(START, "call_model")
    .compile()
)

for message_chunk, metadata in graph.stream( # (2)!
    {"topic": "ice cream"},
    stream_mode="messages",
):
    if message_chunk.content:
        print(message_chunk.content, end="|", flush=True)
```

1. 注意，即使使用 `.invoke` 而不是 `.stream` 运行 LLM，也会发出消息事件。
2. "messages" 流式传输模式返回一个 `(message_chunk, metadata)` 元组的迭代器，其中 `message_chunk` 是 LLM 流式传输的令牌，`metadata` 是包含调用 LLM 的图节点信息及其他信息的字典。

#### 按 LLM 调用过滤

你可以为 LLM 调用关联 `tags`，以按 LLM 调用过滤流式传输的令牌。

```python {highlight={10}} theme={null}
from langchain.chat_models import init_chat_model

llm_1 = init_chat_model(model="openai:gpt-4o-mini", tags=['joke']) # (1)!
llm_2 = init_chat_model(model="openai:gpt-4o-mini", tags=['poem']) # (2)!

graph = ... # 定义一个使用这些 LLM 的图

async for msg, metadata in graph.astream(  # (3)!
    {"topic": "cats"},
    stream_mode="messages",
):
    if metadata["tags"] == ["joke"]: # (4)!
        print(msg.content, end="|", flush=True)
```

1. llm\_1 被标记为 "joke"。
2. llm\_2 被标记为 "poem"。
3. `stream_mode` 设置为 "messages" 以流式传输 LLM 令牌。`metadata` 包含有关 LLM 调用的信息，包括标签。
4. 通过 `metadata` 中的 `tags` 字段过滤流式传输的令牌，仅包含标记为 "joke" 的 LLM 调用的令牌。

<Accordion title="扩展示例：按标签过滤">
  ```python {highlight={42}} theme={null}
  from typing import TypedDict

  from langchain.chat_models import init_chat_model
  from langgraph.graph import START, StateGraph

  joke_model = init_chat_model(model="openai:gpt-4o-mini", tags=["joke"]) # (1)!
  poem_model = init_chat_model(model="openai:gpt-4o-mini", tags=["poem"]) # (2)!


  class State(TypedDict):
        topic: str
        joke: str
        poem: str


  async def call_model(state, config):
        topic = state["topic"]
        print("正在写笑话...")
        # 注意：在 Python < 3.11 中，必须显式传递 config
        # 因为在此之前未添加上下文变量支持：https://docs.python.org/3/library/asyncio-task.html#creating-tasks
        joke_response = await joke_model.ainvoke(
              [{"role": "user", "content": f"写一个关于 {topic} 的笑话"}],
              config, # (3)!
        )
        print("\n\n正在写诗...")
        poem_response = await poem_model.ainvoke(
              [{"role": "user", "content": f"写一首关于 {topic} 的短诗"}],
              config, # (3)!
        )
        return {"joke": joke_response.content, "poem": poem_response.content}


  graph = (
        StateGraph(State)
        .add_node(call_model)
        .add_edge(START, "call_model")
        .compile()
  )

  async for msg, metadata in graph.astream(
        {"topic": "cats"},
        stream_mode="messages", # (4)!
  ):
      if metadata["tags"] == ["joke"]: # (4)!
          print(msg.content, end="|", flush=True)
  ```

  1. `joke_model` 被标记为 "joke"。
  2. `poem_model` 被标记为 "poem"。
  3. 显式传递 `config` 以确保上下文变量正确传播。这在 Python \< 3.11 中使用异步代码时是必需的。详情请参阅[异步部分](#async)。
  4. `stream_mode` 设置为 "messages" 以流式传输 LLM 令牌。`metadata` 包含有关 LLM 调用的信息，包括标签。
</Accordion>

#### 按节点过滤

要仅从特定节点流式传输令牌，请使用 `stream_mode="messages"` 并通过流式传输元数据中的 `langgraph_node` 字段过滤输出：

```python {highlight={3,5}} theme={null}
for msg, metadata in graph.stream( # (1)!
    inputs,
    stream_mode="messages",
):
    if msg.content and metadata["langgraph_node"] == "some_node_name": # (2)!
        ...
```

1. "messages" 流式传输模式返回一个 `(message_chunk, metadata)` 元组，其中 `message_chunk` 是 LLM 流式传输的令牌，`metadata` 是包含调用 LLM 的图节点信息及其他信息的字典。
2. 通过 `metadata` 中的 `langgraph_node` 字段过滤流式传输的令牌，仅包含来自 `write_poem` 节点的令牌。

<Accordion title="扩展示例：从特定节点流式传输 LLM 令牌">
  ```python {highlight={40,44}} theme={null}
  from typing import TypedDict
  from langgraph.graph import START, StateGraph
  from langchain_openai import ChatOpenAI

  model = ChatOpenAI(model="gpt-4o-mini")


  class State(TypedDict):
        topic: str
        joke: str
        poem: str


  def write_joke(state: State):
        topic = state["topic"]
        joke_response = model.invoke(
              [{"role": "user", "content": f"写一个关于 {topic} 的笑话"}]
        )
        return {"joke": joke_response.content}


  def write_poem(state: State):
        topic = state["topic"]
        poem_response = model.invoke(
              [{"role": "user", "content": f"写一首关于 {topic} 的短诗"}]
        )
        return {"poem": poem_response.content}


  graph = (
        StateGraph(State)
        .add_node(write_joke)
        .add_node(write_poem)
        # 并发地写笑话和诗
        .add_edge(START, "write_joke")
        .add_edge(START, "write_poem")
        .compile()
  )

  for msg, metadata in graph.stream( # (1)!
      {"topic": "cats"},
      stream_mode="messages",
  ):
      if msg.content and metadata["langgraph_node"] == "write_poem": # (2)!
          print(msg.content, end="|", flush=True)
  ```

  1. "messages" 流式传输模式返回一个 `(message_chunk, metadata)` 元组，其中 `message_chunk` 是 LLM 流式传输的令牌，`metadata` 是包含调用 LLM 的图节点信息及其他信息的字典。
  2. 通过 `metadata` 中的 `langgraph_node` 字段过滤流式传输的令牌，仅包含来自 `write_poem` 节点的令牌。
</Accordion>

## 流式传输自定义数据

要从 LangGraph 节点或工具内部发送**自定义用户定义数据**，请按照以下步骤操作：

1. 使用 `get_stream_writer()` 访问流式写入器并发出自定义数据。
2. 在调用 `.stream()` 或 `.astream()` 时设置 `stream_mode="custom"` 以在流中获取自定义数据。你可以组合多种模式（例如，`["updates", "custom"]`），但至少必须包含 `"custom"`。

<Warning>
  **Python \< 3.11 的异步中无 `get_stream_writer()`**
  在 Python \< 3.11 上运行的异步代码中，`get_stream_writer()` 将无法工作。
  相反，请在你的节点或工具中添加一个 `writer` 参数并手动传递它。
  有关用法示例，请参阅[异步与 Python \< 3.11](#async)。
</Warning>

<Tabs>
  <Tab title="节点">
    ```python theme={null}
    from typing import TypedDict
    from langgraph.config import get_stream_writer
    from langgraph.graph import StateGraph, START

    class State(TypedDict):
        query: str
        answer: str

    def node(state: State):
        writer = get_stream_writer()  # (1)!
        writer({"custom_key": "在节点内部生成自定义数据"}) # (2)!
        return {"answer": "some data"}

    graph = (
        StateGraph(State)
        .add_node(node)
        .add_edge(START, "node")
        .compile()
    )

    inputs = {"query": "example"}

    # 用法
    for chunk in graph.stream(inputs, stream_mode="custom"):  # (3)!
        print(chunk)
    ```

    1. 获取流式写入器以发送自定义数据。
    2. 发出一个自定义键值对（例如，进度更新）。
       3\. 设置 `stream_mode="custom"` 以在流中接收自定义数据。
  </Tab>

  <Tab title="工具">
    ```python {highlight={8,10}} theme={null}
    from langchain_core.tools import tool
    from langgraph.config import get_stream_writer

    @tool
    def query_database(query: str) -> str:
        """查询数据库。"""
        writer = get_stream_writer() # (1)!
        writer({"data": "已检索 0/100 条记录", "type": "progress"}) # (2)!
        # 执行查询
        writer({"data": "已检索 100/100 条记录", "type": "progress"}) # (3)!
        return "some-answer"


    graph = ... # 定义一个使用此工具的图

    for chunk in graph.stream(inputs, stream_mode="custom"): # (4)!
        print(chunk)
    ```

    1. 访问流式写入器以发送自定义数据。
    2. 发出一个自定义键值对（例如，进度更新）。
    3. 发出另一个自定义键值对。
       4\. 设置 `stream_mode="custom"` 以在流中接收自定义数据。
  </Tab>
</Tabs>

## 与任意 LLM 一起使用

你可以使用 `stream_mode="custom"` 从**任意 LLM API** 流式传输数据 —— 即使该 API **未**实现 LangChain 聊天模型接口。

这使你可以集成原始 LLM 客户端或提供自身流式接口的外部服务，使 LangGraph 对于自定义设置具有高度灵活性。

```python {highlight={5,8,20}} theme={null}
from langgraph.config import get_stream_writer

def call_arbitrary_model(state):
    """调用任意模型并流式传输输出的示例节点"""
    writer = get_stream_writer() # (1)!
    # 假设你有一个生成块的流式客户端
    for chunk in your_custom_streaming_client(state["topic"]): # (2)!
        writer({"custom_llm_chunk": chunk}) # (3)!
    return {"result": "completed"}

graph = (
    StateGraph(State)
    .add_node(call_arbitrary_model)
    # 根据需要添加其他节点和边
    .compile()
)

for chunk in graph.stream(
    {"topic": "cats"},
    stream_mode="custom", # (4)!
):
    # 块将包含从 LLM 流式传输的自定义数据
    print(chunk)
```

1. 获取流式写入器以发送自定义数据。
2. 使用你的自定义流式客户端生成 LLM 令牌。
3. 使用写入器将自定义数据发送到流中。
4. 设置 `stream_mode="custom"` 以在流中接收自定义数据。

<Accordion title="扩展示例：流式传输任意聊天模型">
  ```python theme={null}
  import operator
  import json

  from typing import TypedDict
  from typing_extensions import Annotated
  from langgraph.graph import StateGraph, START

  from openai import AsyncOpenAI

  openai_client = AsyncOpenAI()
  model_name = "gpt-4o-mini"


  async def stream_tokens(model_name: str, messages: list[dict]):
      response = await openai_client.chat.completions.create(
          messages=messages, model=model_name, stream=True
      )
      role = None
      async for chunk in response:
          delta = chunk.choices[0].delta

          if delta.role is not None:
              role = delta.role

          if delta.content:
              yield {"role": role, "content": delta.content}


  # 这是我们的工具
  async def get_items(place: str) -> str:
      """使用此工具列出你被询问的地方可能找到的物品。"""
      writer = get_stream_writer()
      response = ""
      async for msg_chunk in stream_tokens(
          model_name,
          [
              {
                  "role": "user",
                  "content": (
                      "你能告诉我以下地方可能找到哪些物品吗：'{place}'。"
                      "至少列出 3 个这样的物品，用逗号分隔。"
                      "并为每个物品提供简要描述。"
                  ),
              }
          ],
      ):
          response += msg_chunk["content"]
          writer(msg_chunk)

      return response


  class State(TypedDict):
      messages: Annotated[list[dict], operator.add]


  # 这是调用工具的图节点
  async def call_tool(state: State):
      ai_message = state["messages"][-1]
      tool_call = ai_message["tool_calls"][-1]

      function_name = tool_call["function"]["name"]
      if function_name != "get_items":
          raise ValueError(f"不支持工具 {function_name}")

      function_arguments = tool_call["function"]["arguments"]
      arguments = json.loads(function_arguments)

      function_response = await get_items(**arguments)
      tool_message = {
          "tool_call_id": tool_call["id"],
          "role": "tool",
          "name": function_name,
          "content": function_response,
      }
      return {"messages": [tool_message]}


  graph = (
      StateGraph(State)
      .add_node(call_tool)
      .add_edge(START, "call_tool")
      .compile()
  )
  ```

  让我们用一个包含工具调用的 AI 消息调用图：

  ```python theme={null}
  inputs = {
      "messages": [
          {
              "content": None,
              "role": "assistant",
              "tool_calls": [
                  {
                      "id": "1",
                      "function": {
                          "arguments": '{"place":"卧室"}',
                          "name": "get_items",
                      },
                      "type": "function",
                  }
              ],
          }
      ]
  }

  async for chunk in graph.astream(
      inputs,
      stream_mode="custom",
  ):
      print(chunk["content"], end="|", flush=True)
  ```
</Accordion>

## 为特定聊天模型禁用流式传输

如果你的应用程序混合了支持流式传输和不支持流式传输的模型，你可能需要显式地为不支持流式传输的模型禁用流式传输。

在初始化模型时设置 `disable_streaming=True`。

<Tabs>
  <Tab title="init_chat_model">
    ```python {highlight={5}} theme={null}
    from langchain.chat_models import init_chat_model

    model = init_chat_model(
        "anthropic:claude-3-7-sonnet-latest",
        disable_streaming=True # (1)!
    )
    ```

    1. 设置 `disable_streaming=True` 以禁用聊天模型的流式传输。
  </Tab>

  <Tab title="聊天模型接口">
    ```python theme={null}
    from langchain_openai import ChatOpenAI

    llm = ChatOpenAI(model="o1-preview", disable_streaming=True) # (1)!
    ```

    1. 设置 `disable_streaming=True` 以禁用聊天模型的流式传输。
  </Tab>
</Tabs>

<a id="async" />

### Python \< 3.11 的异步

在 Python 版本 \< 3.11 中，[asyncio 任务](https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task)不支持 `context` 参数。
这限制了 LangGraph 自动传播上下文的能力，并以两种关键方式影响 LangGraph 的流式传输机制：

1. 你**必须**在异步 LLM 调用（例如，`ainvoke()`）中显式传递 [`RunnableConfig`](https://python.langchain.com/docs/concepts/runnables/#runnableconfig)，因为回调不会自动传播。
2. 你**不能**在异步节点或工具中使用 `get_stream_writer()` —— 你必须直接传递一个 `writer` 参数。

<Accordion title="扩展示例：带手动配置的异步 LLM 调用">
  ```python {highlight={16,29}} theme={null}
  from typing import TypedDict
  from langgraph.graph import START, StateGraph
  from langchain.chat_models import init_chat_model

  llm = init_chat_model(model="openai:gpt-4o-mini")

  class State(TypedDict):
      topic: str
      joke: str

  async def call_model(state, config): # (1)!
      topic = state["topic"]
      print("正在生成笑话...")
      joke_response = await llm.ainvoke(
          [{"role": "user", "content": f"写一个关于 {topic} 的笑话"}],
          config, # (2)!
      )
      return {"joke": joke_response.content}

  graph = (
      StateGraph(State)
      .add_node(call_model)
      .add_edge(START, "call_model")
      .compile()
  )

  async for chunk, metadata in graph.astream(
      {"topic": "ice cream"},
      stream_mode="messages", # (3)!
  ):
      if chunk.content:
          print(chunk.content, end="|", flush=True)
  ```

  1. 在异步节点函数中接受 `config` 作为参数。
  2. 将 `config` 传递给 `llm.ainvoke()` 以确保上下文正确传播。
  3. 设置 `stream_mode="messages"` 以流式传输 LLM 令牌。
</Accordion>

<Accordion title="扩展示例：带流式写入器的异步自定义流式传输">
  ```python {highlight={8,21}} theme={null}
  from typing import TypedDict
  from langgraph.types import StreamWriter

  class State(TypedDict):
        topic: str
        joke: str

  async def generate_joke(state: State, writer: StreamWriter): # (1)!
        writer({"custom_key": "在生成笑话时流式传输自定义数据"})
        return {"joke": f"这是一个关于 {state['topic']} 的笑话"}

  graph = (
        StateGraph(State)
        .add_node(generate_joke)
        .add_edge(START, "generate_joke")
        .compile()
  )

  async for chunk in graph.astream(
        {"topic": "ice cream"},
        stream_mode="custom", # (2)!
  ):
        print(chunk)
  ```

  1. 在异步节点或工具的函数签名中添加 `writer` 作为参数。LangGraph 将自动将流式写入器传递给函数。
  2. 设置 `stream_mode="custom"` 以在流中接收自定义数据。
</Accordion>