MAF快速入门（12）主工作流+子工作流

目录简介1 子工作流模式介绍2 主工作流子工作流实验案例2.1 关键依赖包引入2.2 定义数据传输模型2.3 定义产品质量处理子工作流2.4 定义物流问题处理子工作流2.5 构建主工作流2.6 测试工作流3 小结4 示例源码简介大家好我是Edison。上一篇我们学习了MAF中如何进行并行工作流。本篇我们来了解下在MAF中如何开启 主工作流 自工作流 的模式相信这会是大家可能会碰到的场景。1 子工作流模式介绍在实际业务场景中往往需要在主工作流中根据工单类型分发或移交到不同的标准化子流程中进行后续处理。在MAF中我们可以使用 Sub-Worklfow 模式来实现这个目的如下代码片段所示。首先将子工作流转换为一个Executorvar logisticsSubExecutor logisticsSubWorkflow.BindAsExecutor(LogisticsSubWorkflow); var productQualitySubExecutor productQualitySubWorkflow.BindAsExecutor(ProductQualitySubWorkflow);然后构建主工作流并根据条件路由到子工作流Executorvar mainWorkflow new WorkflowBuilder(classifierExecutor) .AddEdgeComplaintProcessingRecord(classifierExecutor, productQualitySubExecutor, condition: record record.Category 产品质量) .AddEdgeComplaintProcessingRecord(classifierExecutor, logisticsSubExecutor, condition: record record.Category 物流问题) .AddEdge(productQualitySubExecutor, complianceExecutor) .AddEdge(logisticsSubExecutor, complianceExecutor) .AddEdge(complianceExecutor, sentimentExecutor) .WithOutputFrom(sentimentExecutor) .Build();这样就实现了主工作流进行分类转发自工作流进行详细处理。2 主工作流子工作流实验案例假设我们是一个企业的客服中心每天都会收到大量的工单需要根据工单类型比如是 产品质量话题 还是 物流话题分发到不同的标准化子流程去做后续处理最后生成一个用户友好 且 满足合规要求 的回复给用户。因此我们的目标是配置 1个主工作流 2个子工作流 来覆盖这个需求。2.1 关键依赖包引入在今天的这个案例中我们仍然创建了一个.NET控制台应用程序安装了以下NuGet包Microsoft.Agents.AI.OpenAIMicrosoft.Agents.AI.WorkflowsMicrosoft.Extensions.AI.OpenAI2.2 定义数据传输模型首先我们定义一下在这个工作流中需要生成传递的数据模型投诉数据模型 及 共享状态namespace SubWorkflow.Models; // 共享状态投诉处理记录各执行器会更新此对象 internal class ComplaintProcessingRecord { public CustomerComplaint Original { get; set; } public string Category { get; set; } 未分类; public string Handler { get; set; } 待分配; public Liststring ProcessingSteps { get; set; } new(); public string AIGeneratedResponse { get; set; } ; public string ComplianceStatus { get; set; } 待审核; public string SentimentScore { get; set; } 未评估; } // 投诉数据模型 internal record CustomerComplaint( string OrderId, string CustomerName, string ComplaintText, DateTime SubmittedAt );2.3 定义产品质量处理子工作流1问题评估封装产品质量评估的逻辑这里仅仅做演示用无实际逻辑internal sealed class ProductEvaluationExecutor() : ExecutorComplaintProcessingRecord, ComplaintProcessingRecord(nameof(ProductEvaluationExecutor)) { public override ValueTaskComplaintProcessingRecord HandleAsync(ComplaintProcessingRecord record, IWorkflowContext context, CancellationToken cancellationToken default) { Console.WriteLine( 正在评估产品质量问题...); record.ProcessingSteps.Add([产品评估] 检测到屏幕外观缺陷符合质量问题定义); record.Handler 产品质量团队; return ValueTask.FromResult(record); } }2退/换货判定负责判定是退货 还是 换货 处理。​​​​​​​internal sealed class ReturnPolicyExecutor() : ExecutorComplaintProcessingRecord, ComplaintProcessingRecord(nameof(ReturnPolicyExecutor)) { public override ValueTaskComplaintProcessingRecord HandleAsync(ComplaintProcessingRecord record, IWorkflowContext context, CancellationToken cancellationToken default) { Console.WriteLine( 判定退换货政策...); var daysFromOrder (DateTime.Now - record.Original.SubmittedAt).Days; if (daysFromOrder 7) { record.ProcessingSteps.Add([退换货判定] 符合7天无理由退货政策批准全额退款); } else { record.ProcessingSteps.Add([退换货判定] 超过退货期限建议换货或部分补偿); } return ValueTask.FromResult(record); } }3AI回复在投诉内容和判定结果的基础上生成友好的回复返回给用户。​​​​​​​internal sealed class AIResponseGeneratorExecutor(IChatClient chatClient) : ExecutorComplaintProcessingRecord, ComplaintProcessingRecord(nameof(AIResponseGeneratorExecutor)) { public override async ValueTaskComplaintProcessingRecord HandleAsync(ComplaintProcessingRecord record, IWorkflowContext context, CancellationToken cancellationToken default) { Console.WriteLine( AI 正在生成客户回复...); var prompt $你是专业的客服主管。根据以下投诉处理信息生成一封正式、有同理心的客户回复邮件150字内 客户{record.Original.CustomerName} 订单号{record.Original.OrderId} 投诉内容{record.Original.ComplaintText} 处理步骤 {string.Join(\n, record.ProcessingSteps)} 要求 1. 表达歉意和理解 2. 说明处理方案 3. 提供后续联系方式 4. 语气真诚、专业; var response await chatClient.GetResponseAsync(prompt, cancellationToken: cancellationToken); record.AIGeneratedResponse response.Text ?? AI 生成失败; record.ProcessingSteps.Add($[AI 回复] 已生成客户回复模板{record.AIGeneratedResponse.Length}字); return record; } }最后构建子工作流并将其转换为一个Executor以便后续集成到主工作流中​​​​​​​var productEvalExecutor new ProductEvaluationExecutor(); var returnPolicyExecutor new ReturnPolicyExecutor(); var aiResponseExecutor new AIResponseGeneratorExecutor(chatClient); var productQualitySubWorkflow new WorkflowBuilder(productEvalExecutor) .AddEdge(productEvalExecutor, returnPolicyExecutor) .AddEdge(returnPolicyExecutor, aiResponseExecutor) .WithOutputFrom(aiResponseExecutor) .Build(); var productQualitySubExecutor productQualitySubWorkflow.BindAsExecutor(ProductQualitySubWorkflow);2.4 定义物流问题处理子工作流1物流追踪封装物流查询逻辑这里仅仅做演示用无实际逻辑​​​​​​​internal sealed class LogisticsTrackingExecutor() : ExecutorComplaintProcessingRecord, ComplaintProcessingRecord(nameof(LogisticsTrackingExecutor)) { public override ValueTaskComplaintProcessingRecord HandleAsync(ComplaintProcessingRecord record, IWorkflowContext context, CancellationToken cancellationToken default) { Console.WriteLine( 正在查询物流信息...); record.ProcessingSteps.Add([物流追踪] 包裹在中转站滞留3天当前状态运输中); record.Handler 物流运营团队; return ValueTask.FromResult(record); } }2延迟分析封装延迟分析逻辑这里仅仅做演示用无实际逻辑​​​​​​​internal sealed class DelayAnalysisExecutor() : ExecutorComplaintProcessingRecord, ComplaintProcessingRecord(nameof(DelayAnalysisExecutor)) { public override ValueTaskComplaintProcessingRecord HandleAsync(ComplaintProcessingRecord record, IWorkflowContext context, CancellationToken cancellationToken default) { Console.WriteLine( 分析延迟原因...); record.ProcessingSteps.Add([延迟分析] 延迟原因暴雨导致道路封闭预计2天内恢复配送); record.ProcessingSteps.Add([补偿方案] 提供50元优惠券 免运费); return ValueTask.FromResult(record); } }最后构建子工作流并将其转换为一个Executor以便后续集成到主工作流中​​​​​​​var logisticsTrackExecutor new LogisticsTrackingExecutor(); var delayAnalysisExecutor new DelayAnalysisExecutor(); var logisticsAIResponseExecutor new AIResponseGeneratorExecutor(chatClient); var logisticsSubWorkflow new WorkflowBuilder(logisticsTrackExecutor) .AddEdge(logisticsTrackExecutor, delayAnalysisExecutor) .AddEdge(delayAnalysisExecutor, logisticsAIResponseExecutor) .WithOutputFrom(logisticsAIResponseExecutor) .Build(); var logisticsSubExecutor logisticsSubWorkflow.BindAsExecutor(LogisticsSubWorkflow);2.5 构建主工作流现在开始构建主工作流Step1: 获取ChatClient​​​​​​​var chatClient new OpenAIClient( new ApiKeyCredential(openAIProvider.ApiKey), new OpenAIClientOptions { Endpoint new Uri(openAIProvider.Endpoint) }) .GetChatClient(openAIProvider.ModelId) .AsIChatClient();Step2: 创建工单分类Executor,合规审核Executor 以及 情绪评估ExecutorNOTE前面的子工作流均生成了AI回复这里需要对其进行合规审核以防止不合规的回复到用户那里。​​​​​​​// 1. 工单分类执行器 internal sealed class ComplaintClassifierExecutor() : ExecutorComplaintProcessingRecord, ComplaintProcessingRecord(nameof(ComplaintClassifierExecutor)) { public override ValueTaskComplaintProcessingRecord HandleAsync(ComplaintProcessingRecord record, IWorkflowContext context, CancellationToken cancellationToken default) { Console.WriteLine(️ 正在分类投诉类型...); // 简单规则分类实际可用 AI 分类 if (record.Original.ComplaintText.Contains(划痕) || record.Original.ComplaintText.Contains(质量) || record.Original.ComplaintText.Contains(退货)) { record.Category 产品质量; } else if (record.Original.ComplaintText.Contains(延迟) || record.Original.ComplaintText.Contains(物流) || record.Original.ComplaintText.Contains(配送)) { record.Category 物流问题; } else { record.Category 其他; } record.ProcessingSteps.Add($[分类器] 投诉类型识别为{record.Category}); return ValueTask.FromResult(record); } } // 2. 合规审核执行器 internal sealed class ComplianceCheckExecutor() : ExecutorComplaintProcessingRecord, ComplaintProcessingRecord(nameof(ComplianceCheckExecutor)) { public override ValueTaskComplaintProcessingRecord HandleAsync(ComplaintProcessingRecord record, IWorkflowContext context, CancellationToken cancellationToken default) { Console.WriteLine(️ 合规审核中...); var checks new Liststring(); if (record.AIGeneratedResponse.Contains(歉意)) checks.Add(✅ 包含道歉); if (record.AIGeneratedResponse.Length 300) checks.Add(✅ 字数合规); if (!record.AIGeneratedResponse.Contains(法律) !record.AIGeneratedResponse.Contains(诉讼)) checks.Add(✅ 无敏感法律词汇); record.ComplianceStatus checks.Count 2 ? ✅ 审核通过 : ⚠️ 需人工复审; record.ProcessingSteps.Add($[合规审核] {record.ComplianceStatus} - {string.Join(, , checks)}); return ValueTask.FromResult(record); } } // 3. 情绪评估执行器使用 AI internal sealed class SentimentAnalysisExecutor(IChatClient chatClient) : ExecutorComplaintProcessingRecord, ComplaintProcessingRecord(nameof(SentimentAnalysisExecutor)) { public override async ValueTaskComplaintProcessingRecord HandleAsync(ComplaintProcessingRecord record, IWorkflowContext context, CancellationToken cancellationToken default) { Console.WriteLine( AI 情绪评估中...); var prompt $分析以下客服回复的情绪基调用一个词概括如友好、专业、冷淡、热情 {record.AIGeneratedResponse} 只返回一个词不要解释。; var response await chatClient.GetResponseAsync(prompt, cancellationToken: cancellationToken); record.SentimentScore response.Text?.Trim() ?? 中性; record.ProcessingSteps.Add($[情绪评估] AI 评估语气为{record.SentimentScore}); return record; } }Step3: 创建主工作流​​​​​​​var classifierExecutor new ComplaintClassifierExecutor(); var complianceExecutor new ComplianceCheckExecutor(); var sentimentExecutor new SentimentAnalysisExecutor(chatClient); // 主工作流分类 → 条件路由到子流程 → 合规 → 情绪 var mainWorkflow new WorkflowBuilder(classifierExecutor) .AddEdgeComplaintProcessingRecord(classifierExecutor, productQualitySubExecutor, condition: record record.Category 产品质量) .AddEdgeComplaintProcessingRecord(classifierExecutor, logisticsSubExecutor, condition: record record.Category 物流问题) .AddEdge(productQualitySubExecutor, complianceExecutor) .AddEdge(logisticsSubExecutor, complianceExecutor) .AddEdge(complianceExecutor, sentimentExecutor) .WithOutputFrom(sentimentExecutor) .Build(); Console.OutputEncoding Encoding.UTF8; Console.WriteLine(✅ Main Sub Workflow 构建完成);2.6 测试工作流定义投诉的工单内容如下这是一个产品质量相关的投诉。​​​​​​​var processingRecord new ComplaintProcessingRecord { Original new CustomerComplaint( OrderId: ORD-2025-8821, CustomerName: 张先生, ComplaintText: 收到的手机屏幕有明显划痕要求退货退款, SubmittedAt: DateTime.Now) };通过Streaming流式执行​​​​​​​Console.WriteLine( 开始执行投诉处理流水线...); Console.WriteLine(new string(━, 60)); await using (var streaming await InProcessExecution.StreamAsync(mainWorkflow, processingRecord)) { await foreach (WorkflowEvent evt in streaming.WatchStreamAsync()) { switch (evt) { case ExecutorInvokedEvent started: Console.WriteLine($\n {started.ExecutorId} 开始执行); break; case ExecutorCompletedEvent completed when completed.Data is ComplaintProcessingRecord rec: Console.WriteLine($ 共享状态更新 → 处理步骤数{rec.ProcessingSteps.Count}); break; case WorkflowOutputEvent outputEvt when outputEvt.Data is ComplaintProcessingRecord finalRecord: Console.WriteLine(\n new string(━, 60)); Console.WriteLine( 投诉处理完成最终处理记录:\n); Console.WriteLine(JsonSerializer.Serialize(finalRecord)); Console.WriteLine(\n 处理步骤详情); foreach (var step in finalRecord.ProcessingSteps) { Console.WriteLine($ {step}); } Console.WriteLine(\n AI 生成的客户回复); Console.WriteLine(new string(─, 60)); Console.WriteLine(finalRecord.AIGeneratedResponse); Console.WriteLine(new string(─, 60)); break; } } }测试结果如下图所示首先可以通过监控日志发现准确地进入到了产品质量子流程其次最后由AI生成的回复内容也很友好同时也满足合规审核的要求3 小结本文介绍了MAF中主工作流 子工作流的工作模式最后通过一个企业客服中心处理投诉工单的案例介绍了这种模式的代码实现。需要注意的是建议每个子工作流都是用标准统一的状态模型以便在主工作流中复用 和 传递。建议每个子工作流只保持3~4个步骤单一职责在工作流中同样适用下一篇​​​​​​​我们将继续学习MAF中智能体的几种常见编排模式 - 顺序并发 和 移交。4 示例源码Github: https://github.com/EdisonTalk/MAFDhttps://github.com/mzhongl524/MAFD2026-8月更新示例引入地址