> Discover all available pages from the documentation index: https://mastra.zisheng.pro/llms.txt

# Workflow\.foreach()

`.foreach()` 方法创建循环，为数组中的每个项目执行一个步骤。它始终返回包含每次迭代输出的数组，并保持原始顺序。

## 使用示例

```typescript
workflow.foreach(step1, { concurrency: 2 })
```

## 参数

**step** (`Step`): 在循环中执行的步骤实例。前一个步骤必须返回数组类型。

**opts** (`object`): 循环的可选配置。concurrency 选项控制可以并行运行的迭代次数（默认值：1）

## 返回值

**workflow** (`Workflow`): 用于方法链式调用的 workflow 实例。输出类型为步骤输出类型的数组。

## 行为

### 执行和等待

`.foreach()` 方法会在执行下一个步骤前处理所有项目。`.foreach()` 后面的步骤仅在每次迭代都完成后运行，与并发设置无关。使用 `concurrency: 1`（默认值）时，项目按顺序处理。使用更高并发时，项目以并行批次处理，但下一个步骤仍会等待所有批次完成。

如果你需要为每个项目运行多个操作，请使用嵌套 workflow 作为步骤。这样可将每个项目的所有操作归在一起，比链接多个 `.foreach()` 调用更清晰。示例请参阅[在 foreach 中嵌套 workflow](https://mastra.zisheng.pro/docs/workflows/control-flow)。

### 输出结构

`.foreach()` 始终输出一个数组。输出数组中的每个元素对应于处理输入数组中相同索引元素的结果。

```typescript
// Input: [{ value: 1 }, { value: 2 }, { value: 3 }]
// Step adds 10 to each value
// Output: [{ value: 11 }, { value: 12 }, { value: 13 }]
```

### 在 `.foreach()` 后使用 `.then()`

在 `.foreach()` 后链接 `.then()` 时，下一个步骤会接收整个输出数组作为输入。你可以聚合所有结果，或一起处理它们。

```typescript
workflow
  .foreach(processItemStep) // Output: array of processed items
  .then(aggregateStep) // Input: the entire array
  .commit()
```

### 在 `.foreach()` 后使用 `.map()`

使用 `.map()` 转换数组输出，然后再将其传递给下一个步骤：

```typescript
workflow
  .foreach(processItemStep)
  .map(async ({ inputData }) => ({
    total: inputData.reduce((sum, item) => sum + item.value, 0),
    count: inputData.length,
  }))
  .then(nextStep)
  .commit()
```

### 链接多个 `.foreach()` 调用

链接 `.foreach()` 调用时，每个调用都在前一个步骤的数组上操作：

```typescript
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()
```

如果 `.foreach()` 中的步骤返回数组，输出会成为数组的数组。使用带有 `.flat()` 的 `.map()` 将其扁平化：

```typescript
workflow
  .foreach(chunkStep) // Output: [[chunk1, chunk2], [chunk3, chunk4]]
  .map(async ({ inputData }) => inputData.flat()) // Output: [chunk1, chunk2, chunk3, chunk4]
  .foreach(embedStep)
  .commit()
```

### 流式传输期间的进度事件

使用 `run.stream()` 时，foreach 步骤会在每次迭代完成后发出 `workflow-step-progress` 事件。这让你无需等待整个 foreach 完成，即可跟踪实时进度。

```typescript
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"
  }
}
```

每个进度事件载荷包含：

**id** (`string`): foreach 步骤的步骤 ID

**completedCount** (`number`): 截至目前已完成的迭代次数

**totalCount** (`number`): 迭代总次数

**currentIndex** (`number`): 刚刚完成的迭代的索引

**iterationStatus** (`'success' | 'failed' | 'suspended'`): 刚刚完成的迭代的状态

**iterationOutput** (`Record<string, any>`): 迭代的输出（当 iterationStatus 为 'success' 时存在）

### 恢复单次迭代

当 `.foreach()` 内的步骤暂停时，每次迭代会独立暂停。将 `forEachIndex` 传递给 [`run.resume()`](https://mastra.zisheng.pro/reference/workflows/run-methods/resume)，以使用各自的 `resumeData` 每次恢复一个迭代。省略 `forEachIndex` 会使用相同的数据恢复每个暂停的迭代。

```typescript
await run.resume({
  step: 'approve',
  resumeData: { ok: true },
  forEachIndex: 1,
})
```

## 相关内容

- [使用 foreach 循环](https://mastra.zisheng.pro/docs/workflows/control-flow)
- [run.resume()](https://mastra.zisheng.pro/reference/workflows/run-methods/resume)