Map Reduce | Collecting Interrupts To Batch User Inputs

OSS Product Help LangGraph
1

I am implementing a custom supervisor pattern to handle a hierarchical multi-agents langgraph in langgraph.js.

I am using parallel function calling to identify multiple hand-offs that I want to hand-off to subagents. Each sub-agent has its own graph and interrupt logic for handling user input and confirmation.

Rather than executing each sequentially, I want to execute them all concurrently with Send() as per the map-reduce pattern. How would this work with interrupts? I would like to collect all the interrupts while they are concurrently running and then batch with a synthesizer node that generates a refined message to the user for asking for additional context.

The batching the user inputs is where I am confused. Is this possible with langgraph?

2

Hi @jordanparker6

is this something that could help (CLI example)?

// Minimal working example: fan-out subagents, collect concurrent interrupts, batch-resume
// Node.js 18+, ESM
import {
  Annotation,
  Command,
  END,
  interrupt,
  INTERRUPT,
  MemorySaver,
  Send,
  START,
  StateGraph,
} from "@langchain/langgraph";
import { createInterface } from "node:readline/promises";
import { stdin as input, stdout as output } from "node:process";

// Build graph with an Annotation-based state schema
const StateAnnotation = Annotation.Root({
  agents: Annotation({
    default: () => [{ id: "alpha" }, { id: "beta" }],
  }),
  agent: Annotation(),
  answers: Annotation({
    reducer: (left, right) => ({ ...(left ?? {}), ...(right ?? {}) }),
    default: () => ({}),
  }),
});
const builder = new StateGraph(StateAnnotation);

// Fan-out to parallel workers using Send (map phase)
builder.addNode(
  "fanout",
  async (state) => {
    const agents = state.agents ?? [{ id: "alpha" }, { id: "beta" }];
    return new Command({
      goto: agents.map((agent) => new Send("worker", { agent })),
    });
  },
  { ends: ["worker"] },
);

// Each worker interrupts for user input
builder.addNode("worker", async (state) => {
  const id = state.agent.id;
  const response = await interrupt({
    kind: "ask",
    agentId: id,
    prompt: `Please provide details for ${id}`,
  });
  // Persist the answer keyed by agent id
  return { answers: { [id]: response } };
});

// Wiring
builder.addEdge(START, "fanout");
builder.addEdge("worker", END); // all workers must finish for the graph to end

// Compile with a checkpointer so we can pause/resume across interrupts
const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });

// Driver: run, collect all interrupts, synthesize one prompt, resume in batch
const config = { configurable: { thread_id: "demo-thread" } };

// Kick off parallel workers
let result = await graph.invoke(
  { agents: [{ id: "alpha" }, { id: "beta" }] },
  config,
);

// Collect concurrent interrupts surfaced by the runtime
const interrupts = result?.[INTERRUPT] ?? [];
if (interrupts.length > 0) {
  const rl = createInterface({ input, output });
  const resumeMap = {};
  for (const intr of interrupts) {
    console.log("Interrupted by:", intr);
    const { agentId, prompt } = intr.value;
    resumeMap[intr.id] = await rl.question(`${agentId}: ${prompt}\n> `);
  }
  await rl.close();

  // Resume all interrupted workers in one call
  result = await graph.invoke(new Command({ resume: resumeMap }), config);
}

console.log("Answers:", result.answers);