Advanced Patterns Examples

import { StateGraph, Annotation, Send, START, END } from "@langchain/langgraph";

const State = Annotation.Root({
  items: Annotation<string[]>,
  results: Annotation<string[]>({
    reducer: (a, b) => a.concat(b),
    default: () => []
  }),
  summary: Annotation<string>()
});

// Fan out to process each item
const fanOut = (state: typeof State.State) => {
  return state.items.map(item =>
    new Send("process", { items: [item], results: [] })
  );
};

// Process individual item
const process = (state: typeof State.State) => ({
  results: [`Processed: ${state.items[0]}`]
});

// Aggregate results
const aggregate = (state: typeof State.State) => ({
  summary: `Processed ${state.results.length} items`
});

const graph = new StateGraph(State)
  .addNode("process", process)
  .addNode("aggregate", aggregate)
  .addConditionalEdges(START, fanOut)
  .addEdge("process", "aggregate")
  .addEdge("aggregate", END)
  .compile();

const result = await graph.invoke({ items: ["a", "b", "c"] });
// { items: ["a", "b", "c"], results: [...], summary: "Processed 3 items" }

const subgraph = new StateGraph(SubState)
  .addNode("step1", step1)
  .addNode("step2", step2)
  .compile();

const mainGraph = new StateGraph(MainState)
  .addNode("prepare", prepareNode)
  .addNode("subprocess", subgraph)  // Use compiled graph as node
  .addNode("finalize", finalizeNode)
  .addEdge(START, "prepare")
  .addEdge("prepare", "subprocess")
  .addEdge("subprocess", "finalize")
  .addEdge("finalize", END)
  .compile();

import { interrupt, isInterrupted, INTERRUPT, Command } from "@langchain/langgraph";

const State = Annotation.Root({
  data: Annotation<any>(),
  approved: Annotation<boolean>()
});

const processNode = (state: typeof State.State) => ({
  data: transformData(state.data)
});

const reviewNode = async (state: typeof State.State) => {
  const approval = interrupt({
    message: "Please review the processed data",
    data: state.data
  });

  return { approved: approval.approved };
};

const graph = new StateGraph(State)
  .addNode("process", processNode)
  .addNode("review", reviewNode)
  .addEdge(START, "process")
  .addEdge("process", "review")
  .addEdge("review", END)
  .compile({ checkpointer: new MemorySaver() });

const config = { configurable: { thread_id: "approval-1" } };

// First call - interrupts
const result1 = await graph.invoke({ data: inputData }, config);

if (isInterrupted(result1)) {
  console.log("Waiting for approval");
  console.log(result1[INTERRUPT][0].value);

  // Resume with approval
  const result2 = await graph.invoke(
    new Command({ resume: { approved: true } }),
    config
  );
  console.log("Approved:", result2.approved);
}