```
### Python \< 3.11 的异步处理
在 Python \< 3.11 版本中，[asyncio tasks](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-4.1-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 令牌
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"])
```
***
[Edit this page on GitHub](https://github.com/langchain-ai/docs/edit/main/src/i18n\zh-CN\oss\langgraph\streaming.mdx) or [file an issue](https://github.com/langchain-ai/docs/issues/new/choose).
[Connect these docs](/use-these-docs) to Claude, VSCode, and more via MCP for real-time answers.