LangGraph中将流式处理列为了核心功能之一，可以通过 stream() 方法来获取流式内容。

在获取流式内容的同时，还可以通过 streamMode 来指定不同的类型：

这节课我们先来看前面2种流类型：

• 值流
• 更新流

值流和更新流

值流

每次节点执行后，直接把整个 state（完整对象）流给你。

特点

• 简单粗暴，适合入门；
• 每一步都能看到整个状态；
• 但 payload 可能比较大。

const stream = await graph.stream(input, {
  streamMode: "values",
});

更新流

只告诉你“变化了什么”，而不是给整份 state。

特点

• 性能更高，输出更小
• 能明确看到某一次节点只更新了哪些字段
• 适合大状态对象（如企业级复杂 workflow）

const stream = await graph.stream(input, {
  streamMode: "updates",
});

消息流

前置知识

之前我们在使用langchain的时候，无论是模型还是链，常用的方法有两个：

1. invoke
2. stream

import { ChatOpenAI } from "@langchain/openai";

const model = new ChatOpenAI({
  model: "gpt-4o-mini",
  streaming: true,
});

async function runInvoke() {
  const result = await model.invoke("简单介绍一下 LangGraph");

  // invoke 会在模型完整生成答案后，一次性返回结果
  console.log(result.content);
}

runInvoke();

import { ChatOpenAI } from "@langchain/openai";

const model = new ChatOpenAI({
  model: "gpt-4o-mini",
  streaming: true,
});

async function runStream() {
  const stream = await model.stream("简单介绍一下 LangGraph");

  for await (const chunk of stream) {
    // 每一次循环，都会拿到模型生成的一部分内容
    process.stdout.write(chunk.content ?? "");
  }
}

runStream();

两个方法的区别：

• invoke：非流式输出
• stream：流式输出

无论是invoke还是stream，大模型那边其实都是按照流式进行计算的

• invoke：把这些 token 先在内部“攒起来”，等模型生成结束后，再一次性把完整结果返回给你；
• stream：是把“生成过程”直接暴露出来，每生成一段（一个 token 或一小段文本）就立刻交给你处理。

写一篇文章：一个字一个字写的

• invoke：等待整篇文章写好之后，再一次性交稿
• stream：一边写，一边将稿子内容对外开放

消息流

langgraph 消息流，就是在“图执行过程中”，把大模型生成过程中的 token 以及对应的节点元数据一起以流式暴露出来。

换句话说：无论模型节点内部使用的是invoke还是stream，langgraph的messages消息流都能够实时的接收到LLM返回的token.

流式每一次拿到的 chunk，大致如下：

[
  LLM token (AIMessageChunk), // 大模型这一次返回的内容
  metadata // 元数据信息
]

• 第一项是消息token
• 第二项则是节点元数据信息

因此 langgraph 这里的消息流，信息会更加丰富一些，不仅有模型 token，还包含“当前执行的是哪个节点”等元数据信息。

const stream = await graph.stream(input, {
  streamMode: "messages",
});

for await (const item of stream) {
  const message = item[0];
  if (typeof message.content === "string") {
    process.stdout.write(message.content);
  } else if (Array.isArray(message.content)) {
    // 说明当前的内容是一个 ContentBlock 内容结构块
    // 内容结构块有很多类型：Text块、Image块...
    // Text块: { "type": "text", "text": "Hello world", "index": 0 }
    // Reasoning块: { "type": "reasoning", "reasoning": "让我思考一下这个问题...", "index": 1 }
    // Image块: { "type": "image", "mimeType
    // 接下来就是针对不同类型的块，做不同的处理
    for (const block of message.content) {
      if (
        block &&
        typeof block === "object" &&
        "text" in block &&
        typeof block.text === "string"
      ) {
        console.log(block.text);
      }
    }
  }
}

消息流返回的每一项如下：

item>>> [
  AIMessageChunk {
    "id": "chatcmpl-CiaoCyawrhuUnI0POuvmDLNivmEgh",
    "content": "理解",
    "additional_kwargs": {},
    "response_metadata": {
      "model_provider": "openai",
      "usage": {}
    },
    "tool_calls": [],
    "tool_call_chunks": [],
    "invalid_tool_calls": []
  },
  {
    tags: [],
    name: undefined,
    langgraph_step: 1,
    langgraph_node: 'writeIntro',
    langgraph_triggers: [ 'branch:to:writeIntro' ],
    langgraph_path: [ '__pregel_pull', 'writeIntro' ],
    langgraph_checkpoint_ns: 'writeIntro:21213964-afa8-53ab-a00b-144e50cd45ed',
    __pregel_task_id: '21213964-afa8-53ab-a00b-144e50cd45ed',
    checkpoint_ns: 'writeIntro:21213964-afa8-53ab-a00b-144e50cd45ed',
    ls_provider: 'openai',
    ls_model_name: 'gpt-4o-mini',
    ls_model_type: 'chat',
    ls_temperature: undefined,
    ls_max_tokens: undefined,
    ls_stop: undefined
  }
]

stream 的每一项是一个数组，数组里面有两个对象：

• item[0]：这是 LangChain 标准内容块的一部分。专门表示：

  • 模型输出的某一段 token（可能为空字符串）
  • 或者模型输出的部分 message 内容
  • 或者工具调用相关 chunk

• item[1]：LangGraph元数据对象，包含当前chunk来自哪个节点、当前步骤、当前模型以及当前路径等信息：

  • LangGraph 当前正在执行哪个节点

    langgraph_node: "writeIntro";
    langgraph_step: 1;
    

  • 是什么触发了该节点执行

    langgraph_triggers: ["branch:to:writeIntro"];
    

  • 当前执行路径

    langgraph_path: ["__pregel_pull", "writeIntro"];
    

  • 当前节点的 checkpoint 空间

    langgraph_checkpoint_ns: "writeIntro:21213964-...";
    

  • LLM 调用的元数据

    ls_provider: "openai";
    ls_model_name: "gpt-4o-mini";
    ls_model_type: "chat";
    

  item[1] 是 LangGraph 给出的“执行事件元数据”。

事件流

graph.stream()：面向“结果”的流

例如：

• messages 模式下，可以拿到 LLM 生成的 message chunk
• updates 模式下，可以拿到 state 的增量变化

这些内容通常是：

• 已经整理好的
• 适合直接给前端消费的
• 不需要理解太多内部执行细节

因此，在大多数应用场景中，优先使用 graph.stream() 就够了。

graph.streamEvents()：面向“过程”的流

相比之下，graph.streamEvents() 更偏底层，它暴露的是图执行过程中产生的各种事件。

这些事件可能包括：

• 某个 node 开始执行
• 某个 node 执行结束
• LLM 产生了一个 token
• 工具被调用
• 子图被调度
• 状态被写入 checkpoint

也就是说，streamEvents() 并不是只关注“输出是什么”，而是关注 “图是如何一步一步跑起来的”。

graph.streamEvents() 方法返回的是事件流，而非内容流，因此不依赖 streamMode。

事件流举例

比如这是我们之前在讲工具调用时，写到的一段代码片段：

// 这里拿到的就是事件流，而非内容流
const events = await graph.streamEvents(
  {
    messages,
  },
  {
    version: "v2",
  },
);

process.stdout.write("助理：");
let finalText = ""; // 拼接最终完整的回复

// 针对每一次事件，做一个事件类型的判断
// 不同的事件类型，做不同的事情
for await (const ev of events) {
  if (ev.event === "on_chat_model_stream") {
    const text = ev.data.chunk.text; // 拿到此次 token 得到的文本值
    finalText += text;
    process.stdout.write(text);
  }

  if (ev.event === "on_tool_start") {
    process.stdout.write(`\n【正在调用工具 ${ev.name}】\n`);
  }
  if (ev.event === "on_tool_end") {
    process.stdout.write(`\n【调用工具 ${ev.name} 完成】\n`);
  }
}

这个地方就用到了事件流，然后根据事件的不同类型，做不同的事情。

常见类型

从架构角度看，streamEvents() 本质上是一个事件驱动接口（event-driven API），非常适合用来构建可观测性、调试工具和复杂 Agent 的执行监控。

常见的事件类型有这么几类：

第一类：模型相关事件

常见的包括：

• on_chat_model_start：大模型开始一次调用
• on_chat_model_stream：大模型正在流式生成 token
• on_chat_model_end：大模型本次调用结束，完整结果已生成
• on_chat_model_error：模型调用过程中发生错误

第二类：工具相关事件

常见的有：

• on_tool_start：工具即将被调用
• on_tool_end：工具调用完成
• on_tool_error：工具执行过程中发生错误

第三类：节点执行相关事件

这一类是很多同学第一次接触时容易忽略，但非常重要的。

例如：

• on_node_start：某个 graph node 开始执行
• on_node_end：某个 graph node 执行完成
• on_node_error：节点执行出错

第四类：链 / 子图 / 调度相关事件

在复杂图里，这类事件会非常有用。

例如：

• on_chain_start
• on_chain_end
• on_chain_error

或者子图相关的调度事件。

示例代码：

// 遍历事件流
for await (const ev of events) {
  // 第一类：模型相关事件
  if (ev.event === "on_chat_model_start") {
    process.stdout.write(`\n【模型开始调用】\n`);
  }
  if (ev.event === "on_chat_model_stream") {
    const text = ev.data.chunk.text; // 拿到此次 token 得到的文本值
    finalText += text;
    process.stdout.write(text);
  }
  if (ev.event === "on_chat_model_end") {
    process.stdout.write(`\n【模型调用结束】\n`);
  }
  if (ev.event === "on_chat_model_error") {
    process.stdout.write(`\n【模型调用出错】\n`);
  }

  // 第二类：工具相关事件
  if (ev.event === "on_tool_start") {
    process.stdout.write(`\n【正在调用工具 ${ev.name}】\n`);
  }
  if (ev.event === "on_tool_end") {
    process.stdout.write(`\n【调用工具 ${ev.name} 完成】\n`);
  }
  if (ev.event === "on_tool_error") {
    process.stdout.write(`\n【工具调用出错 ${ev.name}】\n`);
  }

  // 第三类：节点执行相关事件
  if (ev.event === "on_node_start") {
    process.stdout.write(`\n【节点 ${ev.name} 开始执行】\n`);
  }
  if (ev.event === "on_node_end") {
    process.stdout.write(`\n【节点 ${ev.name} 执行完成】\n`);
  }
  if (ev.event === "on_node_error") {
    process.stdout.write(`\n【节点 ${ev.name} 执行出错】\n`);
  }

  // 第四类：链/子图/调度相关事件
  if (ev.event === "on_chain_start") {
    process.stdout.write(`\n【链 ${ev.name} 开始执行】\n`);
  }
  if (ev.event === "on_chain_end") {
    process.stdout.write(`\n【链 ${ev.name} 执行完成】\n`);
  }
  if (ev.event === "on_chain_error") {
    process.stdout.write(`\n【链 ${ev.name} 执行出错】\n`);
  }
}

自定义流

langgraph 的流式机制中，除了常见的：

• values：完整状态
• updates：状态差异
• messages：LLM token

还有一个最灵活、企业级应用中最常用的模式：custom

这个模式允许开发者在节点内部随时主动 push 任意数据到前端。换句话说：

模型生成 token 不是唯一的数据来源，开发者也可以向前端不断推送事件。

来看一个具体的场景

例如一个企业级审批流，可能需要向前端推送：

• 当前正在执行的中间风控流程
• 风险评分计算步骤
• 进度
• 外部系统调用结果（耗时 / 成功 / 返回内容）

总之就是一些和业务相关的信息，这些信息既不是 LLM token，也不是 state 更新。要推送这种业务信息，就可以用 custom 就可以实现。

自定义流常用于下面的场景：

• 进度条：{ progress: 37% }
• 后端任务进度：{ stage: "fetching_data" }
• 业务事件：{ event: "payment_verified" }
• 中间计算结果：{ partialResult: ... }
• 可视化所需数值，例如风控流水线、规则命中情况

前置知识

首先介绍一下 getWriter() 方法：

import { getWriter } from "@langchain/langgraph";

getWriter() 方法是 langgraph 提供的一个内部上下文函数：

• 在节点执行期间自动绑定到当前执行上下文
• 返回一个函数：writer(data)
• writer(data) 可以把 data 推到外部流（graph.stream）
• 数据类型完全由开发者控制（JSON 对象即可）

具体用法：

首先调用 getWriter() 方法得到一个 writer

const writer = getWriter();

返回的 writer 实质上是一个函数：

type Writer = (data: any) => void;

利用 writer 可以书写 custom 事件数据，例如：

writer({ step: 1, message: "处理中" });
writer({ progress: 50 });
writer({ rule: "R19", hit: true });
writer({ vectorChunk: [0.123, 0.55, ...] });

langgraph 会把这些数据包装成事件的一部分。例如：

writer({ step: 1, message: "处理中" });

外部收到的是：

{
  type: "custom",
  data: { step: 1, message: "处理中" },
  // 其他 langgraph 元信息
  node: "longTask",
  step: 1,
  timestamp: "...",
  checkpoint_ns: "...",
}

// 任务节点
async function longTask() {
  const writer = getWriter();

  if (!writer) return;

  // 书写自定义的数据
  writer({ step: 1, message: "正在准备中..." });

  await sleep(3000);

  writer({ step: 2, message: "正在处理中..." });

  await sleep(3000);

  writer({ step: 3, message: "快要完成了..." });

  await sleep(3000);

  // 更新状态
  return {
    result: "任务完成",
  };
}

// 构建图
const graph = new StateGraph(StateSchema)
  .addNode("longTask", longTask)
  .addEdge(START, "longTask")
  .addEdge("longTask", END)
  .compile();

async function main() {
  console.log("===== custom 流式输出 =====\n");

  let input = {};

  const stream = await graph.stream(input, {
    streamMode: "custom",
  });

  for await (const item of stream) {
    console.log("custom自定义流每一次的输出");
    console.log(item);
    console.log("---------------------");
  }

  console.log("\n===== 执行结束 =====");
}
main();

流式其它细节

1. debug

1. debug模式是什么？

debug 是 langgraph 流式输出中的最详细、最底层的事件模式。当你启用：

streamMode: "debug";

langgraph 会在执行图时，把所有的底层事件（包括执行路径、节点生命周期等）实时推送给你。

它属于开发者调试模式，不是给用户看的。

2. debug模式输出什么内容？

当一个图执行时，每走一步，langgraph 都会产生一类事件。在 debug 模式里可以实时看到：

• node_start：哪个节点开始执行，包含：
  • node 名称
  • 输入状态
  • 当前 superstep
  • 路径信息
• node_end：节点执行结束，包含：
  • 输出状态更新（delta）
  • 执行耗时
  • next steps
• task_scheduled：有哪些节点被调度执行（内部调度信息）
• task_completed：某个任务完成执行
• graph_start / graph_end：整个流程开始 / 结束（可用于全局监听）
• superstep信息：显示每次迭代的 superstep 信息

2. 子图输出

在 langgraph 中，一个节点不仅可以是函数，还能是子图：

ParentGraph
 ├─ node1（普通节点）
 └─ node2（这里是一个完整的subgraph）
      ├─ subgraphNode1
      └─ subgraphNode2

如果对 graph.stream() 进行流式监听：

• 默认情况下，只会看到父图每个节点执行后的状态更新。
• 子图内部节点（subgraphNode1、subgraphNode2）的执行过程不会被流式输出。

但是如果开发者要做诸如：

• 可视化执行路径
• 监控复杂嵌套工作流
• 前端演示“图执行动画”

那就必须知道：每个子图内部的节点什么时候执行了、执行结果是什么。

因此 LangGraph 提供了一个配置：

subgraphs: true;

这样不仅流式输出父图的更新，还流式输出所有子图内部节点的更新。

开启子图流之后，每条流式输出是一个数组：

[namespace, update];

例如：

[[], { node1: { foo: "hi! foo" } }][
  (["node2:<task_id>"], { subgraphNode1: { bar: "bar" } })
][(["node2:<task_id>"], { subgraphNode2: { foo: "hi! foobar" } })][
  ([], { node2: { foo: "hi! foobar" } })
];

这两部分分别表示：

1. namespace
2. update

1. namespace

用于标记当前输出来自哪个图。结构是一个数组：

["nodeName:taskId", "childNode:taskId", ...]

它就是一条“执行路径”。举例：

["node2:dfddc4...", "subgraphNode1:ab3de..."];

含义：

• 当前输出来自父图的节点node2
• node2是一个子图
• 子图里正在执行subgraphNode1

你可以理解成类似 URL 路径：

/node2/subgraphNode1

2. update

该节点执行后的 state 变化。例如：

{'subgraphNode1': {'bar': 'bar'}}

或：

{'node1': {'foo': 'hi! foo'}}

流式输出可以逐步看到：

• 哪个节点执行了？
• 它对 state 做了什么更新？
• 执行顺序是什么？

示例代码：

// 演示子图的输出
import { StateGraph, START } from "@langchain/langgraph";
import { z } from "zod/v4";

// 定义子图状态
const SubState = z.object({
  foo: z.string(),
  bar: z.string(),
});

// 构建子图
const subgraph = new StateGraph(SubState)
  .addNode("subgraphNode1", () => ({ bar: "bar" })) // 更新bar的值
  .addNode("subgraphNode2", (state) => ({ foo: state.foo + state.bar })) // 更新foo的值
  .addEdge(START, "subgraphNode1")
  .addEdge("subgraphNode1", "subgraphNode2")
  .compile();

// 定义父图的状态
const ParentState = z.object({
  foo: z.string(),
  bar: z.string(),
});

const graph = new StateGraph(ParentState)
  .addNode("node1", (state) => ({ foo: "hi! " + state.foo })) // 对foo字段进行更新
  .addNode("node2", subgraph) // 子图节点
  .addEdge(START, "node1")
  .addEdge("node1", "node2")
  .compile();

async function main() {
  const stream = await graph.stream(
    { foo: "这是一个测试" },
    {
      streamMode: "updates",
      subgraph: true, // show the sub graph updates information
    },
  );

  for await (const item of stream) {
    console.log("updates流输出的每一项");
    console.log(item);
    console.log("---------------------");
  }
}
main();