DEV Community

Cover image for A Planner That Spawns N Workers: LangGraph.js Send() for Fan-Out You Can't Hardcode
Gabriel Anhaia
Gabriel Anhaia

Posted on AI-assisted

A Planner That Spawns N Workers: LangGraph.js Send() for Fan-Out You Can't Hardcode

#typescript #langgraph #ai #agents

Yesterday the planner node split a job into four subtasks. Today a user typed "check every package in the monorepo" and it returned 312. Your graph has four worker nodes wired by hand, and there is no fifth edge to send the rest down.

LangGraph.js has a primitive for work whose width you only learn at runtime: Send. A conditional edge returns one Send per subtask, and each one starts its own run of a worker node with its own input. A join node runs once every branch is done.

The graph below uses a stub model, so the file runs offline. The production problem is the 312 itself: nothing in Send stops a bad plan from turning into 312 model calls, so the planner has to.

Static edges know the width at build time

addEdge wires a fixed shape. Three edges out of a node give you three branches on every run. The Graph API docs name the case this can't express: map-reduce, where the number of downstream edges and their inputs isn't known until runtime. Their answer is to let a conditional edge return Send objects, and they define it in one line: "Send takes two arguments: first is the name of the node, and second is the state to pass to that node."

That second argument is the useful part. Each worker gets its own input, not the whole graph state. The worker handling subtask 7 sees subtask 7. It doesn't know there are 311 siblings, and it can't read their data by accident.

Install LangGraph, its @langchain/core peer and Zod, and run the file with tsx:

npm i @langchain/langgraph @langchain/core zod
npm i -D tsx typescript
npx tsx planner.ts
Enter fullscreen mode Exit fullscreen mode

Version 1.5.1 is the newest @langchain/langgraph on the releases page as I write. I haven't run this file, so pin whatever version you install and compare what it prints with the description after part 4.

The graph: planner, Send, workers, join

One file, planner.ts, in four parts. Paste them in order. First the state. task is a plain key that only ever arrives through a Send payload. results gets a reducer, because the docs say "nodes that run in parallel are part of the same super-step," and N workers writing one plain key in one superstep is an error. I covered that error and its fix in an earlier post on reducers.

// planner.ts (1/4): state
import {
  StateGraph, StateSchema, ReducedValue, Send,
  START, END, GraphRecursionError,
} from "@langchain/langgraph";
import * as z from "zod";

const Subtask = z.object({ id: z.string(), query: z.string() });
const Result = z.object({
  id: z.string(),
  ok: z.boolean(),
  text: z.string(),
});
type Result = z.infer<typeof Result>;

// N workers write `results` in one superstep: needs a reducer.
// Sorting by id keeps the output independent of finish order.
const byId = (cur: Result[], next: Result[]) =>
  [...cur, ...next].sort((a, b) => a.id.localeCompare(b.id));

const State = new StateSchema({
  goal: z.string(),
  plan: z.array(Subtask).default(() => []),
  cut: z.number().default(0),
  task: Subtask.optional(),
  results: new ReducedValue(
    z.array(Result).default(() => []),
    { reducer: byId },
  ),
  answer: z.string().default(""),
});
type S = typeof State.State;
Enter fullscreen mode Exit fullscreen mode

Next, the stub model and the planner. The stub throws for any prompt containing "flaky", so one branch fails on purpose. The planner caps the plan at MAX_WIDTH and records how many subtasks it cut. The cap lives in a node, not the router, because only a node can write to state.

// planner.ts (2/4): stub model and planner
const MAX_WIDTH = 8;

const sleep = (ms: number) =>
  new Promise((resolve) => setTimeout(resolve, ms));

// Stands in for an LLM call so the file runs offline.
async function fakeModel(prompt: string): Promise<string> {
  await sleep(20);
  if (prompt.includes("flaky")) {
    throw new Error(`model 503 on "${prompt}"`);
  }
  return `notes on ${prompt}`;
}

// A real planner asks the model for this list.
// The stub splits the goal on commas.
const planner = async (s: S) => {
  const all = s.goal
    .split(",")
    .map((q) => q.trim())
    .filter(Boolean)
    .map((query, i) => ({
      id: String(i).padStart(3, "0"),
      query,
    }));
  return {
    plan: all.slice(0, MAX_WIDTH),
    cut: Math.max(0, all.length - MAX_WIDTH),
  };
};
Enter fullscreen mode Exit fullscreen mode

Then the fan-out, the worker and the join. fanOut returns one Send per subtask, or the string "join" when the plan is empty, so an empty plan still produces an answer. Each worker reads only s.task.

// planner.ts (3/4): fan-out, worker, join
const fanOut = (s: S) =>
  s.plan.length > 0
    ? s.plan.map((task) => new Send("worker", { task }))
    : "join";

const worker = async (s: S) => {
  const t = s.task;
  if (!t) throw new Error("worker needs a task");
  try {
    const text = await fakeModel(t.query);
    return { results: [{ id: t.id, ok: true, text }] };
  } catch (err) {
    const text = err instanceof Error ? err.message : String(err);
    return { results: [{ id: t.id, ok: false, text }] };
  }
};

const join = async (s: S) => {
  const failed = s.results.filter((r) => !r.ok).length;
  const lines = s.results.map(
    (r) => `${r.id} ${r.ok ? "ok " : "ERR"} ${r.text}`,
  );
  lines.push(`${failed} failed, ${s.cut} cut by the cap`);
  return { answer: lines.join("\n") };
};
Enter fullscreen mode Exit fullscreen mode

Last, the graph and a run. The worker -> join edge is the join: the docs' own map-reduce example wires its fan-out target to the reduce node the same way, with one plain addEdge.

// planner.ts (4/4): graph and run
const graph = new StateGraph(State)
  .addNode("planner", planner)
  .addNode("worker", worker)
  .addNode("join", join)
  .addEdge(START, "planner")
  .addConditionalEdges("planner", fanOut)
  .addEdge("worker", "join")
  .addEdge("join", END)
  .compile();

async function main() {
  const goal = "postgres, mysql, sqlite, duckdb, flaky-db";
  const out = await graph.invoke(
    { goal },
    { maxConcurrency: 4, recursionLimit: 10 },
  );
  console.log(out.answer);
}

main().catch((err) => {
  if (err instanceof GraphRecursionError) {
    console.error("graph ran past recursionLimit");
  } else {
    console.error(err);
  }
});
Enter fullscreen mode Exit fullscreen mode

Going by the code, answer holds five lines sorted by id, with 004 marked ERR and the 503 message, then a last line reading 1 failed, 0 cut by the cap. The planner decided the width. Nothing in the graph definition mentions five.

When one branch throws

The worker catches its own error and turns it into a result with ok: false. To see why, delete the try/catch and run the same goal.

The how-to guide says the superstep is transactional: "If any of these branches raises an exception, none of the updates are applied to the state." Your four good workers ran and their model calls were billed. Their results are discarded with the failure, and invoke rejects. In the LangGraph.js runner source, a single node error is rethrown as-is, and several are wrapped in an AggregateError.

The siblings don't stop the moment one fails, either. The same runner aborts a shared signal that every task in the step receives, then waits for all of them to settle before it rethrows. A branch stops early only if its code watches that signal. A model call that ignores it runs to completion and usually gets billed.

With a checkpointer attached, the docs add one mitigation: results from the nodes that succeeded in that superstep are saved and don't repeat when you resume. Without one, rerunning the graph pays for every branch again.

Catching inside the node has its own cost. LangGraph's retryPolicy only sees errors that escape the node, so the try/catch above also switches off its retries. If you want them, retry inside the catch yourself, or move to the node options. The fault-tolerance docs describe an errorHandler that runs "only after retries are exhausted," next to a per-node timeout in milliseconds. Both need @langchain/langgraph 1.4.0 or later. The plain try/catch works on any version and keeps the failure inside the branch that had it, which is usually what a fan-out wants: the join decides whether four answers out of five is good enough.

recursionLimit counts depth, not width

recursionLimit is the setting people reach for when they worry about a runaway agent. The Graph API docs define it as "the maximum number of super-steps the graph can execute during a single execution," with a default of 25 (it's DEFAULT_RECURSION_LIMIT = 25 in the config source). Going past it raises GraphRecursionError, which main() above catches. It goes at the top level of the config, not inside configurable.

Count the supersteps in this graph. The planner is one. All the workers run in parallel, so together they are one more. The join is a third. Five subtasks or five hundred, it's three supersteps. recursionLimit measures how deep the run goes and never sees how wide it is.

It earns its place once the graph loops. Add a replan edge, where join routes back to planner when too many branches failed, and every round adds three supersteps. With recursionLimit: 10 you get about three rounds before GraphRecursionError fires. That is its job: a ceiling on rounds. The width of each round is a separate number, and no config key sets it for you.

Cap the width where state can see it

maxConcurrency looks like the cap. The how-to guide calls it the setting that controls "the maximum number of concurrent tasks," and in the runner, new tasks start only while fewer than that many are in flight. Five hundred Sends with maxConcurrency: 4 is still 500 model calls, four at a time. That protects your rate limit. It does nothing for your bill.

The total is yours to enforce, and the planner node is the place to do it. A conditional edge only routes. The docs type it with ConditionalEdgeRouter and describe it as "no state updates, just routing." Trim the plan inside fanOut and the cut is invisible: state still holds 500 subtasks, the join sees 8 results, and nothing records what happened. Trimming in the planner writes both numbers to state. Drop this into main() to try a bad plan:

const huge = Array.from({ length: 500 }, (_, i) => `pkg-${i}`);
const big = await graph.invoke({ goal: huge.join(", ") });
console.log(big.answer.split("\n").at(-1));
Enter fullscreen mode Exit fullscreen mode

Eight workers run, cut is 492, and the join's last line says so.

A few decisions sit next to the cap:

  • Truncate or refuse. Truncating returns a partial answer that admits it is partial. Throwing in the planner when the plan is over the cap is better when partial is wrong, such as a migration that has to touch every table.
  • Tell the model, enforce in code. Put the limit in the planner prompt and keep the slice anyway, because the model can ignore the prompt and can't ignore the slice.
  • Dedupe before you slice. A planner that repeats a subtask spends cap slots on duplicates.
  • Budget is width times rounds. With a replan loop, worst-case worker calls are MAX_WIDTH multiplied by the rounds your recursionLimit allows. Work that number out before the graph ships.

If this was useful

A planner that decides the shape of the work at runtime, state that survives many parallel writers, and loops that know when to stop are what AI That Plans is about. It is book 4 of AI in TypeScript and works through LangGraph.js state, memory, and multi-step and multi-agent graphs, all in TypeScript. If Send was the missing piece in a graph you are building, the book covers the pieces around it.

AI That Plans — AI in TypeScript

The whole series, from your first LLM call to agents in production, is at xgabriel.com/ai-in-typescript.

Top comments (0)