Parallel & foreach - ai_kit

createParallelStep and createForEachStep let you run multiple sub-steps concurrently. Combine them to transform lists or aggregate analytical workloads efficiently.

Need to parallelise full workflow branches (each with multiple steps)? Use the builder-level branchParallel helper instead of createParallelStep. The two features target different layers.

Chunking pipeline

import {
  Chunk,
  TChunkDocument,
  createForEachStep,
  createParallelStep,
  createStep,
  createWorkflow,
} from "@ai_kit/core";
import { z } from "zod";

const chunkText = createStep<{ text: string }, Chunk[]>({
  id: "chunk-text",
  description: "Split the source text into homogeneous segments",
  handler: async ({ input }) => {
    const document = TChunkDocument.fromText(input.text);
    return document.chunk({
      chunkSize: 200,
      chunkOverlap: 20,
      metadata: { source: "raw-text" },
    });
  },
});

const embedChunk = createStep<Chunk, number[]>({
  id: "embed-chunk",
  description: "Compute an embedding for a chunk",
  handler: async ({ input }) => {
    return Array.from({ length: 3 }, (_, i) => input.content.length * (i + 1));
  },
});

const tagChunk = createStep<Chunk, string[]>({
  id: "tag-chunk",
  description: "Extract keywords from the chunk",
  handler: async ({ input }) => {
    return input.content
      .split(/[^a-zA-ZÀ-ÿ]+/)
      .filter(word => word.length > 4)
      .slice(0, 5);
  },
});

const processChunk = createParallelStep({
  id: "process-chunk",
  description: "Run analytical tasks in parallel for a chunk",
  steps: {
    embedding: embedChunk,
    tags: tagChunk,
  },
});

const foreachChunk = createForEachStep({
  id: "foreach-chunk",
  description: "Process each chunk reusing the parallel step",
  items: ({ input }) => input,
  itemStep: processChunk,
  concurrency: 4,
});

export const chunkingWorkflow = createWorkflow({
  id: "chunking-parallel-pipeline",
  description: "Chunking + parallel processing for each segment",
  inputSchema: z.object({ text: z.string().min(1) }),
  outputSchema: z.array(
    z.object({
      embedding: z.array(z.number()),
      tags: z.array(z.string()),
    }),
  ),
})
  .then(chunkText)
  .then(foreachChunk)
  .commit();

createForEachStep returns an array by default—set collect when you need to merge the results (for example to concatenate embeddings). TChunkDocument ensures consistent chunking and propagates metadata (source: "raw-text").

Manage concurrency

concurrency in createForEachStep limits how many items run in parallel (defaults to 1). Increase it when handlers are I/O bound.
Steps declared in createParallelStep run simultaneously and their outputs are grouped into an object.
Mix parallel and foreach to optimise pipelines without sacrificing readability.

Observability

Every nested execution emits events (step:parallel:start, step:parallel:success, …). Watch them with run.watch() to monitor which branches are active and how long they take. Need dynamic decisions? Read Conditional branches. Want to loop until a predicate is met? Head to While loops.