 構建智能內容路由)
Mastra 條件工作流實戰用 .branch() 構建智能內容路由【免費下載鏈接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.項目地址: https://gitcode.com/GitHub_Trending/ma/mastra導讀本篇教程聚焦 Mastra 工作流引擎的條件分支能力如何基于數據評估結果讓同一份工作流把不同內容智能路由到不同處理路徑。你將完整掌握createWorkflow().branch()的語法、條件函數編寫、邏輯運算符組合、多分支并行執行語義以及如何在 Playground 和代碼中驗證分支結果。讀完即可獨立構建一個短內容快速處理、長內容深度處理的實戰型條件工作流。本文是 Mastra 官方工作流課程docs/src/course/04-workflows/中條件分支系列的第四篇建議先閱讀 理解條件分支 與 創建條件步驟再進入本文的完整工作流組裝。條件工作流要解決的問題在實際業務中工作流往往需要看數據說話同樣一份內容輸入是 30 個詞的推文還是 5000 字的長文后續處理策略完全不同。用一串固定的.then()鏈式步驟無法表達這種分叉邏輯而條件分支conditional branching正是為此設計智能決策根據上一步驟的輸出數據選擇不同處理路徑性能優化簡單內容跳過昂貴處理例如長文才需要 AI 摘要個性化體驗不同場景得到不同的處理結果與建議可擴展邏輯新增條件與處理路徑時無需改動既有步驟。Mastra 工作流引擎通過.branch()方法把這一能力變成了一等公民。在源碼層面.branch()會向工作流的步驟流step flow中壓入一個type: conditional的入口見 types.ts其中包含有序的conditions數組與對應步驟這正是下文條件求值語義的底層基礎。第一步評估步驟——決定路由的依據條件分支的核心前提是有據可依。我們先創建一個評估步驟assessContentStep它負責分析內容輸出categoryshort / medium / long與complexitysimple / moderate / complex兩個路由判據import { createStep } from mastra/core import { z } from zod const assessContentStep createStep({ id: assess-content, description: Assesses content to determine processing path, inputSchema: z.object({ content: z.string(), type: z.enum([article, blog, social]).default(article), }), outputSchema: z.object({ content: z.string(), type: z.enum([article, blog, social]).default(article), wordCount: z.number(), complexity: z.enum([simple, moderate, complex]), category: z.enum([short, medium, long]), }), execute: async ({ inputData }) { const { content, type } inputData const words content.trim().split(/\s/) const wordCount words.length // Determine category by length let category: short | medium | long short if (wordCount 50) category medium if (wordCount 200) category long // Determine complexity by average word length const avgWordLength words.reduce((sum, word) sum word.length, 0) / wordCount let complexity: simple | moderate | complex simple if (avgWordLength 5) complexity moderate if (avgWordLength 7) complexity complex console.log( Assessment: ${category} content, ${complexity} complexity) return { content, type, wordCount, complexity, category, } }, })這里的路由策略是字數決定長度類別50 詞為 short50–199 為 medium≥200 為 long平均詞長決定復雜程度5 字符為 moderate7 字符為 complex。從源碼實現看createStep內部通過isStepParams分支識別這種帶execute的步驟定義并對其 schema 做標準規范化見 workflow.ts因此inputSchema/outputSchema會在執行前自動完成運行時校驗。第二步兩個分支處理步驟針對短且簡單與其余全部兩類內容準備兩條處理流水線?？焖偬幚聿襟E——服務于 short simple 內容輸出最小化建議const quickProcessingStep createStep({ id: quick-processing, description: Quick processing for short and simple content, inputSchema: z.object({ content: z.string(), type: z.enum([article, blog, social]).default(article), wordCount: z.number(), complexity: z.enum([simple, moderate, complex]), category: z.enum([short, medium, long]), }), outputSchema: z.object({ processedContent: z.string(), processingType: z.string(), recommendations: z.array(z.string()), }), execute: async ({ inputData }) { console.log(? Quick processing for short and simple content...) return { processedContent: inputData.content, processingType: quick, recommendations: [Content is concise, Consider expanding for more detail], } }, })通用處理步驟——承接非 short/simple 內容模擬更重的處理流程此處以 500ms 延遲示意昂貴操作const generalProcessingStep createStep({ id: general-processing, description: General processing for all other content, inputSchema: z.object({ content: z.string(), type: z.enum([article, blog, social]).default(article), wordCount: z.number(), complexity: z.enum([simple, moderate, complex]), category: z.enum([short, medium, long]), }), outputSchema: z.object({ processedContent: z.string(), processingType: z.string(), recommendations: z.array(z.string()), }), execute: async ({ inputData }) { console.log( General processing for non-short/simple content...) // Simulate more involved processing await new Promise(resolve setTimeout(resolve, 500)) return { processedContent: inputData.content, processingType: general, recommendations: [ Consider simplifying content, Break up long paragraphs, Add examples or explanations if needed, ], } }, })注意兩個步驟的輸出 schema 完全一致processedContent/processingType/recommendations這樣無論路由到哪條分支下游代碼都能以統一結構消費結果——這是設計分支步驟時值得借鑒的約定。第三步用 .branch() 組裝條件工作流現在把評估步驟與兩個處理步驟組合為完整的條件工作流export const conditionalWorkflow createWorkflow({ id: conditional-workflow, description: Content processing with conditional branching, inputSchema: z.object({ content: z.string(), type: z.enum([article, blog, social]).default(article), }), outputSchema: z.object({ processedContent: z.string(), processingType: z.string(), recommendations: z.array(z.string()), }), }) .then(assessContentStep) .branch([ // Branch 1: Short and simple content [async ({ inputData }) inputData.category short inputData.complexity simple, quickProcessingStep], // Branch 2: Everything else [ async ({ inputData }) !(inputData.category short inputData.complexity simple), generalProcessingStep, ], ]) .commit()這里的流水線邏輯清晰可讀then()先行評估 →branch()依據評估結果分叉 →commit()收尾定型。分支條件中的inputData指向上一步驟assessContentStep的輸出這正是評估結果驅動路由的關鍵。深入 .branch() API元組數組與兩種條件形式從源碼看.branch()接收的是一個二元組數組每個元組為[條件, 步驟]branchTBranchSteps extends Array [ConditionFunction... | { predicate: Predicate }, Step...] (steps: TBranchSteps, options?: StepFlowEntryOptions)見 workflow.ts這意味著條件位置有兩種等價寫法閉包函數本教程主線async ({ inputData }) boolean靈活、可直接書寫任意判斷邏輯聲明式 predicate 對象{ predicate: { op: gt, left: { path: inputData.value }, right: { literal: 10 } } }可序列化、便于可視化與持久化。在.branch()內部閉包形式會被直接使用其序列化標簽為condition.toString()而 predicate 形式會被轉換成等價的運行時條件函數并生成可讀標簽。兩者還可以在同一個branch()數組中混用——packages/core/src/workflows/__tests__/predicate-builder.test.ts中的測試用例驗證了這一行為見 predicate-builder.test.ts。不過該測試也指出含閉包條件的分支圖不可整體序列化持久化會拋出 closure predicates do not round-trip 錯誤需要跨請求恢復工作流時請優先使用聲明式 predicate。條件組合、|| 與 !單條條件往往不夠Mastra 的條件函數是標準 JS 表達式可以自由組合邏輯運算符運算符語義示例與——兩者都為真才命中category short complexity simple\|\|或——任一為真即命中category long \|\| complexity complex!非——條件必須為假才命中!(category short complexity simple)以本文為例Short Simplecategory short complexity simple→ 走快速處理建議最少Everything Else!(category short complexity simple)→ 走通用處理給出更多優化建議。兩條條件互為補集保證任何輸入都恰好命中一條分支是互斥路由的經典寫法。條件求值語義順序檢查、并行執行、無匹配跳過理解.branch()的運行時語義對寫出正確的工作流至關重要按順序檢查條件按照在數組中的書寫順序依次求值多條件可同時命中與 if/else 不同分支之間不是互斥關系——若多個條件同時為真對應步驟會并行執行無匹配則跳過若所有條件都不為真工作流不執行任何分支步驟直接繼續后續流程。第 2 點多分支并行有明確的測試佐證packages/core/src/workflows/evented/evented-workflow.test.ts中有一條用例讓兩個條件恒為true的步驟同時進入分支最終兩個步驟的狀態更新被正確合并{ first: 1, second: 1 }說明引擎對命中的分支步驟是并發調度、隨后合并狀態的見 evented-workflow.test.ts。因此如果你需要二選一的嚴格互斥路由務必像本文這樣把條件寫成互補形式而不是依賴分支之間的天然排他。注冊工作流并在 Playground 中驗證構建完成后把新工作流注冊進 Mastra 實例通常在src/mastra/index.tsimport { contentWorkflow, aiContentWorkflow, parallelAnalysisWorkflow, conditionalWorkflow, } from ./workflows/content-workflow export const mastra new Mastra({ workflows: { contentWorkflow, aiContentWorkflow, parallelAnalysisWorkflow, conditionalWorkflow, // Add the conditional workflow }, // ... rest of configuration })隨后在 Mastra Playground 中打開conditional-workflow分別用不同長度與類型的內容測試短內容如 20 詞、平均詞長 6應命中quick-processingprocessingType為quick推薦 2 條建議長內容如 300 詞或高復雜詞應命中general-processingprocessingType為general推薦 3 條建議。完整測試指引可參考課程下一節 測試條件邏輯。整個流程的運轉路徑為評估步驟分析內容 → 分支條件對照評估結果 → 命中步驟執行 → 輸出攜帶processingType標記實際走的分支路徑據此即可直觀確認路由是否正確。調試條件分支的實用技巧如果發現內容沒有進入預期分支按以下順序排查檢查評估步驟輸出先確認category/complexity是否符合預期——路由判據錯分支必錯核對條件邏輯把條件表達式與評估輸出的取值逐一代入驗證布爾結果單獨隔離測試將單個條件抽出來獨立求值排除多條件組合的干擾加日志追蹤在條件函數或步驟execute中增加console.log觀察求值順序與命中情況。小結至此你已掌握 Mastra 條件工作流的完整構建鏈路評估步驟產出路由判據 →.branch()依據inputData分叉 → 互補條件實現互斥路由 → 統一輸出結構消費分支結果。條件分支讓工作流從固定流水線進化為智能路由是構建內容處理、審核分流、異常兜底等場景的基礎能力。下一步可以繼續學習工作流結果流式輸出進一步提升用戶體驗?！久赓M下載鏈接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.項目地址: https://gitcode.com/GitHub_Trending/ma/mastra創作聲明：本文部分內容由AI輔助生成（AIGC），僅供參考 Mastra 條件工作流實戰：用 .branch() 構建智能內容路由
Mastra 條件工作流實戰用 .branch() 構建智能內容路由【免費下載鏈接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.項目地址: https://gitcode.com/GitHub_Trending/ma/mastra導讀本篇教程聚焦 Mastra 工作流引擎的條件分支能力如何基于數據評估結果讓同一份工作流把不同內容智能路由到不同處理路徑。你將完整掌握createWorkflow().branch()的語法、條件函數編寫、邏輯運算符組合、多分支并行執行語義以及如何在 Playground 和代碼中驗證分支結果。讀完即可獨立構建一個短內容快速處理、長內容深度處理的實戰型條件工作流。本文是 Mastra 官方工作流課程docs/src/course/04-workflows/中條件分支系列的第四篇建議先閱讀 理解條件分支 與 創建條件步驟再進入本文的完整工作流組裝。條件工作流要解決的問題在實際業務中工作流往往需要看數據說話同樣一份內容輸入是 30 個詞的推文還是 5000 字的長文后續處理策略完全不同。用一串固定的.then()鏈式步驟無法表達這種分叉邏輯而條件分支conditional branching正是為此設計智能決策根據上一步驟的輸出數據選擇不同處理路徑性能優化簡單內容跳過昂貴處理例如長文才需要 AI 摘要個性化體驗不同場景得到不同的處理結果與建議可擴展邏輯新增條件與處理路徑時無需改動既有步驟。Mastra 工作流引擎通過.branch()方法把這一能力變成了一等公民。在源碼層面.branch()會向工作流的步驟流step flow中壓入一個type: conditional的入口見 types.ts其中包含有序的conditions數組與對應步驟這正是下文條件求值語義的底層基礎。第一步評估步驟——決定路由的依據條件分支的核心前提是有據可依。我們先創建一個評估步驟assessContentStep它負責分析內容輸出categoryshort / medium / long與complexitysimple / moderate / complex兩個路由判據import { createStep } from mastra/core import { z } from zod const assessContentStep createStep({ id: assess-content, description: Assesses content to determine processing path, inputSchema: z.object({ content: z.string(), type: z.enum([article, blog, social]).default(article), }), outputSchema: z.object({ content: z.string(), type: z.enum([article, blog, social]).default(article), wordCount: z.number(), complexity: z.enum([simple, moderate, complex]), category: z.enum([short, medium, long]), }), execute: async ({ inputData }) { const { content, type } inputData const words content.trim().split(/\s/) const wordCount words.length // Determine category by length let category: short | medium | long short if (wordCount 50) category medium if (wordCount 200) category long // Determine complexity by average word length const avgWordLength words.reduce((sum, word) sum word.length, 0) / wordCount let complexity: simple | moderate | complex simple if (avgWordLength 5) complexity moderate if (avgWordLength 7) complexity complex console.log( Assessment: ${category} content, ${complexity} complexity) return { content, type, wordCount, complexity, category, } }, })這里的路由策略是字數決定長度類別50 詞為 short50–199 為 medium≥200 為 long平均詞長決定復雜程度5 字符為 moderate7 字符為 complex。從源碼實現看createStep內部通過isStepParams分支識別這種帶execute的步驟定義并對其 schema 做標準規范化見 workflow.ts因此inputSchema/outputSchema會在執行前自動完成運行時校驗。第二步兩個分支處理步驟針對短且簡單與其余全部兩類內容準備兩條處理流水線?？焖偬幚聿襟E——服務于 short simple 內容輸出最小化建議const quickProcessingStep createStep({ id: quick-processing, description: Quick processing for short and simple content, inputSchema: z.object({ content: z.string(), type: z.enum([article, blog, social]).default(article), wordCount: z.number(), complexity: z.enum([simple, moderate, complex]), category: z.enum([short, medium, long]), }), outputSchema: z.object({ processedContent: z.string(), processingType: z.string(), recommendations: z.array(z.string()), }), execute: async ({ inputData }) { console.log(? Quick processing for short and simple content...) return { processedContent: inputData.content, processingType: quick, recommendations: [Content is concise, Consider expanding for more detail], } }, })通用處理步驟——承接非 short/simple 內容模擬更重的處理流程此處以 500ms 延遲示意昂貴操作const generalProcessingStep createStep({ id: general-processing, description: General processing for all other content, inputSchema: z.object({ content: z.string(), type: z.enum([article, blog, social]).default(article), wordCount: z.number(), complexity: z.enum([simple, moderate, complex]), category: z.enum([short, medium, long]), }), outputSchema: z.object({ processedContent: z.string(), processingType: z.string(), recommendations: z.array(z.string()), }), execute: async ({ inputData }) { console.log( General processing for non-short/simple content...) // Simulate more involved processing await new Promise(resolve setTimeout(resolve, 500)) return { processedContent: inputData.content, processingType: general, recommendations: [ Consider simplifying content, Break up long paragraphs, Add examples or explanations if needed, ], } }, })注意兩個步驟的輸出 schema 完全一致processedContent/processingType/recommendations這樣無論路由到哪條分支下游代碼都能以統一結構消費結果——這是設計分支步驟時值得借鑒的約定。第三步用 .branch() 組裝條件工作流現在把評估步驟與兩個處理步驟組合為完整的條件工作流export const conditionalWorkflow createWorkflow({ id: conditional-workflow, description: Content processing with conditional branching, inputSchema: z.object({ content: z.string(), type: z.enum([article, blog, social]).default(article), }), outputSchema: z.object({ processedContent: z.string(), processingType: z.string(), recommendations: z.array(z.string()), }), }) .then(assessContentStep) .branch([ // Branch 1: Short and simple content [async ({ inputData }) inputData.category short inputData.complexity simple, quickProcessingStep], // Branch 2: Everything else [ async ({ inputData }) !(inputData.category short inputData.complexity simple), generalProcessingStep, ], ]) .commit()這里的流水線邏輯清晰可讀then()先行評估 →branch()依據評估結果分叉 →commit()收尾定型。分支條件中的inputData指向上一步驟assessContentStep的輸出這正是評估結果驅動路由的關鍵。深入 .branch() API元組數組與兩種條件形式從源碼看.branch()接收的是一個二元組數組每個元組為[條件, 步驟]branchTBranchSteps extends Array [ConditionFunction... | { predicate: Predicate }, Step...] (steps: TBranchSteps, options?: StepFlowEntryOptions)見 workflow.ts這意味著條件位置有兩種等價寫法閉包函數本教程主線async ({ inputData }) boolean靈活、可直接書寫任意判斷邏輯聲明式 predicate 對象{ predicate: { op: gt, left: { path: inputData.value }, right: { literal: 10 } } }可序列化、便于可視化與持久化。在.branch()內部閉包形式會被直接使用其序列化標簽為condition.toString()而 predicate 形式會被轉換成等價的運行時條件函數并生成可讀標簽。兩者還可以在同一個branch()數組中混用——packages/core/src/workflows/__tests__/predicate-builder.test.ts中的測試用例驗證了這一行為見 predicate-builder.test.ts。不過該測試也指出含閉包條件的分支圖不可整體序列化持久化會拋出 closure predicates do not round-trip 錯誤需要跨請求恢復工作流時請優先使用聲明式 predicate。條件組合、|| 與 !單條條件往往不夠Mastra 的條件函數是標準 JS 表達式可以自由組合邏輯運算符運算符語義示例與——兩者都為真才命中category short complexity simple\|\|或——任一為真即命中category long \|\| complexity complex!非——條件必須為假才命中!(category short complexity simple)以本文為例Short Simplecategory short complexity simple→ 走快速處理建議最少Everything Else!(category short complexity simple)→ 走通用處理給出更多優化建議。兩條條件互為補集保證任何輸入都恰好命中一條分支是互斥路由的經典寫法。條件求值語義順序檢查、并行執行、無匹配跳過理解.branch()的運行時語義對寫出正確的工作流至關重要按順序檢查條件按照在數組中的書寫順序依次求值多條件可同時命中與 if/else 不同分支之間不是互斥關系——若多個條件同時為真對應步驟會并行執行無匹配則跳過若所有條件都不為真工作流不執行任何分支步驟直接繼續后續流程。第 2 點多分支并行有明確的測試佐證packages/core/src/workflows/evented/evented-workflow.test.ts中有一條用例讓兩個條件恒為true的步驟同時進入分支最終兩個步驟的狀態更新被正確合并{ first: 1, second: 1 }說明引擎對命中的分支步驟是并發調度、隨后合并狀態的見 evented-workflow.test.ts。因此如果你需要二選一的嚴格互斥路由務必像本文這樣把條件寫成互補形式而不是依賴分支之間的天然排他。注冊工作流并在 Playground 中驗證構建完成后把新工作流注冊進 Mastra 實例通常在src/mastra/index.tsimport { contentWorkflow, aiContentWorkflow, parallelAnalysisWorkflow, conditionalWorkflow, } from ./workflows/content-workflow export const mastra new Mastra({ workflows: { contentWorkflow, aiContentWorkflow, parallelAnalysisWorkflow, conditionalWorkflow, // Add the conditional workflow }, // ... rest of configuration })隨后在 Mastra Playground 中打開conditional-workflow分別用不同長度與類型的內容測試短內容如 20 詞、平均詞長 6應命中quick-processingprocessingType為quick推薦 2 條建議長內容如 300 詞或高復雜詞應命中general-processingprocessingType為general推薦 3 條建議。完整測試指引可參考課程下一節 測試條件邏輯。整個流程的運轉路徑為評估步驟分析內容 → 分支條件對照評估結果 → 命中步驟執行 → 輸出攜帶processingType標記實際走的分支路徑據此即可直觀確認路由是否正確。調試條件分支的實用技巧如果發現內容沒有進入預期分支按以下順序排查檢查評估步驟輸出先確認category/complexity是否符合預期——路由判據錯分支必錯核對條件邏輯把條件表達式與評估輸出的取值逐一代入驗證布爾結果單獨隔離測試將單個條件抽出來獨立求值排除多條件組合的干擾加日志追蹤在條件函數或步驟execute中增加console.log觀察求值順序與命中情況。小結至此你已掌握 Mastra 條件工作流的完整構建鏈路評估步驟產出路由判據 →.branch()依據inputData分叉 → 互補條件實現互斥路由 → 統一輸出結構消費分支結果。條件分支讓工作流從固定流水線進化為智能路由是構建內容處理、審核分流、異常兜底等場景的基礎能力。下一步可以繼續學習工作流結果流式輸出進一步提升用戶體驗?！久赓M下載鏈接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.項目地址: https://gitcode.com/GitHub_Trending/ma/mastra創作聲明：本文部分內容由AI輔助生成（AIGC），僅供參考