流式传输 - LangChain 文档

LangGraph 实现了一个流系统来提供实时更新。流对于增强基于 LLM 的应用程序的响应能力至关重要。通过逐步显示输出，即使在完整响应准备好之前，流也能显著改善用户体验（UX），尤其是在处理 LLM 的延迟时。 LangGraph 流可以实现：

流式图状态 — 使用 updates 和 values 模式获取状态更新/值。
流式子图输出 — 包括父图和任何嵌套子图的输出。
流式 LLM 令牌 — 从任何地方捕获令牌流：节点内部、子图或工具。
流式自定义数据 — 直接从工具函数发送自定义更新或进度信号。
使用多种流模式 — 从 values（完整状态）、updates（状态增量）、messages（LLM 令牌 + 元数据）、custom（任意用户数据）或 debug（详细跟踪）中选择。

支持的流模式

将以下一个或多个流模式作为列表传递给 stream 或 astream 方法

模式	描述
`值`	在图的每个步骤后流式传输状态的完整值。
`更新`	在图的每个步骤后流式传输状态的更新。如果同一步骤中进行了多次更新（例如，运行了多个节点），则这些更新会单独流式传输。
`自定义`	从图节点内部流式传输自定义数据。
`消息`	从任何调用 LLM 的图节点流式传输 2 元组（LLM 令牌，元数据）。
`调试`	在图执行过程中尽可能多地流式传输信息。

基本用法示例

LangGraph 图公开了 stream (同步) 和 astream (异步) 方法，以迭代器形式生成流式输出。

for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)

扩展示例：流式更新

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()
)

# The stream() method returns an iterator that yields streamed outputs
for chunk in graph.stream(  
    {"topic": "ice cream"},
    # Set stream_mode="updates" to stream only the updates to the graph state after each node
    # Other stream modes are also available. See supported stream modes for details
    stream_mode="updates",  
):
    print(chunk)

{'refineTopic': {'topic': 'ice cream and cats'}}
{'generateJoke': {'joke': 'This is a joke about ice cream and cats'}}

流式传输多种模式

您可以将列表作为 stream_mode 参数传递，以同时流式传输多种模式。流式输出将是 (mode, chunk) 元组，其中 mode 是流模式的名称，chunk 是该模式流式传输的数据。

for mode, chunk in graph.stream(inputs, stream_mode=["updates", "custom"]):
    print(chunk)

流式图状态

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

updates 在图的每个步骤后流式传输状态的更新。
values 在图的每个步骤后流式传输状态的完整值。

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(
    {"topic": "ice cream"},
    stream_mode="updates",  
):
    print(chunk)

流式子图输出

要在流式输出中包含来自子图的输出，您可以在父图的 .stream() 方法中设置 subgraphs=True。这将流式传输来自父图和任何子图的输出。输出将作为元组 (namespace, data) 流式传输，其中 namespace 是一个包含调用子图的节点路径的元组，例如 ("parent_node:<task_id>", "child_node:<task_id>")。

for chunk in graph.stream(
    {"foo": "foo"},
    # Set subgraphs=True to stream outputs from subgraphs
    subgraphs=True,  
    stream_mode="updates",
):
    print(chunk)

扩展示例：从子图流式传输

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

# Define subgraph
class SubgraphState(TypedDict):
    foo: str  # note that this key is shared with the parent graph state
    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()

# Define parent graph
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",
    # Set subgraphs=True to stream outputs from subgraphs
    subgraphs=True,  
):
    print(chunk)

((), {'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'}})

注意，我们不仅接收节点更新，还接收命名空间，它告诉我们正在从哪个图（或子图）进行流式传输。

调试

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

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="debug",  
):
    print(chunk)

LLM Tokens

使用 messages 流模式，从图的任何部分（包括节点、工具、子图或任务）逐个令牌地流式传输大型语言模型 (LLM) 输出。来自 messages 模式的流式输出是元组 (message_chunk, metadata)，其中：

message_chunk：来自 LLM 的令牌或消息段。
metadata：一个包含图节点和 LLM 调用详细信息的字典。

如果您的 LLM 不作为 LangChain 集成提供，您可以使用 custom 模式流式传输其输出。有关详细信息，请参阅与任何 LLM 一起使用。

Python < 3.11 异步所需的手动配置 在 Python < 3.11 中使用异步代码时，您必须显式地将 RunnableConfig 传递给 ainvoke() 以启用正确的流式传输。有关详细信息，请参阅 Python < 3.11 的异步或升级到 Python 3.11+。

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 = ""


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

def call_model(state: MyState):
    """Call the LLM to generate a joke about a topic"""
    # Note that message events are emitted even when the LLM is run using .invoke rather than .stream
    model_response = model.invoke(  
        [
            {"role": "user", "content": f"Generate a joke about {state.topic}"}
        ]
    )
    return {"joke": model_response.content}

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

# The "messages" stream mode returns an iterator of tuples (message_chunk, metadata)
# where message_chunk is the token streamed by the LLM and metadata is a dictionary
# with information about the graph node where the LLM was called and other information
for message_chunk, metadata in graph.stream(
    {"topic": "ice cream"},
    stream_mode="messages",  
):
    if message_chunk.content:
        print(message_chunk.content, end="|", flush=True)

按 LLM 调用过滤

您可以将 tags 与 LLM 调用相关联，以根据 LLM 调用过滤流式令牌。

from langchain.chat_models import init_chat_model

# model_1 is tagged with "joke"
model_1 = init_chat_model(model="gpt-4o-mini", tags=['joke'])
# model_2 is tagged with "poem"
model_2 = init_chat_model(model="gpt-4o-mini", tags=['poem'])

graph = ... # define a graph that uses these LLMs

# The stream_mode is set to "messages" to stream LLM tokens
# The metadata contains information about the LLM invocation, including the tags
async for msg, metadata in graph.astream(
    {"topic": "cats"},
    stream_mode="messages",  
):
    # Filter the streamed tokens by the tags field in the metadata to only include
    # the tokens from the LLM invocation with the "joke" tag
    if metadata["tags"] == ["joke"]:
        print(msg.content, end="|", flush=True)

扩展示例：按标签过滤

from typing import TypedDict

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

# The joke_model is tagged with "joke"
joke_model = init_chat_model(model="gpt-4o-mini", tags=["joke"])
# The poem_model is tagged with "poem"
poem_model = init_chat_model(model="gpt-4o-mini", tags=["poem"])


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


async def call_model(state, config):
      topic = state["topic"]
      print("Writing joke...")
      # Note: Passing the config through explicitly is required for python < 3.11
      # Since context var support wasn't added before then: https://docs.pythonlang.cn/3/library/asyncio-task.html#creating-tasks
      # The config is passed through explicitly to ensure the context vars are propagated correctly
      # This is required for Python < 3.11 when using async code. Please see the async section for more details
      joke_response = await joke_model.ainvoke(
            [{"role": "user", "content": f"Write a joke about {topic}"}],
            config,
      )
      print("\n\nWriting poem...")
      poem_response = await poem_model.ainvoke(
            [{"role": "user", "content": f"Write a short poem about {topic}"}],
            config,
      )
      return {"joke": joke_response.content, "poem": poem_response.content}


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

# The stream_mode is set to "messages" to stream LLM tokens
# The metadata contains information about the LLM invocation, including the tags
async for msg, metadata in graph.astream(
      {"topic": "cats"},
      stream_mode="messages",
):
    if metadata["tags"] == ["joke"]:
        print(msg.content, end="|", flush=True)

按节点过滤

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

# The "messages" stream mode returns a tuple of (message_chunk, metadata)
# where message_chunk is the token streamed by the LLM and metadata is a dictionary
# with information about the graph node where the LLM was called and other information
for msg, metadata in graph.stream(
    inputs,
    stream_mode="messages",  
):
    # Filter the streamed tokens by the langgraph_node field in the metadata
    # to only include the tokens from the specified node
    if msg.content and metadata["langgraph_node"] == "some_node_name":
        ...

扩展示例：从特定节点流式传输 LLM 令牌

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"Write a joke about {topic}"}]
      )
      return {"joke": joke_response.content}


def write_poem(state: State):
      topic = state["topic"]
      poem_response = model.invoke(
            [{"role": "user", "content": f"Write a short poem about {topic}"}]
      )
      return {"poem": poem_response.content}


graph = (
      StateGraph(State)
      .add_node(write_joke)
      .add_node(write_poem)
      # write both the joke and the poem concurrently
      .add_edge(START, "write_joke")
      .add_edge(START, "write_poem")
      .compile()
)

# The "messages" stream mode returns a tuple of (message_chunk, metadata)
# where message_chunk is the token streamed by the LLM and metadata is a dictionary
# with information about the graph node where the LLM was called and other information
for msg, metadata in graph.stream(
    {"topic": "cats"},
    stream_mode="messages",  
):
    # Filter the streamed tokens by the langgraph_node field in the metadata
    # to only include the tokens from the write_poem node
    if msg.content and metadata["langgraph_node"] == "write_poem":
        print(msg.content, end="|", flush=True)

流式自定义数据

要从 LangGraph 节点或工具内部发送自定义用户定义数据，请遵循以下步骤

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

Python < 3.11 异步中没有 get_stream_writer 在 Python < 3.11 上运行的异步代码中，get_stream_writer 将不起作用。相反，将 writer 参数添加到您的节点或工具中并手动传递它。有关用法示例，请参阅 Python < 3.11 的异步。

节点
工具

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):
    # Get the stream writer to send custom data
    writer = get_stream_writer()
    # Emit a custom key-value pair (e.g., progress update)
    writer({"custom_key": "Generating custom data inside node"})
    return {"answer": "some data"}

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

inputs = {"query": "example"}

# Set stream_mode="custom" to receive the custom data in the stream
for chunk in graph.stream(inputs, stream_mode="custom"):
    print(chunk)

与任何 LLM 一起使用

您可以使用 stream_mode="custom" 从任何 LLM API 流式传输数据——即使该 API 没有实现 LangChain 聊天模型接口。这使您可以集成提供自己的流式接口的原始 LLM 客户端或外部服务，从而使 LangGraph 对于自定义设置具有高度灵活性。

from langgraph.config import get_stream_writer

def call_arbitrary_model(state):
    """Example node that calls an arbitrary model and streams the output"""
    # Get the stream writer to send custom data
    writer = get_stream_writer()  
    # Assume you have a streaming client that yields chunks
    # Generate LLM tokens using your custom streaming client
    for chunk in your_custom_streaming_client(state["topic"]):
        # Use the writer to send custom data to the stream
        writer({"custom_llm_chunk": chunk})  
    return {"result": "completed"}

graph = (
    StateGraph(State)
    .add_node(call_arbitrary_model)
    # Add other nodes and edges as needed
    .compile()
)
# Set stream_mode="custom" to receive the custom data in the stream
for chunk in graph.stream(
    {"topic": "cats"},
    stream_mode="custom",  

):
    # The chunk will contain the custom data streamed from the llm
    print(chunk)

扩展示例：流式传输任意聊天模型

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}


# this is our tool
async def get_items(place: str) -> str:
    """Use this tool to list items one might find in a place you're asked about."""
    writer = get_stream_writer()
    response = ""
    async for msg_chunk in stream_tokens(
        model_name,
        [
            {
                "role": "user",
                "content": (
                    "Can you tell me what kind of items "
                    f"i might find in the following place: '{place}'. "
                    "List at least 3 such items separating them by a comma. "
                    "And include a brief description of each item."
                ),
            }
        ],
    ):
        response += msg_chunk["content"]
        writer(msg_chunk)

    return response


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


# this is the tool-calling graph node
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"Tool {function_name} not supported")

    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()
)

让我们使用包含工具调用的 AIMessage 来调用图

inputs = {
    "messages": [
        {
            "content": None,
            "role": "assistant",
            "tool_calls": [
                {
                    "id": "1",
                    "function": {
                        "arguments": '{"place":"bedroom"}',
                        "name": "get_items",
                    },
                    "type": "function",
                }
            ],
        }
    ]
}

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

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

如果您的应用程序混合了支持流式传输的模型和不支持流式传输的模型，您可能需要为不支持流式传输的模型显式禁用流式传输。在初始化模型时设置 disable_streaming=True。

init_chat_model
聊天模型接口

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "claude-sonnet-4-5-20250929",
    # Set disable_streaming=True to disable streaming for the chat model
    disable_streaming=True

)

Python < 3.11 的异步

在 Python 版本 < 3.11 中，asyncio 任务不支持 context 参数。这限制了 LangGraph 自动传播上下文的能力，并以两种关键方式影响 LangGraph 的流式传输机制

您必须显式将 RunnableConfig 传递到异步 LLM 调用中（例如，ainvoke()），因为回调不会自动传播。
您不能在异步节点或工具中使用 get_stream_writer — 您必须直接传递 writer 参数。

扩展示例：使用手动配置的异步 LLM 调用

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

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

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

# Accept config as an argument in the async node function
async def call_model(state, config):
    topic = state["topic"]
    print("Generating joke...")
    # Pass config to model.ainvoke() to ensure proper context propagation
    joke_response = await model.ainvoke(  
        [{"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()
)

# Set stream_mode="messages" to stream LLM tokens
async for chunk, metadata in graph.astream(
    {"topic": "ice cream"},
    stream_mode="messages",  
):
    if chunk.content:
        print(chunk.content, end="|", flush=True)

扩展示例：带流式写入器的异步自定义流式传输

from typing import TypedDict
from langgraph.types import StreamWriter

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

# Add writer as an argument in the function signature of the async node or tool
# LangGraph will automatically pass the stream writer to the function
async def generate_joke(state: State, writer: StreamWriter):  
      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()
)

# Set stream_mode="custom" to receive the custom data in the stream  #
async for chunk in graph.astream(
      {"topic": "ice cream"},
      stream_mode="custom",
):
      print(chunk)

在 GitHub 上编辑此页面源文件。

以编程方式连接这些文档到 Claude、VSCode 等，通过 MCP 获取实时答案。

LangGraph v1.0

入门

能力

生产

LangGraph API

流式处理

支持的流模式

基本用法示例

流式传输多种模式

流式图状态

流式子图输出

调试

LLM Tokens

按 LLM 调用过滤

按节点过滤

流式自定义数据

与任何 LLM 一起使用

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

Python < 3.11 的异步

LangGraph v1.0

入门

能力

生产

LangGraph API

​支持的流模式

​基本用法示例

​流式传输多种模式

​流式图状态

​流式子图输出

​调试

​LLM Tokens

​按 LLM 调用过滤

​按节点过滤

​流式自定义数据

​与任何 LLM 一起使用

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

​Python < 3.11 的异步

支持的流模式

基本用法示例

流式传输多种模式

流式图状态

流式子图输出

调试

LLM Tokens

按 LLM 调用过滤

按节点过滤

流式自定义数据

与任何 LLM 一起使用

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

Python < 3.11 的异步