Workflow.foreach()

그만큼.foreach()메서드는 배열의 각 항목에 대해 단계를 실행하는 루프를 생성합니다. 항상 원래 순서를 유지하면서 각 반복의 출력을 포함하는 배열을 반환합니다.

사용예

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

매개변수

step:

Step

루프에서 실행할 단계 인스턴스입니다. 이전 단계는 배열 타입을 반환해야 합니다.

opts?:

object

루프의 선택적 구성입니다. concurrency 옵션은 병렬로 실행할 수 있는 반복 수를 제어합니다(기본값: 1)

number

보고

workflow:

Workflow

메서드 체이닝을 위한 Workflow 인스턴스입니다. 출력 타입은 단계 출력 타입의 배열입니다.

행동

실행 및 대기

.foreach() 메서드는 다음 단계가 실행되기 전에 모든 항목을 처리합니다. 동시성 설정과 관계없이 .foreach() 다음 단계는 모든 반복이 완료된 후에만 실행됩니다. concurrency: 1(기본값)이면 항목을 순차적으로 처리합니다. 동시성을 높이면 항목을 병렬 배치로 처리하지만, 다음 단계는 여전히 모든 배치가 끝날 때까지 기다립니다. 항목마다 여러 작업을 실행해야 한다면 중첩 Workflow를 단계로 사용하세요. 이 방식은 각 항목의 모든 작업을 함께 유지하며 여러 .foreach() 호출을 연결하는 것보다 깔끔합니다. 예시는 foreach 내부의 중첩 Workflow를 참조하세요.

출력 구조

.foreach()항상 배열을 출력합니다. 출력 배열의 각 요소는 입력 배열의 동일한 인덱스에 있는 요소를 처리한 결과에 해당합니다.

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

.foreach() 다음에 .then() 사용

.foreach() 다음에 .then()을 연결하면 다음 단계는 전체 출력 배열을 입력으로 받습니다. 모든 결과를 함께 집계하거나 처리할 수 있습니다.

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

.foreach() 다음에 .map() 사용

다음 단계에 전달하기 전에 배열 출력을 변환하려면 .map()을 사용하세요.

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

여러 체인 연결.foreach() calls

.foreach() 호출을 연결하면 각 호출은 이전 단계의 배열을 대상으로 작동합니다.

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()을 사용하세요.

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가 끝날 때까지 기다리지 않고 실시간 진행 상황을 추적할 수 있습니다.

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>

Output of the iteration (present when iterationStatus is 'success')

단일 반복 재개

.foreach() 내부에서 중단되면 각 반복이 독립적으로 중단됩니다. 각 반복을 고유한 resumeData로 하나씩 재개하려면 run.resume()에 forEachIndex를 전달하세요. forEachIndex를 생략하면 중단된 모든 반복이 동일한 데이터로 재개됩니다.

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

관련된

• foreach를 사용한 반복
• 실행.이력서()