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 模式来实现这个目的，如下代码片段所示。

首先，将子工作流转换为一个Executor：

var logisticsSubExecutor = logisticsSubWorkflow.BindAsExecutor("LogisticsSubWorkflow"); var productQualitySubExecutor = productQualitySubWorkflow.BindAsExecutor("ProductQualitySubWorkflow");

然后，构建主工作流并根据条件路由到子工作流Executor：

var mainWorkflow = new WorkflowBuilder(classifierExecutor) .AddEdge<ComplaintProcessingRecord>(classifierExecutor, productQualitySubExecutor, condition: record => record.Category == "产品质量") .AddEdge<ComplaintProcessingRecord>(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.OpenAI
Microsoft.Agents.AI.Workflows
Microsoft.Extensions.AI.OpenAI

2.2 定义数据传输模型

首先，我们定义一下在这个工作流中需要生成传递的数据模型：

投诉数据模型及共享状态

namespace SubWorkflow.Models; // 共享状态：投诉处理记录（各执行器会更新此对象） internal class ComplaintProcessingRecord { public CustomerComplaint Original { get; set; } public string Category { get; set; } = "未分类"; public string Handler { get; set; } = "待分配"; public List<string> 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() : Executor<ComplaintProcessingRecord, ComplaintProcessingRecord>(nameof(ProductEvaluationExecutor)) { public override ValueTask<ComplaintProcessingRecord> HandleAsync(ComplaintProcessingRecord record, IWorkflowContext context, CancellationToken cancellationToken = default) { Console.WriteLine("🔍 正在评估产品质量问题..."); record.ProcessingSteps.Add("[产品评估] 检测到屏幕外观缺陷，符合质量问题定义"); record.Handler = "产品质量团队"; return ValueTask.FromResult(record); } }

（2）退/换货判定：负责判定是退货还是换货处理。

internal sealed class ReturnPolicyExecutor() : Executor<ComplaintProcessingRecord, ComplaintProcessingRecord>(nameof(ReturnPolicyExecutor)) { public override ValueTask<ComplaintProcessingRecord> 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); } }

（3）AI回复：在投诉内容和判定结果的基础上生成友好的回复返回给用户。

internal sealed class AIResponseGeneratorExecutor(IChatClient chatClient) : Executor<ComplaintProcessingRecord, ComplaintProcessingRecord>(nameof(AIResponseGeneratorExecutor)) { public override async ValueTask<ComplaintProcessingRecord> 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() : Executor<ComplaintProcessingRecord, ComplaintProcessingRecord>(nameof(LogisticsTrackingExecutor)) { public override ValueTask<ComplaintProcessingRecord> 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() : Executor<ComplaintProcessingRecord, ComplaintProcessingRecord>(nameof(DelayAnalysisExecutor)) { public override ValueTask<ComplaintProcessingRecord> 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 以及情绪评估Executor

NOTE：前面的子工作流均生成了AI回复，这里需要对其进行合规审核以防止不合规的回复到用户那里。

// 1. 工单分类执行器 internal sealed class ComplaintClassifierExecutor() : Executor<ComplaintProcessingRecord, ComplaintProcessingRecord>(nameof(ComplaintClassifierExecutor)) { public override ValueTask<ComplaintProcessingRecord> 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() : Executor<ComplaintProcessingRecord, ComplaintProcessingRecord>(nameof(ComplianceCheckExecutor)) { public override ValueTask<ComplaintProcessingRecord> HandleAsync(ComplaintProcessingRecord record, IWorkflowContext context, CancellationToken cancellationToken = default) { Console.WriteLine("🛡️ 合规审核中..."); var checks = new List<string>(); 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) : Executor<ComplaintProcessingRecord, ComplaintProcessingRecord>(nameof(SentimentAnalysisExecutor)) { public override async ValueTask<ComplaintProcessingRecord> 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) .AddEdge<ComplaintProcessingRecord>(classifierExecutor, productQualitySubExecutor, condition: record => record.Category == "产品质量") .AddEdge<ComplaintProcessingRecord>(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生成的回复内容也很友好，同时也满足合规审核的要求：