> Discover all available pages from the documentation index: https://mastra.zisheng.pro/zh-HK/llms.txt # 控制流程 Workflow 會執行一連串預先定義的任務,而你可以控制流程的執行方式。任務分為多個**步驟**,並可按需要以不同方式執行:依次執行、並行執行,或根據條件採用不同路徑。 每個步驟都透過已定義的 schema 連接至 Workflow 中的下一個步驟,確保資料受控且一致。 ## 核心原則 - 第一個步驟的 `inputSchema` 必須與 Workflow 的 `inputSchema` 相符。 - 最後一個步驟的 `outputSchema` 必須與 Workflow 的 `outputSchema` 相符。 - 每個步驟的 `outputSchema` 必須與下一個步驟的 `inputSchema` 相符。 - 如不相符,請使用[輸入資料映射](#input-data-mapping),將資料轉換成所需結構。 ## 使用 `.then()` 串連步驟 使用 `.then()` 依次執行步驟,讓每個步驟都可存取前一個步驟的結果。 ![使用 .then() 串連步驟](/zh-HK/assets/images/workflows-control-flow-then-bde5e0fbefe5c64c19a8c3471c0e8439.jpg) ```typescript const step1 = createStep({ inputSchema: z.object({ message: z.string(), }), outputSchema: z.object({ formatted: z.string(), }), }) const step2 = createStep({ inputSchema: z.object({ formatted: z.string(), }), outputSchema: z.object({ emphasized: z.string(), }), }) export const testWorkflow = createWorkflow({ inputSchema: z.object({ message: z.string(), }), outputSchema: z.object({ emphasized: z.string(), }), }) .then(step1) .then(step2) .commit() ``` ## 使用 `.parallel()` 同時執行步驟 使用 `.parallel()` 同時執行步驟。所有並行步驟都必須完成,Workflow 才會繼續執行下一個步驟。各步驟的 `id` 會在定義後續步驟的 `inputSchema` 時使用;此值亦會成為 `inputData` 物件中的 key,用來存取前一個步驟的值。之後的步驟可引用或合併各並行步驟的輸出。 ![使用 .parallel() 並行執行步驟](/zh-HK/assets/images/workflows-control-flow-parallel-8e7fe60f1c4daa510431b37c973f6f8d.jpg) ```typescript const step1 = createStep({ id: 'step-1', }) const step2 = createStep({ id: 'step-2', }) const step3 = createStep({ id: 'step-3', inputSchema: z.object({ 'step-1': z.object({ formatted: z.string(), }), 'step-2': z.object({ emphasized: z.string(), }), }), outputSchema: z.object({ combined: z.string(), }), execute: async ({ inputData }) => { const { formatted } = inputData['step-1'] const { emphasized } = inputData['step-2'] return { combined: `${formatted} | ${emphasized}`, } }, }) export const testWorkflow = createWorkflow({ inputSchema: z.object({ message: z.string(), }), outputSchema: z.object({ combined: z.string(), }), }) .parallel([step1, step2]) .then(step3) .commit() ``` ### 輸出結構 步驟並行執行時,輸出是一個物件;每個 key 都是步驟的 `id`,而其值則是該步驟的輸出。你可以獨立存取每個並行步驟的結果。 ```typescript const step1 = createStep({ id: 'format-step', inputSchema: z.object({ message: z.string() }), outputSchema: z.object({ formatted: z.string() }), execute: async ({ inputData }) => ({ formatted: inputData.message.toUpperCase(), }), }) const step2 = createStep({ id: 'count-step', inputSchema: z.object({ message: z.string() }), outputSchema: z.object({ count: z.number() }), execute: async ({ inputData }) => ({ count: inputData.message.length, }), }) const step3 = createStep({ id: 'combine-step', // The inputSchema must match the structure of parallel outputs inputSchema: z.object({ 'format-step': z.object({ formatted: z.string() }), 'count-step': z.object({ count: z.number() }), }), outputSchema: z.object({ result: z.string() }), execute: async ({ inputData }) => { // Access each parallel step's output by its id const formatted = inputData['format-step'].formatted const count = inputData['count-step'].count return { result: `${formatted} (${count} characters)`, } }, }) export const testWorkflow = createWorkflow({ id: 'parallel-output-example', inputSchema: z.object({ message: z.string() }), outputSchema: z.object({ result: z.string() }), }) .parallel([step1, step2]) .then(step3) .commit() // When executed with { message: "hello" } // The parallel output structure will be: // { // "format-step": { formatted: "HELLO" }, // "count-step": { count: 5 } // } ``` **重點:** - 每個並行步驟的輸出都以其 `id` 作為 key - 所有並行步驟會同時執行 - 下一個步驟會收到包含所有並行步驟輸出的物件 - 你必須定義後續步驟的 `inputSchema`,使其符合此結構 ### 處理步驟失敗 如果任何並行步驟拋出錯誤,整個並行區塊都會失敗。若要建立可容許部分步驟失敗的穩健並行 Workflow(例如多個研究 Agent 中,其中一個的驗證 token 可能已過期),請在步驟本身使用 try/catch 處理錯誤: ```typescript const resilientStep = createStep({ id: 'researcher', inputSchema: z.object({ query: z.string() }), outputSchema: z.object({ brief: z.string().nullable(), failed: z.boolean(), }), execute: async ({ inputData }) => { try { const result = await fetchExternalData(inputData.query) return { brief: result, failed: false } } catch { return { brief: null, failed: true } } }, }) ``` 這樣一來,步驟總會成功並傳回具型別的結果,而下游步驟則可篩除失敗的結果: ```typescript const writerStep = createStep({ id: 'writer', inputSchema: z.object({ 'researcher-a': z.object({ brief: z.string().nullable(), failed: z.boolean() }), 'researcher-b': z.object({ brief: z.string().nullable(), failed: z.boolean() }), }), outputSchema: z.object({ synthesis: z.string() }), execute: async ({ inputData }) => { const briefs = Object.values(inputData) .filter(v => !v.failed && v.brief) .map(v => v.brief) return { synthesis: briefs.join('; ') } }, }) ``` 請參閱[選擇合適的模式](#choosing-the-right-pattern),了解何時應使用 `.parallel()` 或 `.foreach()`。 ## 使用 `.branch()` 實作條件邏輯 使用 `.branch()` 根據條件選擇要執行的步驟。分支中的所有步驟都需要相同的 `inputSchema` 和 `outputSchema`,因為分支必須採用一致的 schema,Workflow 才能沿不同路徑執行。 ![使用 .branch() 實作條件分支](/zh-HK/assets/images/workflows-control-flow-branch-1913ef107ba0198d73aa3c0a65145b7a.jpg) ```typescript const step1 = createStep({...}) const stepA = createStep({ inputSchema: z.object({ value: z.number() }), outputSchema: z.object({ result: z.string() }) }); const stepB = createStep({ inputSchema: z.object({ value: z.number() }), outputSchema: z.object({ result: z.string() }) }); export const testWorkflow = createWorkflow({ inputSchema: z.object({ value: z.number() }), outputSchema: z.object({ result: z.string() }) }) .then(step1) .branch([ [async ({ inputData: { value } }) => value > 10, stepA], [async ({ inputData: { value } }) => value <= 10, stepB] ]) .commit(); ``` ### 輸出結構 使用條件分支時,只會執行條件最先求值為 `true` 的一個分支。輸出結構與 `.parallel()` 類似,結果以已執行步驟的 `id` 作為 key。 ```typescript const step1 = createStep({ id: 'initial-step', inputSchema: z.object({ value: z.number() }), outputSchema: z.object({ value: z.number() }), execute: async ({ inputData }) => inputData, }) const highValueStep = createStep({ id: 'high-value-step', inputSchema: z.object({ value: z.number() }), outputSchema: z.object({ result: z.string() }), execute: async ({ inputData }) => ({ result: `High value: ${inputData.value}`, }), }) const lowValueStep = createStep({ id: 'low-value-step', inputSchema: z.object({ value: z.number() }), outputSchema: z.object({ result: z.string() }), execute: async ({ inputData }) => ({ result: `Low value: ${inputData.value}`, }), }) const finalStep = createStep({ id: 'final-step', // The inputSchema must account for either branch's output inputSchema: z.object({ 'high-value-step': z.object({ result: z.string() }).optional(), 'low-value-step': z.object({ result: z.string() }).optional(), }), outputSchema: z.object({ message: z.string() }), execute: async ({ inputData }) => { // Only one branch will have executed const result = inputData['high-value-step']?.result || inputData['low-value-step']?.result return { message: result } }, }) export const testWorkflow = createWorkflow({ id: 'branch-output-example', inputSchema: z.object({ value: z.number() }), outputSchema: z.object({ message: z.string() }), }) .then(step1) .branch([ [async ({ inputData }) => inputData.value > 10, highValueStep], [async ({ inputData }) => inputData.value <= 10, lowValueStep], ]) .then(finalStep) .commit() // When executed with { value: 15 } // Only the high-value-step executes, output structure: // { // "high-value-step": { result: "High value: 15" } // } // When executed with { value: 5 } // Only the low-value-step executes, output structure: // { // "low-value-step": { result: "Low value: 5" } // } ``` **重點:** - 根據條件求值的順序,只會執行一個分支 - 輸出以已執行步驟的 `id` 作為 key - 後續步驟應處理所有可能的分支輸出 - 當下一個步驟需要處理多個可能的分支時,請在 `inputSchema` 中使用可選欄位 - 條件會按定義順序求值 ## 輸入資料映射 使用 `.then()`、`.parallel()` 或 `.branch()` 時,有時需要轉換前一個步驟的輸出,使其與下一個步驟的輸入相符。在這些情況下,你可以使用 `.map()` 存取 `inputData` 並進行轉換,為下一個步驟建立合適的資料結構。 ![使用 .map() 進行映射](/zh-HK/assets/images/workflows-data-mapping-map-87fd84a06b4bbf4b93868a5db99ca179.jpg) ```typescript const step1 = createStep({...}); const step2 = createStep({...}); export const testWorkflow = createWorkflow({...}) .then(step1) .map(async ({ inputData }) => { const { foo } = inputData; return { bar: `new ${foo}`, }; }) .then(step2) .commit(); ``` `.map()` 方法提供額外的輔助函式,以處理更複雜的映射情境。 **可用的輔助函式:** - [`getStepResult()`](https://mastra.zisheng.pro/zh-HK/reference/workflows/workflow-methods/map):存取特定步驟的完整輸出 - [`getInitData()`](https://mastra.zisheng.pro/zh-HK/reference/workflows/workflow-methods/map):存取 Workflow 的初始輸入資料 - [`mapVariable()`](https://mastra.zisheng.pro/zh-HK/reference/workflows/workflow-methods/map):使用聲明式物件語法擷取欄位並重新命名 ### Parallel 和 Branch 輸出 處理 `.parallel()` 或 `.branch()` 的輸出時,你可以先使用 `.map()` 轉換資料結構,再將資料傳遞至下一個步驟。當你需要攤平或重組輸出時,這尤其實用。 ```typescript export const testWorkflow = createWorkflow({...}) .parallel([step1, step2]) .map(async ({ inputData }) => { // Transform the parallel output structure return { combined: `${inputData["step1"].value} - ${inputData["step2"].value}` }; }) .then(nextStep) .commit(); ``` 你亦可以使用 `.map()` 提供的輔助函式: ```typescript export const testWorkflow = createWorkflow({...}) .branch([ [condition1, stepA], [condition2, stepB] ]) .map(async ({ inputData, getStepResult }) => { // Access specific step results const stepAResult = getStepResult("stepA"); const stepBResult = getStepResult("stepB"); // Return the result from whichever branch executed return stepAResult || stepBResult; }) .then(nextStep) .commit(); ``` ## 循環執行步驟 Workflow 支援多種循環方法,讓你重複執行步驟,直至或只要符合某個條件,亦可逐項處理陣列。循環可與 `.then()` 等其他控制方法配合使用。 ### 使用 `.dountil()` 循環 使用 `.dountil()` 重複執行步驟,直至條件變為 true。 ![使用 .dountil() 重複執行](/zh-HK/assets/images/workflows-control-flow-dountil-6b7b06e872f3bd878f69c716b0e38ae6.jpg) ```typescript const step1 = createStep({...}); const step2 = createStep({ execute: async ({ inputData }) => { const { number } = inputData; return { number: number + 1 }; } }); export const testWorkflow = createWorkflow({}) .then(step1) .dountil(step2, async ({ inputData: { number } }) => number > 10) .commit(); ``` ### 使用 `.dowhile()` 循環 使用 `.dowhile()` 在條件維持為 true 期間重複執行步驟。 ![使用 .dowhile() 重複執行](/zh-HK/assets/images/workflows-control-flow-dowhile-09bba2d43fb44352f458c144484326ed.jpg) ```typescript const step1 = createStep({...}); const step2 = createStep({ execute: async ({ inputData }) => { const { number } = inputData; return { number: number + 1 }; } }); export const testWorkflow = createWorkflow({}) .then(step1) .dowhile(step2, async ({ inputData: { number } }) => number < 10) .commit(); ``` ### 使用 `.foreach()` 循環 使用 `.foreach()` 對陣列中的每個項目執行相同步驟。輸入必須是 `array` 型別,循環才能逐一處理其中的值,並將步驟邏輯套用至每個值。關於何時使用 `.foreach()` 或其他方法,請參閱[選擇合適的模式](#choosing-the-right-pattern)。 ![使用 .foreach() 重複執行](/zh-HK/assets/images/workflows-control-flow-foreach-a5b6f38d8797c4d1b7dca93879d709f7.jpg) ```typescript const step1 = createStep({ inputSchema: z.string(), outputSchema: z.string(), execute: async ({ inputData }) => { return inputData.toUpperCase(); } }); const step2 = createStep({...}); export const testWorkflow = createWorkflow({ inputSchema: z.array(z.string()), outputSchema: z.array(z.string()) }) .foreach(step1) .then(step2) .commit(); ``` #### 輸出結構 `.foreach()` 方法一律傳回包含每次迭代輸出的陣列。輸出順序與輸入順序一致。 ```typescript const addTenStep = createStep({ id: 'add-ten', inputSchema: z.object({ value: z.number() }), outputSchema: z.object({ value: z.number() }), execute: async ({ inputData }) => ({ value: inputData.value + 10, }), }) export const testWorkflow = createWorkflow({ id: 'foreach-output-example', inputSchema: z.array(z.object({ value: z.number() })), outputSchema: z.array(z.object({ value: z.number() })), }) .foreach(addTenStep) .commit() // When executed with [{ value: 1 }, { value: 22 }, { value: 333 }] // Output: [{ value: 11 }, { value: 32 }, { value: 343 }] ``` #### 並行處理限制 使用 `concurrency` 控制要處理的陣列項目數量。預設值為 `1`,即依次執行步驟。提高此值可讓 `.foreach()` 同時處理多個項目。 ```typescript const step1 = createStep({...}) export const testWorkflow = createWorkflow({...}) .foreach(step1, { concurrency: 4 }) .commit(); ``` #### 彙總 `.foreach()` 之後的結果 由於 `.foreach()` 會輸出陣列,因此你可以使用 `.then()` 或 `.map()` 彙總或轉換結果。緊接 `.foreach()` 的步驟會收到整個陣列作為輸入。 ```typescript const processItemStep = createStep({ id: 'process-item', inputSchema: z.object({ value: z.number() }), outputSchema: z.object({ processed: z.number() }), execute: async ({ inputData }) => ({ processed: inputData.value * 2, }), }) const aggregateStep = createStep({ id: 'aggregate', // Input is an array of outputs from foreach inputSchema: z.array(z.object({ processed: z.number() })), outputSchema: z.object({ total: z.number() }), execute: async ({ inputData }) => ({ // Sum all processed values total: inputData.reduce((sum, item) => sum + item.processed, 0), }), }) export const testWorkflow = createWorkflow({ id: 'foreach-aggregate-example', inputSchema: z.array(z.object({ value: z.number() })), outputSchema: z.object({ total: z.number() }), }) .foreach(processItemStep) .then(aggregateStep) // Receives the full array from foreach .commit() // When executed with [{ value: 1 }, { value: 2 }, { value: 3 }] // After foreach: [{ processed: 2 }, { processed: 4 }, { processed: 6 }] // After aggregate: { total: 12 } ``` 你亦可以使用 `.map()` 轉換陣列輸出: ```typescript export const testWorkflow = createWorkflow({...}) .foreach(processItemStep) .map(async ({ inputData }) => ({ // Transform the array into a different structure values: inputData.map(item => item.processed), count: inputData.length })) .then(nextStep) .commit(); ``` #### 串連多個 `.foreach()` 呼叫 串連多個 `.foreach()` 呼叫時,每個呼叫都會處理前一個步驟輸出的陣列。如果陣列中的每個項目都需要由多個步驟依次轉換,這種方式便很實用。 ```typescript const chunkStep = createStep({ id: 'chunk', // Takes a document, returns an array of chunks inputSchema: z.object({ content: z.string() }), outputSchema: z.array(z.object({ chunk: z.string() })), execute: async ({ inputData }) => { // Split document into chunks const chunks = inputData.content.match(/.{1,100}/g) || [] return chunks.map(chunk => ({ chunk })) }, }) const embedStep = createStep({ id: 'embed', // Takes a single chunk, returns embedding inputSchema: z.object({ chunk: z.string() }), outputSchema: z.object({ embedding: z.array(z.number()) }), execute: async ({ inputData }) => ({ embedding: [/* vector embedding */], }), }) // For a single document that produces multiple chunks: export const singleDocWorkflow = createWorkflow({ id: 'single-doc-rag', inputSchema: z.object({ content: z.string() }), outputSchema: z.array(z.object({ embedding: z.array(z.number()) })), }) .then(chunkStep) // Returns array of chunks .foreach(embedStep) // Process each chunk -> array of embeddings .commit() ``` 處理多份文件,而每份文件都會產生多個 chunk 時,你有以下選擇: **選項 1:在單一步驟中處理所有文件並控制批次** ```typescript const downloadAndChunkStep = createStep({ id: "download-and-chunk", inputSchema: z.array(z.string()), // Array of URLs outputSchema: z.array(z.object({ chunk: z.string(), source: z.string() })), execute: async ({ inputData: urls }) => { // Control batching/parallelization within the step const allChunks = []; for (const url of urls) { const content = await fetch(url).then(r => r.text()); const chunks = content.match(/.{1,100}/g) || []; allChunks.push(...chunks.map(chunk => ({ chunk, source: url }))); } return allChunks; } }); export const multiDocWorkflow = createWorkflow({...}) .then(downloadAndChunkStep) // Returns flat array of all chunks .foreach(embedStep, { concurrency: 10 }) // Embed each chunk in parallel .commit(); ``` **選項 2:對文件使用 foreach 並彙總 chunk,之後再對 embedding 使用 foreach** ```typescript const downloadStep = createStep({ id: 'download', inputSchema: z.string(), // Single URL outputSchema: z.object({ content: z.string(), source: z.string() }), execute: async ({ inputData: url }) => ({ content: await fetch(url).then(r => r.text()), source: url, }), }) const chunkDocStep = createStep({ id: 'chunk-doc', inputSchema: z.object({ content: z.string(), source: z.string() }), outputSchema: z.array(z.object({ chunk: z.string(), source: z.string() })), execute: async ({ inputData }) => { const chunks = inputData.content.match(/.{1,100}/g) || [] return chunks.map(chunk => ({ chunk, source: inputData.source })) }, }) export const multiDocWorkflow = createWorkflow({ id: 'multi-doc-rag', inputSchema: z.array(z.string()), // Array of URLs outputSchema: z.array(z.object({ embedding: z.array(z.number()) })), }) .foreach(downloadStep, { concurrency: 5 }) // Download docs in parallel .foreach(chunkDocStep) // Chunk each doc -> array of chunk arrays .map(async ({ inputData }) => { // Flatten nested arrays: [[chunks], [chunks]] -> [chunks] return inputData.flat() }) .foreach(embedStep, { concurrency: 10 }) // Embed all chunks .commit() ``` **串連 `.foreach()` 的重點:** - 每個 `.foreach()` 都會處理前一個步驟的陣列 - 如果 `.foreach()` 內的步驟傳回陣列,輸出便會成為陣列的陣列 - 有需要時,配合使用 `.map()` 和 `.flat()` 攤平嵌套陣列 - 對於複雜的 RAG pipeline,選項 1(在單一步驟中處理批次)通常能提供更佳控制 #### foreach 內的嵌套 Workflow `.foreach()` 之後的步驟只會在所有迭代完成後執行。如果每個項目都需要執行多個循序操作,請使用嵌套 Workflow,而不要串連多個 `.foreach()` 呼叫。這樣可將每個項目的所有操作集中在一起,令資料流更清晰。 ```typescript // Define a workflow that processes a single document const processDocumentWorkflow = createWorkflow({ id: 'process-document', inputSchema: z.object({ url: z.string() }), outputSchema: z.object({ embeddings: z.array(z.array(z.number())), metadata: z.object({ url: z.string(), chunkCount: z.number() }), }), }) .then(downloadStep) // Download the document .then(chunkStep) // Split into chunks .then(embedChunksStep) // Embed all chunks for this document .then(formatResultStep) // Format the final output .commit() // Use the nested workflow inside foreach export const batchProcessWorkflow = createWorkflow({ id: 'batch-process-documents', inputSchema: z.array(z.object({ url: z.string() })), outputSchema: z.array( z.object({ embeddings: z.array(z.array(z.number())), metadata: z.object({ url: z.string(), chunkCount: z.number() }), }), ), }) .foreach(processDocumentWorkflow, { concurrency: 3 }) .commit() // Each document goes through all 4 steps before the next document starts (with concurrency: 1) // With concurrency: 3, up to 3 documents process their full pipelines in parallel ``` **使用嵌套 Workflow 的原因:** - **更佳的並行處理**:使用 `concurrency: N` 時,多個項目會同時執行各自的完整 pipeline。串連的 `.foreach().foreach()` 會先讓所有項目完成步驟 1,等待後再讓所有項目執行步驟 2;嵌套 Workflow 則讓每個項目獨立推進 - 收集結果前,單一項目的所有步驟會一併完成 - 相比會產生嵌套陣列的多個 `.foreach()` 呼叫,結構更簡潔 - 每次嵌套 Workflow 執行都互相獨立,並有各自的資料流 - 更容易獨立測試和重用每個項目的邏輯 **運作方式:** 1. 父 Workflow 將每個陣列項目傳遞至嵌套 Workflow 的一個實例 2. 每個嵌套 Workflow 都會為該項目執行完整的步驟序列 3. 當 `concurrency > 1` 時,多個嵌套 Workflow 會並行執行 4. 嵌套 Workflow 的最終輸出會成為結果陣列中的一個元素 5. 所有嵌套 Workflow 完成後,父 Workflow 的下一個步驟會收到完整陣列 ## 選擇合適的模式 選擇合適的控制流程方法時,可參考本節。 ### 快速參考 | 方法 | 用途 | 輸入 | 輸出 | 並行處理 | | ------------------- | ------------- | ----- | --------------------- | --------- | | `.then(step)` | 依次處理 | `T` | `U` | 不適用(每次一個) | | `.parallel([a, b])` | 對相同輸入執行不同操作 | `T` | `{ a: U, b: V }` | 全部同時執行 | | `.foreach(step)` | 對每個陣列項目執行相同操作 | `T[]` | `U[]` | 可設定(預設:1) | | `.branch([...])` | 按條件選擇路徑 | `T` | `{ selectedStep: U }` | 只執行一個分支 | ### `.parallel()` vs `.foreach()` **當單一輸入需要以不同方式處理時,使用 `.parallel()`:** ```typescript // Same user data processed differently in parallel workflow.parallel([validateStep, enrichStep, scoreStep]).then(combineResultsStep) ``` **當多個輸入需要以相同方式處理時,使用 `.foreach()`:** ```typescript // Multiple URLs each processed the same way workflow.foreach(downloadStep, { concurrency: 5 }).then(aggregateStep) ``` ### 何時使用嵌套 Workflow **在 `.foreach()` 內**——當每個陣列項目都需要多個循序步驟時: ```typescript // Each document goes through a full pipeline const processDocWorkflow = createWorkflow({...}) .then(downloadStep) .then(parseStep) .then(embedStep) .commit(); workflow.foreach(processDocWorkflow, { concurrency: 3 }) ``` 使用單一 `.foreach()` 可維持扁平的結果;串連 `.foreach().foreach()` 則會建立嵌套陣列。 **在 `.parallel()` 內**——當並行分支需要自己的多步驟 pipeline 時: ```typescript const pipelineA = createWorkflow({...}).then(step1).then(step2).commit(); const pipelineB = createWorkflow({...}).then(step3).then(step4).commit(); workflow.parallel([pipelineA, pipelineB]) ``` ### 串連模式 | 模式 | 執行情況 | 常見用途 | | ---------------------- | ----------------- | ---------------------------------------------- | | `.then().then()` | 循序步驟 | 簡單 pipeline | | `.parallel().then()` | 並行執行,然後合併 | Fan-out/fan-in | | `.foreach().then()` | 處理所有項目,然後彙總 | Map-reduce | | `.foreach().foreach()` | 建立陣列的陣列 | 避免使用——改用嵌套 Workflow,或配合使用 `.map()` 與 `.flat()` | | `.foreach(workflow)` | 每個項目執行完整 pipeline | 對每個陣列項目進行多步驟處理 | ### 同步:下一個步驟何時執行? `.parallel()` 和 `.foreach()` 都是同步點。只有在所有並行分支或所有陣列迭代完成後,Workflow 中的下一個步驟才會執行。 ```typescript workflow .parallel([stepA, stepB, stepC]) // All 3 run simultaneously .then(combineStep) // Waits for ALL 3 to finish before running .commit() workflow .foreach(processStep, { concurrency: 5 }) // Up to 5 items process at once .then(aggregateStep) // Waits for ALL items to finish before running .commit() ``` 這表示: - `.parallel()` 將所有分支輸出收集到一個物件,再傳遞至下一個步驟 - `.foreach()` 將所有迭代輸出收集到一個陣列,再傳遞至下一個步驟 - 結果無法在完成時以「串流」方式傳遞至下一個步驟 ### 並行處理行為 | 方法 | 行為 | | -------------------------- | --------------------------------------- | | `.then()` | 循序執行——每次一個步驟 | | `.parallel()` | 所有分支同時執行(沒有上限選項) | | `.foreach()` | 透過 `{ concurrency: N }` 控制——預設為 1(循序執行) | | `.foreach()` 內的嵌套 Workflow | 遵從父 Workflow 的並行處理設定 | **效能提示:** 對於 `.foreach()` 中受 I/O 限制的操作,可提高並行處理數量,以並行處理各項目: ```typescript // Process up to 10 items simultaneously workflow.foreach(fetchDataStep, { concurrency: 10 }) ``` ## 循環管理 你可以根據希望循環如何結束,以不同方式實作循環條件。 常見模式會檢查 `inputData` 傳回的值,並設定最大迭代次數;亦可在達到上限時中止執行。 ### 中止循環 使用 `iterationCount` 限制循環的執行次數。如果計數超過你的門檻,請拋出錯誤,讓步驟失敗並停止 Workflow。 ```typescript const step1 = createStep({...}); export const testWorkflow = createWorkflow({...}) .dountil(step1, async ({ inputData: { userResponse, iterationCount } }) => { if (iterationCount >= 10) { throw new Error("Maximum iterations reached"); } return userResponse === "yes"; }) .commit(); ``` ## 相關內容 - [暫停與恢復](https://mastra.zisheng.pro/zh-HK/docs/workflows/suspend-and-resume) - [Human-in-the-loop](https://mastra.zisheng.pro/zh-HK/docs/workflows/human-in-the-loop)