本文演示如何用 Microsoft Agent Framework 用 Executor + Workflow（DAG）模式实现 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 & 两个 ChatClientAgent

OpenAIClient 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."
);

构建 Workflow

var 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.AddHandler<string>(this.HandleAsync))
                          .SendsMessage<ChatMessage>()
                          .SendsMessage<TurnToken>();
}

ConcurrentAggregationExecutor（接收各 Agent 消息并在满足条件时产出聚合结果）

public override async ValueTask HandleAsync(List<ChatMessage> 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);
    }
}

逐步说明（快速教程风格）

1. 准备

• 环境：.NET 版本与 Microsoft Agent Framework SDK（按项目要求 pin 版本）

• 配置：设置 OPENAI_API_KEY 环境变量

• 定义 Agents / Executors

• • 每个子代理用一类 Executor 或 ChatClientAgent 实例表示

  • 明确每个代理的 instructions（prompt）与权限（可调用哪些工具）

• 构建 Workflow（DAG）

• • 先构建起点 Executor（start），再 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 官方文档）