Workflow.foreach()
The .foreach() method creates a loop that executes a step for each item in an array. It always returns an array containing the output from each iteration, preserving the original order.
Usage exampleUsage example的直接链接
workflow.foreach(step1, { concurrency: 2 })
ParametersParameters的直接链接
step:
opts?:
ReturnsReturns的直接链接
workflow:
BehaviorBehavior的直接链接
Execution and waitingExecution and waiting的直接链接
The .foreach() method processes every item before the next step executes, regardless of concurrency settings. The default concurrency: 1 processes items sequentially. Higher values use parallel batches, but the next step still waits for every batch to finish.
If you need to run multiple operations per item, use a nested workflow as the step. This keeps all operations for each item together and is cleaner than chaining multiple .foreach() calls. See Nested workflows inside foreach for examples.
Output structureOutput structure的直接链接
.foreach() always outputs an array. Each element in the output array corresponds to the result of processing the element at the same index in the input array.
// Input: [{ value: 1 }, { value: 2 }, { value: 3 }]
// Step adds 10 to each value
// Output: [{ value: 11 }, { value: 12 }, { value: 13 }]
Using .then() after .foreach()using-then-after-foreach的直接链接
When you chain .then() after .foreach(), the next step receives the entire output array as its input. You can aggregate or process all results together.
workflow
.foreach(processItemStep) // Output: array of processed items
.then(aggregateStep) // Input: the entire array
.commit()
Using .map() after .foreach()using-map-after-foreach的直接链接
Use .map() to transform the array output before passing it to the next step:
workflow
.foreach(processItemStep)
.map(async ({ inputData }) => ({
total: inputData.reduce((sum, item) => sum + item.value, 0),
count: inputData.length,
}))
.then(nextStep)
.commit()
Chaining multiple .foreach() callschaining-multiple-foreach-calls的直接链接
When you chain .foreach() calls, each operates on the array from the previous step:
workflow
.foreach(stepA) // If input is [a, b, c], output is [A, B, C]
.foreach(stepB) // Operates on [A, B, C], output is [A', B', C']
.commit()
If a step inside .foreach() returns an array, the output becomes an array of arrays. Use .map() with .flat() to flatten:
workflow
.foreach(chunkStep) // Output: [[chunk1, chunk2], [chunk3, chunk4]]
.map(async ({ inputData }) => inputData.flat()) // Output: [chunk1, chunk2, chunk3, chunk4]
.foreach(embedStep)
.commit()
Progress events during streamingProgress events during streaming的直接链接
When using run.stream(), foreach steps emit a workflow-step-progress event after each iteration completes. This lets you track real-time progress without waiting for the entire foreach to finish.
const run = await workflow.createRun()
const stream = run.stream({ inputData })
for await (const chunk of stream) {
if (chunk.type === 'workflow-step-progress') {
console.log(`${chunk.payload.completedCount}/${chunk.payload.totalCount}`)
// e.g. "1/3", "2/3", "3/3"
}
}
Each progress event payload contains:
id:
completedCount:
totalCount:
currentIndex:
iterationStatus:
iterationOutput?:
Resuming a single iterationResuming a single iteration的直接链接
When a step inside .foreach() suspends, each iteration suspends independently. Pass forEachIndex to run.resume() to resume one iteration at a time with its own resumeData. Omitting forEachIndex resumes every suspended iteration with the same data.
await run.resume({
step: 'approve',
resumeData: { ok: true },
forEachIndex: 1,
})