```
### Python \< 3.11 中的异步
在 Python 版本 \< 3.11 中，[asyncio 任务](https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task)不支持 `context` 参数。
这限制了 LangGraph 自动传播上下文的能力，并以两种关键方式影响 LangGraph 的流式机制：
1. 你**必须**显式地将 [`RunnableConfig`](https://python.langchain.com/docs/concepts/runnables/#runnableconfig) 传递给异步 LLM 调用（例如 `ainvoke()`），因为回调不会自动传播。
2. 你**不能**在异步节点或工具中使用 [`get_stream_writer`](https://reference.langchain.com/python/langgraph/config/get_stream_writer)——你必须直接传递 `writer` 参数。
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from typing import TypedDict
from langgraph.graph import START, StateGraph
from langchain.chat_models import init_chat_model
model = init_chat_model(model="gpt-5.4-mini")
class State(TypedDict):
topic: str
joke: str
# 在异步节点函数中接受 config 作为参数
async def call_model(state, config):
topic = state["topic"]
print("Generating joke...")
# 将 config 传递给 model.ainvoke() 以确保正确的上下文传播
joke_response = await model.ainvoke( # [!code highlight]
[{"role": "user", "content": f"Write a joke about {topic}"}],
config,
)
return {"joke": joke_response.content}
graph = (
StateGraph(State)
.add_node(call_model)
.add_edge(START, "call_model")
.compile()
)
# 设置 stream_mode="messages" 以流式传输 LLM tokens
async for chunk in graph.astream(
{"topic": "ice cream"},
stream_mode="messages", # [!code highlight]
version="v2", # [!code highlight]
):
if chunk["type"] == "messages":
message_chunk, metadata = chunk["data"]
if message_chunk.content:
print(message_chunk.content, end="|", flush=True)
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from typing import TypedDict
from langgraph.types import StreamWriter
class State(TypedDict):
topic: str
joke: str
# 在异步节点或工具的函数签名中添加 writer 作为参数
# LangGraph 将自动将流写入器传递给函数
async def generate_joke(state: State, writer: StreamWriter): # [!code highlight]
writer({"custom_key": "Streaming custom data while generating a joke"})
return {"joke": f"This is a joke about {state['topic']}"}
graph = (
StateGraph(State)
.add_node(generate_joke)
.add_edge(START, "generate_joke")
.compile()
)
# 设置 stream_mode="custom" 以在流中接收自定义数据 # [!code highlight]
async for chunk in graph.astream(
{"topic": "ice cream"},
stream_mode="custom",
version="v2",
):
if chunk["type"] == "custom":
print(chunk["data"])
```
***
[将这些文档连接](/use-these-docs)到 Claude、VSCode 等，通过 MCP 获取实时答案。
[在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/oss/langgraph/streaming.mdx) 或 [提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。