流式输出

要点

上一篇我们把图跑起来了，但调用方式还是 invoke
stream 本身不是固定只有一种输出形式
官方 streaming 能按不同模式暴露节点更新、完整状态、LLM token 或更底层事件
放到 ReAct Agent 里以后，最常见的需求有两种
streamEvents(...) 不是不能用，而是它更像一层更底的事件视角
下面这段代码沿用上一篇的 ReAct Agent，只把调用方式换成流式

内容

1. 为什么要讲流式输出

上一篇我们把图跑起来了，但调用方式还是 invoke。

invoke 的特点很直接：等整张图执行完，再一次性拿到结果。

如果图里只有一两个节点，这种方式没什么问题。可一旦图里有模型节点、工具节点，还要来回走几轮，用户面对的就是一段很长的等待。页面没有变化，终端也没有输出，只能干等。

流式输出就是把这段等待拆开。

图在运行过程中，每往前走一步，你就先拿到一步的结果。模型开始检索资料时，可以先显示「正在查参考资料」；工具跑完时，可以先把资料摘要打出来；模型开始生成草稿时，也可以边生成边往前端推。

这一篇先讲 graph.stream(...)。

它已经覆盖了大部分日常场景。streamEvents(...) 还会讲，但放在后面，当作更细粒度的补充。

我对照了 LangGraph 官方 streaming 文档。官方把 streaming 拆成几类视角：你可以看每个节点刚返回的 updates，也可以看每一步合并后的 values，还可以看模型 token、debug 信息或更底层事件。也就是说，stream 不是「把最终答案拆成几段文字」这么简单，它更像把图执行过程打开，让外部系统边跑边观察。

放到内容创作智能体里，可以这样理解：

stream 视角	你看到什么	适合展示什么
`updates`	某个节点刚返回的状态更新	「正在查资料」「审核节点已返回」
`values`	当前完整 state	调试整条创作流程状态
token / messages	模型逐 token 输出	草稿正文逐步出现在页面上
events	更细粒度的运行事件	做 tracing、调试和复杂前端反馈

2. 先看最常用的 stream

graph.stream(...) 返回的是一个异步迭代器。

你可以把它理解成一条持续吐数据的管道，图每产生一段新结果，你就从这条管道里读一段出来。

最常见的写法是这样：

typescript

// stream-basic.ts
// 先创建一条流。这里还没有开始逐段消费，只是拿到一个可迭代的流对象。

const stream = await graph.stream(
  {
    messages: [
      { role: 'user', content: '帮我写一篇 LangGraph 入门文章，先查资料。' },
    ],
  },

  {
    configurable: { thread_id: 'stream-001' },

    // updates 表示：每个节点执行完后，把这个节点刚刚返回的状态更新推出来。

    streamMode: 'updates',
  },
)

// 开始一段一段读取结果，是从 for await...of 这里开始。

for await (const chunk of stream) {
  console.log(chunk)
}

这里有两个点先记住就够了。

第一，graph.stream(...) 这一步是在创建流。

第二，一段一段读结果，是后面的 for await...of。

刚接触这套 API 时，很多人会疑惑：既然都 stream 了，为什么前面还有一个 await？

原因就是这里分了两步：

先异步创建一个可消费的流对象
再逐段把里面的数据读出来

3. streamMode 才是这篇的重点

stream 本身不是固定只有一种输出形式。

决定你拿到什么内容的，是 streamMode。

这一层想清楚了，后面的用法就会顺很多。

3.1 updates：按节点看状态更新

updates 是最适合先学的模式。

图里每个节点执行完以后，LangGraph 会把这个节点刚刚返回的状态更新推给你。

typescript

// stream-updates.ts
const stream = await graph.stream(
  {
    messages: [
      { role: 'user', content: '帮我写一篇 LangGraph 入门文章，先查资料。' },
    ],
  },

  {
    configurable: { thread_id: 'stream-002' },

    // 只关心“这一小步刚刚更新了什么”。

    streamMode: 'updates',
  },
)

for await (const chunk of stream) {
  // chunk 的结构一般是：{ 节点名: 该节点这一步返回的状态更新 }

  console.log('--- 收到更新 ---')

  console.log(chunk)
}

输出会是这种结构：

txt

// stream-updates-output.txt
--- 收到更新 ---

{

  callModel: {

    messages: [AIMessage { content: '', tool_calls: [{ name: 'search_references', args: { topic: 'LangGraph' } }] }]

  }

}

--- 收到更新 ---

{

  callTools: {

    messages: [ToolMessage { content: '官方文档、StateGraph 示例、人工审核工作流案例' }]

  }

}

--- 收到更新 ---

{

  callModel: {

    messages: [AIMessage { content: '我先按「为什么需要 LangGraph、核心概念、最小示例」组织一版大纲。' }]

  }

}

这种模式最适合看图到底执行到了哪一步。

读这个输出时，可以按下面这个顺序去理解：

第一个 chunk 是 callModel。模型先看了一眼用户问题，然后决定调 search_references。

第二个 chunk 是 callTools。工具跑完，把参考资料写回状态。

第三个 chunk 又回到 callModel。模型拿到工具结果以后，才生成最终回复。

如果你要在页面上显示：

正在调用哪个工具
哪个节点已经执行完
工具返回了什么结果

updates 就已经够用了。

3.2 values：看完整状态快照

updates 给的是“这一小步刚改了什么”。

values 给的则是“走到当前这一步时，整份状态长什么样”。

typescript

// stream-values.ts
const stream = await graph.stream(
  {
    messages: [
      { role: 'user', content: '帮我写一篇 LangGraph 入门文章，先查资料。' },
    ],
  },

  {
    configurable: { thread_id: 'stream-003' },

    // values 表示：每一步都给你一份完整状态，而不只是增量。

    streamMode: 'values',
  },
)

for await (const state of stream) {
  // 这里拿到的就是当前这一步结束后的完整 state。

  const lastMsg = state.messages.at(-1)

  console.log('当前消息总数:', state.messages.length)

  console.log('最后一条消息:', lastMsg)
}

这个模式更适合调试。

因为你拿到的是完整状态，不需要自己把前面的更新再拼回去。

刚开始学的时候，updates 和 values 最容易混。可以直接这样分：

updates 看增量
values 看全量

3.3 messages：拿模型流式生成的内容

如果你要做打字机效果，重点不是 streamEvents，而是先看 streamMode: "messages"。

它会把模型生成过程里的消息块持续推出来。

对前端来说，这通常就是最直接的流式输出入口。

typescript

// stream-messages.ts
const stream = await graph.stream(
  { messages: [{ role: 'user', content: '用 50 字介绍一下 LangGraph' }] },

  {
    configurable: { thread_id: 'stream-004' },

    // messages 表示：把模型生成过程里的消息块持续推出来。

    streamMode: 'messages',
  },
)

for await (const chunk of stream) {
  // messages 模式下，每次拿到的是一个二元组：

  // [消息块, 元信息]

  const [messageChunk, meta] = chunk

  // 一张图里可能不止一个节点会往外吐消息。

  // 这里先只接模型节点的输出。

  if (meta.langgraph_node === 'callModel') {
    const token = messageChunk.content

    if (typeof token === 'string' && token) {
      // 每来一小块就先输出一小块，终端里看到的就是流式效果。

      process.stdout.write(token)
    }
  }
}

console.log()

这一段的关键不在“记住返回值长什么样”，而在知道它是用来做什么的：

updates 更像图执行进度
messages 更像模型回复本身的流

所以如果你只是想做“边生成边显示”，完全可以先学 streamMode: "messages"，不用一上来就跳到事件流。

4. 在 ReAct Agent 里怎么消费这些流

放到 ReAct Agent 里以后，最常见的需求有两种。

第一种是看 Agent 现在做到哪一步。

第二种是让用户看到回复正在生成。

这两种需求，通常分别对应：

updates
messages

先看 updates 的消费方式：

typescript

// stream-consume-updates.ts
const stream = await graph.stream(
  {
    messages: [
      {
        role: 'user',
        content: '帮我写一篇 LangGraph 入门文章，顺便估算 1500 字读多久。',
      },
    ],
  },

  {
    configurable: { thread_id: 'stream-005' },

    streamMode: 'updates',
  },
)

for await (const chunk of stream) {
  if ('callModel' in chunk) {
    // callModel 这一步结束后，最后一条消息通常就是模型刚刚产出的那条。

    const lastMsg = chunk.callModel.messages.at(-1)

    if (lastMsg?.tool_calls?.length) {
      // 如果模型产出的是 tool_calls，说明它这一步不是在回答用户，

      // 而是在决定下一步要调用哪些工具。

      const toolNames = lastMsg.tool_calls.map((tc) => tc.name).join(', ')

      console.log(`🔧 正在调用: ${toolNames}`)
    } else if (lastMsg?.content) {
      // 如果没有 tool_calls，而是直接有文本内容，

      // 这通常就是模型已经开始生成最终回复了。

      console.log(`💬 ${lastMsg.content}`)
    }
  }

  if ('callTools' in chunk) {
    // 工具节点结束后，可以把工具结果先展示出来，

    // 不用等模型把整轮回复全组织完。

    for (const msg of chunk.callTools.messages) {
      console.log(`📋 工具结果: ${msg.content}`)
    }
  }
}

这段代码适合挂在后端日志里，也适合给前端推“阶段性进展”。

再看 messages 的消费方式：

typescript

// stream-consume-messages.ts
const stream = await graph.stream(
  {
    messages: [
      { role: 'user', content: '帮我写一篇 LangGraph 入门文章，先查资料。' },
    ],
  },

  {
    configurable: { thread_id: 'stream-006' },

    streamMode: 'messages',
  },
)

for await (const [messageChunk, meta] of stream) {
  // 这里只消费模型节点吐出来的文本流。

  // 如果图里还有别的节点也产生消息，可以按 meta 再分流处理。

  if (meta.langgraph_node !== 'callModel') continue

  const token = messageChunk.content

  if (typeof token === 'string' && token) {
    // 每收到一个文本块就往前推一次，聊天界面里就会看到“边生成边显示”。

    process.stdout.write(token)
  }
}

console.log()

这段更接近你在聊天界面里会用到的逻辑。

用户不会先看到整段最终回复，而是会看到它一点点出现。

5. streamEvents 放在什么时候用

streamEvents(...) 不是不能用，而是它更像一层更底的事件视角。

当你已经不满足于：

看节点更新
看模型文本流

而是还想知道：

工具什么时候开始
工具什么时候结束
哪个 runnable 正在开始
哪个 runnable 已经结束

这时候再上 streamEvents(...) 会更合适。

typescript

// stream-events.ts
const eventStream = graph.streamEvents(
  {
    messages: [
      { role: 'user', content: '帮我写一篇 LangGraph 入门文章，先查资料。' },
    ],
  },

  {
    version: 'v2',

    configurable: { thread_id: 'events-001' },
  },
)

for await (const event of eventStream) {
  if (event.event === 'on_tool_start') {
    // 工具开始执行时，这里会先收到一条事件。

    console.log(`🔧 开始调用 ${event.name}`)
  }

  if (event.event === 'on_tool_end') {
    // 工具结束后，再收到一条结束事件。

    console.log(`✅ ${event.name} 已完成`)
  }

  if (event.event === 'on_chat_model_stream') {
    // 如果你想要最细的模型流式事件，可以在这里接 token。

    const token = event.data.chunk?.content

    if (typeof token === 'string' && token) {
      process.stdout.write(token)
    }
  }
}

console.log()

这一层的优点是细。

缺点也一样明显，就是你得自己过滤事件、自己分发处理。

所以如果只是做一个普通聊天界面，先用 stream + streamMode 通常会更顺。

6. 完整例子：同一张图，三种流式方式

下面这段代码沿用上一篇的 ReAct Agent，只把调用方式换成流式。

typescript

// streaming-agent.ts
import {
  StateGraph,
  StateSchema,
  MessagesValue,
  START,
  END,
} from '@langchain/langgraph'

import type { GraphNode, ConditionalEdgeRouter } from '@langchain/langgraph'

import { InMemorySaver } from '@langchain/langgraph'

import { ChatOpenAI } from '@langchain/openai'

import { tool } from '@langchain/core/tools'

import * as z from 'zod'

const searchReferences = tool(
  async ({ topic }) => {
    const data: Record<string, string[]> = {
      'LangGraph 入门': ['官方文档', 'StateGraph 示例', '人工审核工作流案例'],

      LangGraph: ['官方文档', 'StateGraph 示例', '人工审核工作流案例'],

      ReAct: ['工具调用循环', '模型决策', '停止条件'],

      Checkpointer: ['线程状态保存', '中断恢复', '历史快照回溯'],
    }

    return data[topic]?.join('、') ?? `暂无 ${topic} 的资料`
  },

  {
    name: 'search_references',

    description: '为技术文章检索参考资料',

    schema: z.object({ topic: z.string().describe('文章主题或概念名') }),
  },
)

const estimateReadingTime = tool(
  async ({ wordCount }) => {
    const minutes = Math.ceil(wordCount / 500)

    return `约 ${minutes} 分钟读完`
  },

  {
    name: 'estimate_reading_time',

    description: '根据字数估算文章阅读时间',

    schema: z.object({ wordCount: z.number().describe('文章预计字数') }),
  },
)

const tools = [searchReferences, estimateReadingTime]

const toolsByName = Object.fromEntries(tools.map((t) => [t.name, t]))

// 模型先绑定工具。后面模型节点里只负责看消息、决定是回答还是调工具。

const model = new ChatOpenAI({ model: 'gpt-4.1-mini' }).bindTools(tools)

const State = new StateSchema({
  messages: MessagesValue,
})

const callModel: GraphNode<typeof State> = async (state) => {
  // 把当前累积的消息都交给模型，让它决定下一步是直接回答还是继续调工具。

  const response = await model.invoke(state.messages)

  return { messages: [response] }
}

const callTools: GraphNode<typeof State> = async (state) => {
  // 工具节点只看最后一条模型消息里的 tool_calls。

  const lastMsg = state.messages.at(-1)!

  const toolCalls = lastMsg.tool_calls ?? []

  const results = await Promise.all(
    toolCalls.map(async (tc) => {
      try {
        // 按工具名取出对应工具，再把参数传进去执行。

        const result = await toolsByName[tc.name].invoke(tc.args)

        return { role: 'tool' as const, content: result, tool_call_id: tc.id }
      } catch (err) {
        const errorMsg = err instanceof Error ? err.message : String(err)

        return {
          role: 'tool' as const,

          content: `工具执行失败: ${errorMsg}`,

          tool_call_id: tc.id,
        }
      }
    }),
  )

  return { messages: results }
}

const shouldContinue: ConditionalEdgeRouter<typeof State, 'callTools'> = (
  state,
) => {
  // 如果最后一条模型消息里还带着 tool_calls，就继续走工具节点；

  // 否则说明模型已经准备好给最终回复，可以结束整张图。

  const lastMsg = state.messages.at(-1)!

  if (lastMsg.tool_calls && lastMsg.tool_calls.length > 0) {
    return 'callTools'
  }

  return END
}

const graph = new StateGraph(State)

  .addNode('callModel', callModel)

  .addNode('callTools', callTools)

  .addEdge(START, 'callModel')

  .addConditionalEdges('callModel', shouldContinue, ['callTools'])

  .addEdge('callTools', 'callModel')

  .compile({ checkpointer: new InMemorySaver() })

const config = { configurable: { thread_id: 'streaming-demo' } }

console.log('=== updates ===')

const updatesStream = await graph.stream(
  {
    messages: [
      { role: 'user', content: '帮我写一篇 LangGraph 入门文章，先查资料。' },
    ],
  },

  {
    ...config,

    streamMode: 'updates',
  },
)

for await (const chunk of updatesStream) {
  // updates 最适合先看图到底是怎么往前推进的。

  console.log(chunk)
}

console.log('\n=== messages ===')

const messagesStream = await graph.stream(
  { messages: [{ role: 'user', content: '再补一版面向初学者的开头。' }] },

  {
    ...config,

    streamMode: 'messages',
  },
)

for await (const [messageChunk, meta] of messagesStream) {
  // 这里只展示模型节点的文本流，不把别的节点混进来。

  if (meta.langgraph_node !== 'callModel') continue

  const token = messageChunk.content

  if (typeof token === 'string' && token) {
    process.stdout.write(token)
  }
}

console.log('\n\n=== streamEvents ===')

const eventStream = graph.streamEvents(
  { messages: [{ role: 'user', content: '顺便估算 1500 字读多久。' }] },

  {
    version: 'v2',

    configurable: { thread_id: 'streaming-demo' },
  },
)

for await (const event of eventStream) {
  // streamEvents 更适合看一条运行事件到底是在什么时候发生的。

  if (event.event === 'on_tool_start') {
    console.log(`\n🔧 ${event.name} 开始执行`)
  }

  if (event.event === 'on_tool_end') {
    console.log(`✅ ${event.name} 执行完成`)
  }
}

这一段最重要的不是把输出背下来，而是看清一件事：

图完全没变，节点也没变，边也没变。变化只发生在调用方式上。

同一张图：

invoke 是等全部跑完再拿结果
stream(..., { streamMode: "updates" }) 是看每一步状态更新
stream(..., { streamMode: "messages" }) 是看模型生成过程
streamEvents(...) 是看更细的运行事件

7. 这篇先记住什么

如果你现在刚接触 LangGraph 的流式输出，先把下面这层关系记住就够了。

stream 是正文里最该先会的入口。

决定你拿到什么的，是 streamMode。

一般可以这样选：

想看图走到了哪一步，用 updates
想看完整状态快照，用 values
想做打字机效果，用 messages
想看更细的运行事件，再用 streamEvents

下一篇再继续往下讲 Command 与 Send。到那时，图不只是“边跑边输出”，还会在运行过程中动态改变下一步要走到哪里。

流式输出 ​

要点 ​

内容 ​

1. 为什么要讲流式输出 ​

2. 先看最常用的 stream ​

3. streamMode 才是这篇的重点 ​

3.1 updates：按节点看状态更新 ​

3.2 values：看完整状态快照 ​

3.3 messages：拿模型流式生成的内容 ​

4. 在 ReAct Agent 里怎么消费这些流 ​

5. streamEvents 放在什么时候用 ​

6. 完整例子：同一张图，三种流式方式 ​

7. 这篇先记住什么 ​

流式输出

要点

内容

1. 为什么要讲流式输出

2. 先看最常用的 stream

3. streamMode 才是这篇的重点

3.1 updates：按节点看状态更新

3.2 values：看完整状态快照

3.3 messages：拿模型流式生成的内容

4. 在 ReAct Agent 里怎么消费这些流

5. streamEvents 放在什么时候用

6. 完整例子：同一张图，三种流式方式

7. 这篇先记住什么