LangChain 流式输出

这篇适合谁、读完能做到什么

这篇适合正在做智能体应用、想让用户看到模型一边想一边写的开发者。读完你能用 stream_mode 选择要看的流：执行进度、逐字输出和工具里自定义的进度信息；会把多种模式合在一个流里；知道推理内容怎么取，以及在什么场景下要关掉某个模型的流式输出。

是什么：把运行中的信息实时推给应用

流式输出（streaming）让应用在智能体运行过程中拿到实时反馈。大模型的响应往往要等一段时间，如果能把输出一段段地显示出来，即使完整回答还没生成好，用户的体验也会好很多。^[1]

LangChain 提供的流式系统一共三种基础模式，用 stream_mode 指定：updates 在智能体每一步之后输出状态更新，同一步里如果有多次更新，会分别输出；messages 输出从调用了大模型的节点里流出来的消息片段，每一项是 token 和元数据组成的元组；custom 输出你在图的节点里通过流写入器（stream writer）写出的任意数据。^[1]

新应用可以优先考虑事件流（event streaming）。它是 LangChain v1.3 引入的按投影划分的接口，消息、状态值、工具调用和子图各有一个独立的迭代器，不必再通过 stream_mode 的分支去判断。^[1]

准备

版本：自定义 custom 模式和 v2 格式对 LangGraph 版本有要求，v2 需要 LangGraph 1.1 或更高版本。^[1]
持久化：如果要用 thread_id 保存多轮对话的历史，智能体必须配置 checkpointer。本地试验可以像示例那样传入 InMemorySaver。^[1]
语言：示例同时给出 Python 和 JavaScript 两套写法，两者的模式名称相同，写法略有不同。^[1]

上手：先看执行进度

对 agent.stream 或 astream 设置 stream_mode="updates"，智能体每走完一步就会输出一次。^[1]
以一次工具调用为例，你会先看到大模型节点发出的工具调用请求，接着是工具节点返回的 ToolMessage，最后是大模型节点给出的最终回答。^[1]
多轮对话时，在 config 里传入 thread_id，后续的请求就会接着同一段历史往下走。^[1]

逐字输出：stream_mode="messages"

要让回答一个字一个字地出来，把模式换成 messages。下面的代码把每个数据块的节点名和内容打印出来：

from langchain.agents import create_agent

def get_weather(city: str) -> str:
    """Get weather for a given city."""

    return f"It's always sunny in {city}!"

agent = create_agent(
    model="gpt-5-nano",
    tools=[get_weather],
)
for chunk in agent.stream(  # [!code highlight]
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    stream_mode="messages",
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "messages":  # [!code highlight]
        token, metadata = chunk["data"]  # [!code highlight]
        print(f"node: {metadata['langgraph_node']}")
        print(f"content: {token.content_blocks}")
        print("\n")

输出里，工具调用的参数会以 tool_call_chunk 的形式一段段到达，比如先是空的参数字符串，然后依次出现 {"、city、":" 这样的片段。工具节点之后，最终回答也是按小段输出的。^[1] 这说明流式输出的粒度很细，你拿到的是模型生成过程中的增量块，而不是完整消息。

工具里自定义进度：stream_mode="custom"

如果工具执行得比较慢，比如要查很多条数据，可以在工具里用 get_stream_writer 写出进度信息。它从 langgraph.config 导入，写出的数据会通过 custom 模式推给应用。^[1]

from langchain.agents import create_agent
from langgraph.config import get_stream_writer  # [!code highlight]

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    writer = get_stream_writer()  # [!code highlight]
    # stream any arbitrary data
    writer(f"Looking up data for city: {city}")
    writer(f"Acquired data for city: {city}")
    return f"It's always sunny in {city}!"

agent = create_agent(
    model="claude-sonnet-4-6",
    tools=[get_weather],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    stream_mode="custom",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "custom":  # [!code highlight]
        print(chunk["data"])  # [!code highlight]

需要注意：一旦工具里用了 get_stream_writer，这个工具就不能在 LangGraph 的执行上下文之外直接调用，否则会出错。^[1]

一个流里放多种模式

同一个流里可以同时要多种模式，写成列表即可，例如 stream_mode=["updates", "custom"]。这时每个数据块都是一个 StreamPart 字典，带有 type、ns 和 data 三个键，type 说明这一块属于哪种模式，data 是内容。^[1] 下面的例子在 v2 格式下同时收到执行状态和自定义进度：

from langchain.agents import create_agent
from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    writer = get_stream_writer()
    writer(f"Looking up data for city: {city}")
    writer(f"Acquired data for city: {city}")
    return f"It's always sunny in {city}!"

agent = create_agent(
    model="gpt-5-nano",
    tools=[get_weather],
)

for chunk in agent.stream(  # [!code highlight]
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    stream_mode=["updates", "custom"],
    version="v2",  # [!code highlight]
):
    print(f"stream_mode: {chunk['type']}")  # [!code highlight]
    print(f"content: {chunk['data']}")  # [!code highlight]
    print("\n")

JavaScript 版本里，把 streamMode 写成数组，每次输出的是一个 [mode, chunk] 元组，前一项是模式名，后一项是数据。^[1]

常见场景

推理内容。 有些模型会在给出答案前先做内部推理。要流式输出这部分内容，用 stream_mode="messages"，再筛选 type 为 "reasoning" 的内容块。前提是模型上已经开启了推理输出。^[1] 不同厂商的格式并不一样，比如 Anthropic 的 thinking 块和 OpenAI 的推理摘要，LangChain 会通过 content_blocks 统一成标准的 "reasoning" 内容块，所以同一段代码对不同模型都适用。^[1]

工具调用。 如果既想看到工具调用的 JSON 是怎么一点点生成的，又想拿到执行时已经解析好的工具调用，可以组合使用两种模式。messages 给出所有大模型调用产生的增量块；如果完整消息被记录在状态里，再加上 updates，就能从状态更新中拿到解析好的完整消息。^[1]

人工审批。 要在流式输出里处理人工介入的中断，做法是：给智能体配置人工介入中间件和 checkpointer，在 "updates" 模式下收集中断，然后用 command 作出回应。^[1]

多个智能体。 如果一个智能体里有多个大模型在工作，需要分清消息来自谁，就在创建每个智能体时传入 name。流式输出 "messages" 模式时，这个名字会出现在元数据的 lc_agent_name 键里。子智能体的输出还需要在创建流时设置 subgraphs=True。^[1]

关闭某个模型的流式输出

有些场景下，你不希望某个模型的输出被逐字推给用户，比如多智能体系统里只让一部分智能体对外输出，或者把应用部署到 LangSmith 上，不想把某些模型的输出推给客户端。这时在初始化模型时设置 streaming=False。^[1] 如果所用的模型集成不支持 streaming 参数，改用 disable_streaming=True，这个参数是所有聊天模型基类都有的。^[1]

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-6.1-sol",
    streaming=False  # [!code highlight]
)

部署到 LangSmith 时，这个设置要写在图的代码里，在部署之前配好。^[1]

作为子图嵌入时要注意

create_agent 返回的是一个编译好的图。把它当作父 StateGraph 的一个节点使用时，它就成了子图。这时父图上的 stream_mode="messages" 不会输出内层智能体大模型调用产生的 token，除非传入 subgraphs=True。^[1]

限制和常见问题

没有 thread_id 和 checkpointer 时，多轮对话的历史不会被保存。示例为简洁起见省略了 thread_id，实际使用时应当传入。^[1]
在工具里用 get_stream_writer 之后，这个工具不能脱离 LangGraph 执行上下文直接调用。^[1]
v2 格式的 invoke() 返回 GraphOutput，状态在 .value 里，中断在 .interrupts 里，中断为空时是空元组。^[1]
推理内容要求模型本身开启了推理输出，否则筛选不到。^[1]