langchain-ai/langgraphjs
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraphjs.git synced 2026-07-22 00:55:26 -04:00
Code Issues 108 Packages Projects Releases 20 Wiki Activity

Inconsistent behaviour for fan-in - The aggregate node can be triggered multiple times #337

New Issue
Closed
opened 2026-02-15 18:16:00 -05:00 by yindo · 2 comments
yindo commented 2026-02-15 18:16:00 -05:00
Owner
Copy Link
Copy Source

Originally created by @wenshen35105 on GitHub (Aug 14, 2025).

Checked other resources

  • I added a very descriptive title to this issue.
  • I searched the LangGraph.js documentation with the integrated search.
  • I used the GitHub search to find a similar question and didn't find it.
  • I am sure that this is a bug in LangGraph.js rather than my code.
  • The bug is not resolved by updating to the latest stable version of LangGraph (or the specific integration package).

Example Code

/**
 * file: test1.ts
 * 
 * __start__ -> test1 -> test2 -> test6 (is triggered twice) -> __end__ 
 *    |                             ^
 *    |                             | 
 *    | -> test3 -> test4 -> test5 /
 * 
 * Terminal:
 *  > npx ts-node test1.ts
 *    test1 {}
 *    test3 {}
 *    test2 {}
 *    test4 {}
 *    test5 {}
 *    test6 {}
 *    test6 {}
 *  
 */
import fs from "fs";
import path from "path";
import {
  StateGraph,
  Annotation,
  MemorySaver
} from "@langchain/langgraph";

(async () => {
  const OverallState = Annotation.Root({});

  const graph = new StateGraph(OverallState);

  graph
    .addNode("test1", async (state) => {
      console.log("test1", state);
    })
    .addNode("test2", async (state) => {
      console.log("test2", state);
    })
    .addNode("test3", async (state) => {
      console.log("test3", state);
    })
    .addNode("test4", async (state) => {
      console.log("test4", state);
    })
    .addNode("test5", async (state) => {
      console.log("test5", state);
    })
    .addNode("test6", async (state) => {
      console.log("test6", state);
    })
    .addConditionalEdges("__start__", (state) => {
      return ["test1", "test3"]
    }, ["test1", "test3"])

    // test1 -> test2 -> test6
    .addConditionalEdges("test1", (state) => {
      return "test2"
    }, ["test2"])
    .addConditionalEdges("test2", (state) => {
      return "test6"
    }, ["test6"])

    // test3 -> test4 -> test5 -> test6
    .addConditionalEdges("test3", (state) => {
      return "test4"
    }, ["test4"])
    .addConditionalEdges("test4", (state) => {
      return "test5"
    }, ["test5"])
    .addConditionalEdges("test5", (state) => {
      return "test6"
    }, ["test6"])

    .addConditionalEdges("test6", (state) => "__end__", ["__end__"])

  const checkpoint = new MemorySaver();

  const compiled = graph.compile({
    checkpointer: checkpoint
  });

  const png = await (await compiled.getGraphAsync()).drawMermaidPng();
  fs.writeFileSync(path.resolve("test1.png"), Buffer.from(await png.arrayBuffer()));

  try {
    for await (const s of await compiled.stream({}, { configurable: { thread_id: "1"}})) {}
  } catch (err) {
    console.error(err)
  }
})();


/**
 * file: test2.ts
 * 
 * __start__ -> test1 -> test2 -> test6 (is triggered once) -> __end__ 
 *    |                             ^
 *    |                             | 
 *    \-----------> test3 -> test4 /
 * 
 * Terminal:
 *  > npx ts-node test2.ts
 *    test1 {}
 *    test3 {}
 *    test2 {}
 *    test4 {}
 *    test6 {}
 *  
 */
import fs from "fs";
import path from "path";
import {
  StateGraph,
  Annotation,
  MemorySaver
} from "@langchain/langgraph";

(async () => {
  const OverallState = Annotation.Root({});

  const graph = new StateGraph(OverallState);

  graph
    .addNode("test1", async (state) => {
      console.log("test1", state);
    })
    .addNode("test2", async (state) => {
      console.log("test2", state);
    })
    .addNode("test3", async (state) => {
      console.log("test3", state);
    })
    .addNode("test4", async (state) => {
      console.log("test4", state);
    })
    .addNode("test6", async (state) => {
      console.log("test6", state);
    })
    .addConditionalEdges("__start__", (state) => {
      return ["test1", "test3"]
    }, ["test1", "test3"])

    // test1 -> test2 -> test6
    .addConditionalEdges("test1", (state) => {
      return "test2"
    }, ["test2"])
    .addConditionalEdges("test2", (state) => {
      return "test6"
    }, ["test6"])

    // test3 -> test4 -> test6
    .addConditionalEdges("test3", (state) => {
      return "test4"
    }, ["test4"])
    .addConditionalEdges("test4", (state) => {
      return "test6"
    }, ["test6"])

    .addConditionalEdges("test6", (state) => "__end__", ["__end__"])

  const checkpoint = new MemorySaver();

  const compiled = graph.compile({
    checkpointer: checkpoint
  });

  const png = await (await compiled.getGraphAsync()).drawMermaidPng();
  fs.writeFileSync(path.resolve("test2.png"), Buffer.from(await png.arrayBuffer()));

  try {
    for await (const s of await compiled.stream({}, { configurable: { thread_id: "1"}})) {}
  } catch (err) {
    console.error(err)
  }
})();

Error Message and Stack Trace (if applicable)

No response

Description

I observed that the aggregate node can be triggered multiple times if the fan-out paths have different amounts of nodes. According to https://langchain-ai.github.io/langgraphjs/how-tos/branching/#fan-out-fan-in, it seems to me that the aggregate node should only be trigged once.

/**
 * __start__ -> test1 -> test2 -> test6 (is triggered twice) -> __end__ 
 *    |                             ^
 *    |                             | 
 *    | -> test3 -> test4 -> test5 /
 *  
 */


/**
 * __start__ -> test1 -> test2 -> test6 (is triggered once) -> __end__ 
 *    |                             ^
 *    |                             | 
 *    \-----------> test3 -> test4 /
 */

System Info

Node version: v22.14.0
Operating system: darwin arm64
Package manager: npm
Package manager version: undefined

@langchain/textsplitters -> @0.1.0, , @"^0.1.0"
@langchain/core -> @0.3.70, , @"^0.3.42", @">=0.3.58, @">=0.2.31, @">=0.2.21
langsmith -> @0.3.59, , @"^0.3.46"
zod -> @3.25.76, , @"^3.23.8", @"^3.25.32", @"^3.24.1"
@langchain/langgraph -> @0.4.4, , @"^0.4.4"
@langchain/langgraph-checkpoint -> @0.1.0, , @"^0.1.0"
@langchain/langgraph-sdk -> @0.0.107, , @"~0.0.107"

Originally created by @wenshen35105 on GitHub (Aug 14, 2025). ### Checked other resources - [x] I added a very descriptive title to this issue. - [x] I searched the LangGraph.js documentation with the integrated search. - [x] I used the GitHub search to find a similar question and didn't find it. - [x] I am sure that this is a bug in LangGraph.js rather than my code. - [x] The bug is not resolved by updating to the latest stable version of LangGraph (or the specific integration package). ### Example Code ```typescript /** * file: test1.ts * * __start__ -> test1 -> test2 -> test6 (is triggered twice) -> __end__ * | ^ * | | * | -> test3 -> test4 -> test5 / * * Terminal: * > npx ts-node test1.ts * test1 {} * test3 {} * test2 {} * test4 {} * test5 {} * test6 {} * test6 {} * */ import fs from "fs"; import path from "path"; import { StateGraph, Annotation, MemorySaver } from "@langchain/langgraph"; (async () => { const OverallState = Annotation.Root({}); const graph = new StateGraph(OverallState); graph .addNode("test1", async (state) => { console.log("test1", state); }) .addNode("test2", async (state) => { console.log("test2", state); }) .addNode("test3", async (state) => { console.log("test3", state); }) .addNode("test4", async (state) => { console.log("test4", state); }) .addNode("test5", async (state) => { console.log("test5", state); }) .addNode("test6", async (state) => { console.log("test6", state); }) .addConditionalEdges("__start__", (state) => { return ["test1", "test3"] }, ["test1", "test3"]) // test1 -> test2 -> test6 .addConditionalEdges("test1", (state) => { return "test2" }, ["test2"]) .addConditionalEdges("test2", (state) => { return "test6" }, ["test6"]) // test3 -> test4 -> test5 -> test6 .addConditionalEdges("test3", (state) => { return "test4" }, ["test4"]) .addConditionalEdges("test4", (state) => { return "test5" }, ["test5"]) .addConditionalEdges("test5", (state) => { return "test6" }, ["test6"]) .addConditionalEdges("test6", (state) => "__end__", ["__end__"]) const checkpoint = new MemorySaver(); const compiled = graph.compile({ checkpointer: checkpoint }); const png = await (await compiled.getGraphAsync()).drawMermaidPng(); fs.writeFileSync(path.resolve("test1.png"), Buffer.from(await png.arrayBuffer())); try { for await (const s of await compiled.stream({}, { configurable: { thread_id: "1"}})) {} } catch (err) { console.error(err) } })(); ``` ```typescript /** * file: test2.ts * * __start__ -> test1 -> test2 -> test6 (is triggered once) -> __end__ * | ^ * | | * \-----------> test3 -> test4 / * * Terminal: * > npx ts-node test2.ts * test1 {} * test3 {} * test2 {} * test4 {} * test6 {} * */ import fs from "fs"; import path from "path"; import { StateGraph, Annotation, MemorySaver } from "@langchain/langgraph"; (async () => { const OverallState = Annotation.Root({}); const graph = new StateGraph(OverallState); graph .addNode("test1", async (state) => { console.log("test1", state); }) .addNode("test2", async (state) => { console.log("test2", state); }) .addNode("test3", async (state) => { console.log("test3", state); }) .addNode("test4", async (state) => { console.log("test4", state); }) .addNode("test6", async (state) => { console.log("test6", state); }) .addConditionalEdges("__start__", (state) => { return ["test1", "test3"] }, ["test1", "test3"]) // test1 -> test2 -> test6 .addConditionalEdges("test1", (state) => { return "test2" }, ["test2"]) .addConditionalEdges("test2", (state) => { return "test6" }, ["test6"]) // test3 -> test4 -> test6 .addConditionalEdges("test3", (state) => { return "test4" }, ["test4"]) .addConditionalEdges("test4", (state) => { return "test6" }, ["test6"]) .addConditionalEdges("test6", (state) => "__end__", ["__end__"]) const checkpoint = new MemorySaver(); const compiled = graph.compile({ checkpointer: checkpoint }); const png = await (await compiled.getGraphAsync()).drawMermaidPng(); fs.writeFileSync(path.resolve("test2.png"), Buffer.from(await png.arrayBuffer())); try { for await (const s of await compiled.stream({}, { configurable: { thread_id: "1"}})) {} } catch (err) { console.error(err) } })(); ``` ### Error Message and Stack Trace (if applicable) _No response_ ### Description I observed that the aggregate node can be triggered multiple times if the fan-out paths have different amounts of nodes. According to https://langchain-ai.github.io/langgraphjs/how-tos/branching/#fan-out-fan-in, it seems to me that the aggregate node should only be trigged once. ```typescript /** * __start__ -> test1 -> test2 -> test6 (is triggered twice) -> __end__ * | ^ * | | * | -> test3 -> test4 -> test5 / * */ ``` ```typescript /** * __start__ -> test1 -> test2 -> test6 (is triggered once) -> __end__ * | ^ * | | * \-----------> test3 -> test4 / */ ``` ### System Info Node version: v22.14.0 Operating system: darwin arm64 Package manager: npm Package manager version: undefined -------------------- @langchain/textsplitters -> @0.1.0, , @"^0.1.0" @langchain/core -> @0.3.70, , @"^0.3.42", @">=0.3.58, @">=0.2.31, @">=0.2.21 langsmith -> @0.3.59, , @"^0.3.46" zod -> @3.25.76, , @"^3.23.8", @"^3.25.32", @"^3.24.1" @langchain/langgraph -> @0.4.4, , @"^0.4.4" @langchain/langgraph-checkpoint -> @0.1.0, , @"^0.1.0" @langchain/langgraph-sdk -> @0.0.107, , @"~0.0.107"
yindo closed this issue 2026-02-15 18:16:00 -05:00
yindo commented 2026-02-15 18:16:01 -05:00
Author
Owner
Copy Link
Copy Source

@onestardao commented on GitHub (Aug 15, 2025):

this looks like no.13 – multi-agent chaos from our 16-problem map: a coordination race at the fan-in (aggregate) join. multiple upstream branches are emitting more than once (retries / state updates), and the join node treats each emission as a fresh arrival, so it triggers twice.

quick, practical fixes

  • idempotent join: keep a per-run arrival set keyed by {run_id, branch_id, epoch} and only fire when you’ve seen each expected branch exactly once. ignore duplicates.
  • expected count guard: compute the inbound degree at build time and store it on the node; only evaluate when arrivals.size === expected.
  • sequence/epoch numbers: have upstream nodes attach a monotonically increasing epoch; the join discards older/duplicate epochs.
  • dedupe middleware: wrap upstream emits with a small cache (e.g., LRU keyed by hash of payload) to prevent identical replays.
  • exactly-once checkpoint: if langgraph offers a checkpointer, mark the aggregate’s completion with a CAS flag so replays don’t re-enter.

tiny sketch:

type Key = `${string}:${string}` // `${runId}:${branchId}`

class AggregateOnce {
  private arrivals = new Map<string, Set<string>>(); // runId -> set(branchId)
  constructor(private expected: number) {}

  onInput(runId: string, branchId: string, payload: unknown) {
    const set = this.arrivals.get(runId) ?? new Set<string>();
    if (set.has(branchId)) return;            // dedupe
    set.add(branchId);
    this.arrivals.set(runId, set);
    if (set.size === this.expected) {
      this.arrivals.delete(runId);            // reset for next run
      return this.evaluate();                 // safe to trigger once
    }
  }
}

mit-licensed work; we’ve cataloged 16 such patterns and helped many teams stabilize joins like this (cold start → ~600 stars in ~60 days).
if you want the full “join-once + replay guard” snippet we use, say the word and i’ll share.

@onestardao commented on GitHub (Aug 15, 2025): this looks like **no.13 – multi-agent chaos** from our 16-problem map: a coordination race at the fan-in (aggregate) join. multiple upstream branches are emitting more than once (retries / state updates), and the join node treats each emission as a fresh arrival, so it triggers twice. **quick, practical fixes** * **idempotent join:** keep a per-run arrival set keyed by `{run_id, branch_id, epoch}` and only fire when you’ve seen *each expected branch exactly once*. ignore duplicates. * **expected count guard:** compute the inbound degree at build time and store it on the node; only evaluate when `arrivals.size === expected`. * **sequence/epoch numbers:** have upstream nodes attach a monotonically increasing `epoch`; the join discards older/duplicate epochs. * **dedupe middleware:** wrap upstream emits with a small cache (e.g., LRU keyed by hash of payload) to prevent identical replays. * **exactly-once checkpoint:** if langgraph offers a checkpointer, mark the aggregate’s completion with a CAS flag so replays don’t re-enter. tiny sketch: ```ts type Key = `${string}:${string}` // `${runId}:${branchId}` class AggregateOnce { private arrivals = new Map<string, Set<string>>(); // runId -> set(branchId) constructor(private expected: number) {} onInput(runId: string, branchId: string, payload: unknown) { const set = this.arrivals.get(runId) ?? new Set<string>(); if (set.has(branchId)) return; // dedupe set.add(branchId); this.arrivals.set(runId, set); if (set.size === this.expected) { this.arrivals.delete(runId); // reset for next run return this.evaluate(); // safe to trigger once } } } ``` mit-licensed work; we’ve cataloged **16** such patterns and helped many teams stabilize joins like this (cold start → \~600 stars in \~60 days). if you want the full “join-once + replay guard” snippet we use, say the word and i’ll share.
yindo commented 2026-02-15 18:16:01 -05:00
Author
Owner
Copy Link
Copy Source

@wenshen35105 commented on GitHub (Aug 16, 2025):

I think the https://langchain-ai.github.io/langgraphjs/how-tos/defer-node-execution/ is able to slove this issue perfectly! Thanks!

@wenshen35105 commented on GitHub (Aug 16, 2025): I think the https://langchain-ai.github.io/langgraphjs/how-tos/defer-node-execution/ is able to slove this issue perfectly! Thanks!
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
bug
checkpointer
docs-needed
documentation
enhancement
good first issue
good writeup
help wanted
invalid
langgraph-supervisor
open-swe-auto
platform
prebuilt
pull-request
Mirrored from GitHub Pull Request
 question
No Label
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraphjs#337
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?