Microsoft Agent Framework 构建 SubAgent（Multi-Agent）

本文演示如何用 Microsoft Agent Framework 用 Executor WorkflowDAG模式实现 SubAgent子代理架构。通过示例代码来自项目的 txt展示并发 Fan‑Out/Fan‑In 的实现、消息路由与聚合策略最后讨论最佳实践与权衡。假定读者等级:熟悉 C#、异步编程与基本的 LLM/Agent 概念如果不熟先看基础 LLM/Agent 入门。为什么要用 SubAgents将复杂任务拆成职责单一的模块职责分离更易观测、测试与演进可插拔节点支持并行/流水线处理便于扩展与优化核心思路简要Planner/Orchestrator 负责任务拆解与路由每个 SubAgent 为专责 Executor或 Tool用 Workflow/DAG 串联节点支持 Fan‑Out并发与 Fan‑In聚合明确上下文与记忆隔离防止“串味”示例架构节点ConcurrentStartExecutor广播、ChatClientAgent专业子代理、ConcurrentAggregationExecutor聚合流程Start → Fan‑Out 到多个 Agent → Agent 各自处理 → Fan‑In 到 Aggregation → 输出结果示例代码初始化 OpenAI client 两个 ChatClientAgentOpenAIClient client new OpenAIClient(Environment.GetEnvironmentVariable(OPENAI_API_KEY)!); var chatClient client.GetChatClient(gpt-4o-mini).AsIChatClient(); ChatClientAgent physicist new( chatClient, name: Physicist, instructions: You are an expert in physics. You answer questions from a physics perspective. ); ChatClientAgent chemist new( chatClient, name: Chemist, instructions: You are an expert in chemistry. You answer questions from a chemistry perspective. );构建 Workflowvar startExecutor new ConcurrentStartExecutor(); var aggregationExecutor new ConcurrentAggregationExecutor(); var workflow new WorkflowBuilder(startExecutor) .AddFanOutEdge(startExecutor, new[] { physicist, chemist }) .AddFanInBarrierEdge(new[] { physicist, chemist }, aggregationExecutor) .WithOutputFrom(aggregationExecutor) .Build();流式执行并监听输出await using StreamingRun run await InProcessExecution.RunStreamingAsync(workflow, input: What is temperature?); await foreach (WorkflowEvent evt in run.WatchStreamAsync()) { if (evt is WorkflowOutputEvent output) { Console.WriteLine($Workflow completed with results:\n{output.Data}); } }ConcurrentStartExecutor广播用户消息再发 TurnToken 启动子代理[MessageHandler] public async ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken default) { await context.SendMessageAsync(new ChatMessage(ChatRole.User, message), cancellationToken: cancellationToken); await context.SendMessageAsync(new TurnToken(emitEvents: true), cancellationToken: cancellationToken); } protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) { return protocolBuilder.ConfigureRoutes(r r.AddHandlerstring(this.HandleAsync)) .SendsMessageChatMessage() .SendsMessageTurnToken(); }ConcurrentAggregationExecutor接收各 Agent 消息并在满足条件时产出聚合结果public override async ValueTask HandleAsync(ListChatMessage message, IWorkflowContext context, CancellationToken cancellationToken default) { this._messages.AddRange(message); if (this._messages.Count 3) // 示例等待 3 条回复再聚合 { var formattedMessages string.Join(Environment.NewLine, this._messages.Select(m ${m.AuthorName}: {m.Text})); await context.YieldOutputAsync(formattedMessages, cancellationToken); } }逐步说明快速教程风格准备环境.NET 版本与 Microsoft Agent Framework SDK按项目要求 pin 版本配置设置OPENAI_API_KEY环境变量定义 Agents / Executors每个子代理用一类 Executor 或 ChatClientAgent 实例表示明确每个代理的instructionsprompt与权限可调用哪些工具构建 WorkflowDAG先构建起点 Executorstart再 AddNode / AddEdge 或用 Fan‑Out/Fan‑In helper聚合节点负责合并与输出并可实现冲突解决策略运行与监控支持流式执行StreamingRun便于实时观察中间结果在聚合节点加入超时、重试或部分可用逻辑测试单元测试 Executors 的消息处理逻辑集成测试覆盖完整 Workflow 路径最佳实践与权衡推荐Workflow/DAG 模式可观测、适合生产必须上下文隔离避免 prompt 串味、清晰的路由与错误边界权衡更细颗粒的 SubAgents → 更高可组合性但更复杂的调度/监控Tool‑based 子代理更简单但失去自治能力安全控制外部工具调用权限防止子代理滥用常见坑把所有逻辑塞进单个 Agent难以维护忽视失败场景网络、LLM 超时、部分结果不可用不做上下文边界与隐私隔离导致数据泄露或“串味”结论 下一步SubAgent 模式适合将复杂任务分解为可管理、可观测的模块。对于生产级系统推荐用 Workflow/DAG Executor 模式。参考与进一步阅读Microsoft Agent Framework 文档请参考 SDK 官方文档