Cannot use streamMode when streaming from a subgraph #236

New Issue

2026-02-15T17:17:43-05:00

yindo commented

2026-02-15 17:17:43 -05:00

Originally created by @au-re on GitHub (Apr 18, 2025).

I think there is a bug when using streamMode in nested subgraphs. I couldn't find anything about this in the documentation.

Given the following setup with a subgraph / main graph, I would like to create custom events based on state updates in the subgraph. However, only stream events of type "values" are triggered from the subgraph, regardless of the streamMode I specify.

@langchain/langgraph: "0.2.64"

StreamMode not working as expected

import { entrypoint, MemorySaver, task } from "@langchain/langgraph";

let i = 0;

const doSomething = task("do_something", async (text: string) => {
  await new Promise((resolve) => setTimeout(resolve, 1000));
  return `(do ${text}): ${i++}`;
});

const internal = entrypoint({ name: "internal" }, async (params: { query: string }, config) => {
  let internalState: string[] = [];

  internalState.push(await doSomething(params.query));

  internalState.push(await doSomething(params.query));

  internalState.push(await doSomething(params.query));

  return internalState;
});

const main = entrypoint({ checkpointer: new MemorySaver(), name: "main" }, async (params, config) => {
  let mainState: string[] = [];

  const res1 = await doSomething("task 1");

  mainState.push(res1);

  const stream = await internal.stream({ query: "internal task" }, { streamMode: ["updates"] });

  for await (const chunk of stream) {
    console.log("type:", chunk[0]); // streamMode is ignored: yields "values" instead of "updates"

    if (chunk[1]?.do_something) mainState.push(chunk[1].do_something);

    config.writer?.(mainState);
  }

  return mainState;
});

Running the entrypoint

const stream = await main.stream(
  {},
  {
    configurable: { thread_id: "1" },
    streamMode: ["custom"],
  },
);

for await (const chunk of stream) {
  console.log(JSON.stringify(chunk, null, 2));
}

Logs

Only a single event is logged (when the subgraph is done running and we receive the "values" event.

[
  "custom",
  [
    "(do task 1): 0"
  ]
]

I can get the stream to work by passing a new configurables object to the subgraph, but this breaks the checkpointer setup and I am no longer able to resume from the subgraph:

Checkpointer not working as expected

const main = entrypoint({ checkpointer: new MemorySaver(), name: "main" }, async (params, config) => {
  let mainState: string[] = [];

  const res1 = await doSomething("task 1");

  mainState.push(res1);

  const stream = await internal.stream(
    { query: "internal task" },
    {
      ...config,
      configurable: { thread_id: "1" }, // overwriting the configurable object
      streamMode: ["updates"],
    },
  );

  for await (const chunk of stream) {
    console.log("type:", chunk[0]);

    if (chunk[1].do_something) mainState.push(chunk[1].do_something);

    config.writer?.(mainState);
  }

  return mainState;
});

Now we write the events as expected (on update events from the subgraph)

[
  "custom",
  [
    "(do task 1): 0",
    "(do internal task): 1"
  ]
]

[
  "custom",
  [
    "(do task 1): 0",
    "(do internal task): 1",
    "(do internal task): 2"
  ]
]

[
  "custom",
  [
    "(do task 1): 0",
    "(do internal task): 1",
    "(do internal task): 2",
    "(do internal task): 3"
  ]
]

[
  "custom",
  [
    "(do task 1): 0",
    "(do internal task): 1",
    "(do internal task): 2",
    "(do internal task): 3"
  ]
]

But if I overwride the configurables object, this breaks the checkpointer (even if the thread_id is the same). I can no longer resume from the subgraph when adding the following:

const internal = entrypoint({ name: "internal" }, async (params: { query: string }, config) => {
  let internalState: string[] = [];

  internalState.push(await doSomething(params.query));

  internalState.push(await doSomething(params.query));

  internalState.push(await doSomething(params.query));

  const followup: string = interrupt({}); // added an interrupt here

  internalState.push(await doSomething(followup)); 

  return internalState;
});

const resumed = await main.stream(
  new Command({
    resume: "interrupt",
  }),
  {
    configurable: { thread_id: "1" },
    streamMode: ["custom"],
  },
);

for await (const chunk of resumed) {
  console.log(JSON.stringify(chunk, null, 2));
}

Removing the configurables object makes the checkpointer work again but then I am back to not being able to listen to "updates" events.

Originally created by @au-re on GitHub (Apr 18, 2025). I think there is a bug when using streamMode in nested subgraphs. I couldn't find anything about this in the documentation. Given the following setup with a subgraph / main graph, I would like to create custom events based on state updates in the subgraph. However, only stream events of type "values" are triggered from the subgraph, regardless of the streamMode I specify. @langchain/langgraph: "0.2.64" ## StreamMode not working as expected ```js import { entrypoint, MemorySaver, task } from "@langchain/langgraph"; let i = 0; const doSomething = task("do_something", async (text: string) => { await new Promise((resolve) => setTimeout(resolve, 1000)); return `(do ${text}): ${i++}`; }); const internal = entrypoint({ name: "internal" }, async (params: { query: string }, config) => { let internalState: string[] = []; internalState.push(await doSomething(params.query)); internalState.push(await doSomething(params.query)); internalState.push(await doSomething(params.query)); return internalState; }); const main = entrypoint({ checkpointer: new MemorySaver(), name: "main" }, async (params, config) => { let mainState: string[] = []; const res1 = await doSomething("task 1"); mainState.push(res1); const stream = await internal.stream({ query: "internal task" }, { streamMode: ["updates"] }); for await (const chunk of stream) { console.log("type:", chunk[0]); // streamMode is ignored: yields "values" instead of "updates" if (chunk[1]?.do_something) mainState.push(chunk[1].do_something); config.writer?.(mainState); } return mainState; }); ``` ### Running the entrypoint ```js const stream = await main.stream( {}, { configurable: { thread_id: "1" }, streamMode: ["custom"], }, ); for await (const chunk of stream) { console.log(JSON.stringify(chunk, null, 2)); } ``` ### Logs Only a single event is logged (when the subgraph is done running and we receive the "values" event. ``` [ "custom", [ "(do task 1): 0" ] ] ``` **I can get the stream to work by passing a new configurables object to the subgraph**, but this breaks the checkpointer setup and I am no longer able to resume from the subgraph: ## Checkpointer not working as expected ```js const main = entrypoint({ checkpointer: new MemorySaver(), name: "main" }, async (params, config) => { let mainState: string[] = []; const res1 = await doSomething("task 1"); mainState.push(res1); const stream = await internal.stream( { query: "internal task" }, { ...config, configurable: { thread_id: "1" }, // overwriting the configurable object streamMode: ["updates"], }, ); for await (const chunk of stream) { console.log("type:", chunk[0]); if (chunk[1].do_something) mainState.push(chunk[1].do_something); config.writer?.(mainState); } return mainState; }); ``` Now we write the events as expected (on update events from the subgraph) ``` [ "custom", [ "(do task 1): 0", "(do internal task): 1" ] ] [ "custom", [ "(do task 1): 0", "(do internal task): 1", "(do internal task): 2" ] ] [ "custom", [ "(do task 1): 0", "(do internal task): 1", "(do internal task): 2", "(do internal task): 3" ] ] [ "custom", [ "(do task 1): 0", "(do internal task): 1", "(do internal task): 2", "(do internal task): 3" ] ] ``` But if I overwride the configurables object, this breaks the checkpointer (even if the thread_id is the same). I can no longer resume from the subgraph when adding the following: ```js const internal = entrypoint({ name: "internal" }, async (params: { query: string }, config) => { let internalState: string[] = []; internalState.push(await doSomething(params.query)); internalState.push(await doSomething(params.query)); internalState.push(await doSomething(params.query)); const followup: string = interrupt({}); // added an interrupt here internalState.push(await doSomething(followup)); return internalState; }); ``` ```js const resumed = await main.stream( new Command({ resume: "interrupt", }), { configurable: { thread_id: "1" }, streamMode: ["custom"], }, ); for await (const chunk of resumed) { console.log(JSON.stringify(chunk, null, 2)); } ``` Removing the configurables object makes the checkpointer work again but then I am back to not being able to listen to "updates" events.

yindo added the docs-needed label 2026-02-15 17:17:43 -05:00

yindo closed this issue

2026-02-15 17:17:43 -05:00

yindo commented

2026-02-15 18:14:54 -05:00

@benjamincburns commented on GitHub (Apr 28, 2025):

Hmm, it does indeed seem that we have a gap in our docs around this topic. I would've expected to see details about streaming from subgraphs in either the Streaming or Subgraphs sections of the How-to docs, but neither addresses it.

To answer your question - you'll need to set the subgraphs option to true. It can be found on the config object that you pass as the second argument to the stream method.

Please ping me if this doesn't get you on your way and I'll reopen this issue.

@benjamincburns commented on GitHub (Apr 28, 2025): Hmm, it does indeed seem that we have a gap in our docs around this topic. I would've expected to see details about streaming from subgraphs in either the [Streaming](https://langchain-ai.github.io/langgraphjs/how-tos/#streaming) or [Subgraphs](https://langchain-ai.github.io/langgraphjs/how-tos/#subgraphs) sections of the How-to docs, but neither addresses it. To answer your question - you'll need to set the `subgraphs` option to true. It can be found on [the config object](https://langchain-ai.github.io/langgraphjs/reference/interfaces/langgraph.PregelOptions.html) that you pass as the second argument to the `stream` method. Please ping me if this doesn't get you on your way and I'll reopen this issue.

yindo commented

2026-02-15 18:14:54 -05:00

@au-re commented on GitHub (Apr 28, 2025):

@benjamincburns the issue with "subgraphs": true is that if I add it to internal.stream, it won't do anything because that has no other subgraphs which means the code above still won't work as expected. I can add it to main.stream but then how can I update the main graph state on stream events from subgraphs?

What I would like to do is update the state of the main graph each time subgraphs are emitting update / message events.

@au-re commented on GitHub (Apr 28, 2025): @benjamincburns the issue with `"subgraphs": true` is that if I add it to `internal.stream`, it won't do anything because that has no other subgraphs which means the code above still won't work as expected. I can add it to `main.stream` but then how can I update the main graph state on stream events from subgraphs? What I would like to do is update the state of the main graph each time subgraphs are emitting update / message events.

yindo commented

2026-02-15 18:14:55 -05:00

@au-re commented on GitHub (Apr 28, 2025):

@benjamincburns I guess for now the only way to do this is to manage that state at the top level. Which is not ideal when you have nested graphs with different state structures, since you'll have to transform their results in two places: after invoke + at the top level. It's also really tricky to manually reconcile chunks into the main state for every event across all subgraphs at once...

Below is the work around I am currently using, it's ok for this simple example but becomes tricky for more complex graphs:

import { Command, entrypoint, interrupt, MemorySaver, task } from "@langchain/langgraph";

let i = 0;

const doSomething = task("do_something", async (text: string) => {
  await new Promise((resolve) => setTimeout(resolve, 1000));
  return `(do ${text}): ${i++}`;
});

const internal = entrypoint({ name: "internal" }, async (params: { query: string }, config) => {
  let internalState: string[] = [];
  internalState.push(await doSomething(params.query));
  internalState.push(await doSomething(params.query));
  internalState.push(await doSomething(params.query));
  const followup: string = interrupt({});
  internalState.push(await doSomething(followup));
  return internalState;
});

const main = entrypoint({ checkpointer: new MemorySaver(), name: "main" }, async (params, config) => {
  let mainState: string[] = [];
  mainState.push(await doSomething("task 1"));
  const res = await internal.invoke({ query: "internal task" });
  mainState = [...mainState, ...res];
  mainState.push(await doSomething("last task"));
  return mainState;
});

const stream = await main.stream(
  {},
  {
    configurable: { thread_id: "1" },
    streamMode: ["updates"],
    subgraphs: true,
  },
);

const mainState: string[] = [];

for await (const chunk of stream) {
  if (chunk[1] === "updates") {
    if (chunk[2]?.do_something) {
      mainState.push(chunk[2].do_something);
      console.log("STATE:", JSON.stringify(mainState, null, 2));
    }
  }
}

const resumed = await main.stream(
  new Command({
    resume: "interrupt",
  }),
  {
    configurable: { thread_id: "1" },
    streamMode: ["updates"],
    subgraphs: true,
  },
);

const resumedState: string[] = [];

for await (const chunk of resumed) {
  if (chunk[1] === "updates") {
    if (chunk[2]?.do_something) {
      resumedState.push(chunk[2].do_something);
      console.log("STATE:", JSON.stringify(resumedState, null, 2));
    }
  }
}

@au-re commented on GitHub (Apr 28, 2025): @benjamincburns I guess for now the only way to do this is to manage that state at the top level. Which is not ideal when you have nested graphs with different state structures, since you'll have to transform their results in two places: after invoke + at the top level. It's also really tricky to manually reconcile chunks into the main state for every event across all subgraphs at once... Below is the work around I am currently using, it's ok for this simple example but becomes tricky for more complex graphs: ```ts import { Command, entrypoint, interrupt, MemorySaver, task } from "@langchain/langgraph"; let i = 0; const doSomething = task("do_something", async (text: string) => { await new Promise((resolve) => setTimeout(resolve, 1000)); return `(do ${text}): ${i++}`; }); const internal = entrypoint({ name: "internal" }, async (params: { query: string }, config) => { let internalState: string[] = []; internalState.push(await doSomething(params.query)); internalState.push(await doSomething(params.query)); internalState.push(await doSomething(params.query)); const followup: string = interrupt({}); internalState.push(await doSomething(followup)); return internalState; }); const main = entrypoint({ checkpointer: new MemorySaver(), name: "main" }, async (params, config) => { let mainState: string[] = []; mainState.push(await doSomething("task 1")); const res = await internal.invoke({ query: "internal task" }); mainState = [...mainState, ...res]; mainState.push(await doSomething("last task")); return mainState; }); const stream = await main.stream( {}, { configurable: { thread_id: "1" }, streamMode: ["updates"], subgraphs: true, }, ); const mainState: string[] = []; for await (const chunk of stream) { if (chunk[1] === "updates") { if (chunk[2]?.do_something) { mainState.push(chunk[2].do_something); console.log("STATE:", JSON.stringify(mainState, null, 2)); } } } const resumed = await main.stream( new Command({ resume: "interrupt", }), { configurable: { thread_id: "1" }, streamMode: ["updates"], subgraphs: true, }, ); const resumedState: string[] = []; for await (const chunk of resumed) { if (chunk[1] === "updates") { if (chunk[2]?.do_something) { resumedState.push(chunk[2].do_something); console.log("STATE:", JSON.stringify(resumedState, null, 2)); } } } ```

yindo commented

2026-02-15 18:14:55 -05:00

@benjamincburns commented on GitHub (May 22, 2025):

Sorry to be so slow to reopen this @au-re - @dqbd might be worth considering how to address this.

@benjamincburns commented on GitHub (May 22, 2025): Sorry to be so slow to reopen this @au-re - @dqbd might be worth considering how to address this.

yindo commented

2026-02-15 18:14:55 -05:00

@dqbd commented on GitHub (Jun 26, 2025):

This will be fixed with https://github.com/langchain-ai/langgraphjs/issues/1116!

@dqbd commented on GitHub (Jun 26, 2025): This will be fixed with https://github.com/langchain-ai/langgraphjs/issues/1116!

yindo referenced this issue

2026-02-15 19:15:09 -05:00

[PR #236] [CLOSED] feat: vercelKV checkpointer (WIP) #595

Sign in to join this conversation.

Branches Tags

main

dependabot/npm_and_yarn/angular-10c8d2db5f

dependabot/npm_and_yarn/svelte-7618a5540f

dependabot/npm_and_yarn/hono/node-server-2.0.11

open-swe/usestream-status-enum

open-swe/accept-semver-ranges

wfh/fix_store

cb/reconnect-upstream

nh/integration-managed-runtime-tests

hunter/bump-checkpoint-dep

cb/bundle-uuid-cjs-66cc

cb/dedup-stream-sse-decoder-96cd

dependabot/npm_and_yarn/internal/bench/npm_and_yarn-18a2f9e291

cb/normalize-openai-function-call-tool-calls

cb/tmp

cb/remove-docs

nh/sdk-normalize-base-message-v2

nh/sdk-reencode-tool-call-chunks

nh/sdk-normalize-base-message

nh/stream-v2-final-split

open-swe/c802e0ea-c9e5-c8c6-824a-c065dd2bfd6f

hunter/memory-investigation-2

ramonn/checkpoint-durable-objects

open-swe/776ee51f-5e2f-752b-acb1-4584467156d4

hunter/issue-2303-regression-test

cb/api-stream-reducer

open-swe/77892c09-3e94-9f8a-5f5b-3d2787c5f6b9

cb/in-process-streaming

cb/langgraph-streaming-protocol-683b

cb/protocol-v2-tests-6203

cb/fix-custom-submit-options-ad3e

cb/protocol-thread-binding-clean-6303

hunter/keep-important-ids

cb/fix-protocol-thread-binding-6303

cb/react-protocol-runtime-8ca7

cb/pregel-eli5-doc-1103

cb/-bc-5a4705b9-850e-4f7d-95ea-67c0741e9977-92c3

cb/pr-branch-rebase-4816

cb/-bc-2d03efa3-5ef0-4b68-bf53-63005b8e748f-8924-angular-svelte

cb/-bc-2d03efa3-5ef0-4b68-bf53-63005b8e748f-8924

cb/langchain-react-ssr-proposal-972b

cb/agent-streaming-protocol-design-25d1

cb/custom-transport-queue-parity-2bb9

cb/frontend-integration-strategy-64ce

brydar/fix/parallel-interrupt-accumulation

cb/agent-multi-app-support-d2d0

cb/svelte-5-idiomatic-patterns-16ed

cb/frontend-sdks-developer-experience-8030

cb/ui-packages-idiomatic-review-35a4

hunter/fix-vercel-ci

fix/test-exports-vercel-next16-lint

dependabot/npm_and_yarn/internal/environment_tests/test-exports-vercel/npm_and_yarn-229c191f58

cb/stream-protocol-improvements

jdr/fix-sdk-errors

cb/async_hooks-fix

cwlbraa/fix-getmessages-null-state

chore/fix-dependabot-config

hunter/fix-symbol-types

hunter/symbol-fix

hunter/symbol-annotations

test-ci-trigger

hunter/lg-resolve-symbols

colifran/tool-stream-types

colifran/tool-stream-types-temp

dqbd/sdk-ui-other-fw

colifran/tool-streaming-example

colifran/tool-stream-example

fix/security-alerts-2026-02-19

cb/perf-improvements

jk/fix-rolldown-panic

cb/better-node-version-definition

cb/tsdown-setup

cb/create-changesets

cb/pnpm-fix4

cb/pnpm-fix2

wfh/fix_release

demo

cb/browser-tools

v0.3

pr/christymanthara/1787

cb/type-safe-streaming

mdrxy/delete-docs

changeset-release/v1

v1

v0

rd3

isaac/vb

dqbd/remove-get-graph-non-async

dqbd/serde-v1

dqbd/1.0.0-alpha.1

dqbd/max-version-cache

parker/custom-studio-url-cli

dqbd/standard-schema

dqbd/create-agent-middleware

dqbd/1.0.0-alpha

parker/api-linting

dqbd/prebuilt-restart-subgraph

dqbd/api-thread-join-stream

dqbd/persist-resume-map-values

wfh/default

dqbd/node-recursion-limit

sam_support_description_for_mcp

hunter/dedupe-signal-listeners

wfh/studio_auth_access

wfh/consistent_kind

release

brace/pass-update-generic-msgs-zod

dqbd/merge-api

hunter/zod-intersection

vb/add-forward-message-tool

dqbd/zod-metadata-json-schema-extra

wfh/js_node_finished

ben/checkpoint-metadata-custom-fields

dqbd/release-0.2.56-second

ben/docs-learning-path

ben/typed-exec

vb/support-checkpointer-true

dqbd/stream-mode-messages-input-no-emit

dqbd/tsc-module-es2020

jacob/nits

cc/docs_layout

dqbd/tasks-results

brace/update-how-to-hitl-index

jacob/sendv2

dqbd/command-goto

jacob/stream_messages

jacob/supabase

brace/updated-mem-conceptual

471

allan/first-agent-tutorial

brace/req-config

nc/23sep/stream-output

allan/subgraphs

brace/stores

brace/memory

jacob/rc

isaac/contextchannel

brace/subgraphs

wfh/typedocs

0.1

isaac/doclink

jacob/mongo

testing

nc/11aug/archs

jacob/streaming_tool_call

jacob/messagesstate

wfh/pae

wfh/moore_tutorials_tsc

wfh/drd

wfh/edgy

brace/langgraph-self-rag-oss

brace/prebuilt

brace/state-graph

brace/bump-lc-core

brace/typing

brace/examples

nc/fixup

brace/package-infra

brace/allow-channel-string

1 Participants

Notifications

Due Date

No due date set.

Dependencies

No dependencies set.

Reference: langchain-ai/langgraphjs#236

Cannot use streamMode when streaming from a subgraph #236

StreamMode not working as expected

Running the entrypoint

Logs

Checkpointer not working as expected

Cannot use `streamMode` when streaming from a subgraph #236