 构建智能内容路由)
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会在执行前自动完成运行时校验。第二步两个分支处理步骤针对短且简单与其余全部两类内容准备两条处理流水线。快速处理步骤——服务于 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分叉 → 互补条件实现互斥路由 → 统一输出结构消费分支结果。条件分支让工作流从固定流水线进化为智能路由是构建内容处理、审核分流、异常兜底等场景的基础能力。下一步可以继续学习工作流结果流式输出进一步提升用户体验。【免费下载链接】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会在执行前自动完成运行时校验。第二步两个分支处理步骤针对短且简单与其余全部两类内容准备两条处理流水线。快速处理步骤——服务于 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分叉 → 互补条件实现互斥路由 → 统一输出结构消费分支结果。条件分支让工作流从固定流水线进化为智能路由是构建内容处理、审核分流、异常兜底等场景的基础能力。下一步可以继续学习工作流结果流式输出进一步提升用户体验。【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考