破解企业级 AI 复杂性：用 Microsoft Agent Framework 实现多 Agent 编排
架构刚需：为什么多 Agent 编排不可或缺
在现代企业 AI 系统中，真实业务问题的范围与复杂度很快就会超出单体 AI Agent 的能力边界。面对端到端客户旅程管理、多源数据治理、深度 Human-in-the-Loop 审核流程等任务，核心架构问题已经转变为：我们如何高效协调并管理由多个原子化 AI 能力构成的网络？
就像高绩效企业依赖不同专业部门协同一样，我们也需要从“单执行者”模型，升级到协作式多 Agent 网络（Collaborative Multi-Agent Network）。
Microsoft Agent Framework 正是为这一范式转变而设计：它提供统一且可观测的平台，帮助开发者实现两项核心价值。
场景 1：构建专业化 AI Agent 单元
每个 Agent 都是一个专业化、可插拔、可独立运行的执行单元，其智能能力建立在三大关键支柱之上：
- LLM 驱动的意图解析： 借助 Large Language Models（LLMs）能力，准确理解并映射复杂用户输入需求。
- 动作与工具执行： 通过调用外部 API、工具或内部服务（如 MCP servers）执行真实业务逻辑与操作。
- 上下文响应生成： 基于执行结果与当前状态，向用户返回精准、有价值且具上下文感知能力的智能响应。
开发者可灵活选择主流模型提供方（包括 Azure OpenAI、OpenAI、Azure AI Foundry 或本地模型），定制并构建高性能 Agent 原语。
场景 2：通过 Workflow 编排实现动态协同
Workflow 是 Microsoft Agent Framework 的旗舰能力，它将编排从简单线性流程提升为动态协作图，赋予系统高级架构能力：
- 🔗 构建协作图谱： 将专业 Agent 与功能模块连接成高内聚、低耦合网络。
- 🎯 拆解复杂任务： 自动把宏任务分解为可管理、可追踪的子任务步骤并精确执行。
- 🧭 基于上下文的动态路由： 利用中间数据类型与业务规则自动选择最优处理路径或 Agent（Routing）。
- 🔄 支持深度嵌套： 在主 Workflow 中嵌入子 Workflow，实现分层逻辑抽象并最大化复用。
- 💾 定义检查点： 在关键执行节点持久化状态，确保流程可追踪性、数据可验证性与容错能力。
- 🤝 集成 Human-in-the-Loop： 通过清晰的请求/响应契约，在必要时将人工专家引入决策闭环。
更关键的是，Workflow 定义并不局限于 Agent 连接，还可与现有业务逻辑和方法执行器无缝集成，为复杂流程集成提供最大灵活性。
深入解析：Workflow 模式
基于 GitHub Models 示例，我们演示如何利用 Workflow 组件在企业应用中落地结构化流程、并行处理与动态决策。
1. Sequential：强制结构化数据流
- 定义： 执行器按预定义顺序运行，每一步输出都会被校验、序列化，并作为标准化输入传给链路中的下一执行器。
- 架构意义： 该模式对需要严格幂等性（strict idempotency）与阶段间状态管理的流水线至关重要。应在中间节点策略性使用Transformer Executors（如
to_reviewer_result）进行数据格式化、校验或状态日志记录，以建立关键检查点。
# Linear flow: Agent1 -> Agent2 -> Agent3
workflow = (
WorkflowBuilder()
.set_start_executor(agent1)
.add_edge(agent1, agent2)
.add_edge(agent2, agent3)
.build()
)
2. Concurrent：实现高吞吐 Fan-out/Fan-in
- 定义： 在同一 Workflow 中并发启动多个 Agent（或同一 Agent 的多个实例）以降低总体延迟，并在指定的Join Point合并结果。
- 架构意义： 这是Fan-out/Fan-in模式的核心实现。关键组件是Aggregation Function（
aggregate_results_function），需要在其中实现自定义逻辑来汇总多分支返回，常见方式包括投票机制、加权合并或基于优先级的选择。
workflow = (
ConcurrentBuilder()
.participants([agentA, agentB, agentC])
.build()
)
3. Conditional：基于状态的动态决策
- 定义： Workflow 引入决策执行器，根据中间结果或预定义业务规则将流程动态路由到不同分支（如 Save Draft、Rework、Human Review）。
- 架构意义： 该模式的核心在于selection function（
selection_func）。它接收解析后的中间数据（例如ReviewResult），并返回目标执行器 ID 列表，因此不仅支持单路径路由，还支持“一条数据进入多条并行路径”的复杂逻辑。
def select_targets(review, targets):
handle_id, save_id = targets
return [save_id] if review.review_result == "Yes" else [handle_id]
workflow = (
WorkflowBuilder()
.set_start_executor(evangelist_executor)
.add_edge(evangelist_executor, reviewer_executor)
.add_edge(reviewer_executor, to_reviewer_result)
.add_multi_selection_edge_group(to_reviewer_result, [handle_review, save_draft], selection_func=select_targets)
.build()
)
在复杂生产场景中，这些模式通常会叠加使用：例如先进行并发搜索与摘要（Concurrent），再进入条件分支（Conditional），将结果路由到自动发布或顺序化 Human-in-the-Loop 审核流程（Sequential）。
生产级可观测性：用好 DevUI 与 Tracing
对于复杂多 Agent 系统，**可观测性（Observability）**是不可妥协的能力。Microsoft Agent Framework 通过内置 DevUI 提供出色开发体验，可对编排层进行实时可视化、交互追踪与性能监控。
以下精简代码展示了在项目中启用该能力的关键步骤（参见项目 main.py）：
- 核心 Workflow 构建（代码不变）
# Transform and selection function example
@executor(id="to_reviewer_result")
async def to_reviewer_result(response, ctx):
parsed = ReviewAgent.model_validate_json(response.agent_run_response.text)
await ctx.send_message(ReviewResult(parsed.review_result, parsed.reason, parsed.draft_content))
def select_targets(review: ReviewResult, target_ids: list[str]) -> list[str]:
handle_id, save_id = target_ids
return [save_id] if review.review_result == "Yes" else [handle_id]
# Build executors and connect them
evangelist_executor = AgentExecutor(evangelist_agent, id="evangelist_agent")
reviewer_executor = AgentExecutor(reviewer_agent, id="reviewer_agent")
publisher_executor = AgentExecutor(publisher_agent, id="publisher_agent")
workflow = (
WorkflowBuilder()
.set_start_executor(evangelist_executor)
.add_edge(evangelist_executor, to_evangelist_content_result)
.add_edge(to_evangelist_content_result, reviewer_executor)
.add_edge(reviewer_executor, to_reviewer_result)
.add_multi_selection_edge_group(to_reviewer_result, [handle_review, save_draft], selection_func=select_targets)
.add_edge(save_draft, publisher_executor)
.build()
)
- 结合 DevUI 启动可视化（项目
main.py）
from agent_framework.devui import serve
def main():
serve(entities=[workflow], port=8090, auto_open=True, tracing_enabled=True)
if __name__ == "__main__":
main()
实现端到端 Tracing
当多 Agent Workflow 部署到生产或 CI 环境后，健壮的 tracing 与监控是刚需。要确保高可观测性，你需要确认以下事项：
- 环境配置： 启动前通过
.env加载 Agent 与工具所需的全部连接字符串和凭证。 - 事件日志： 在 Agent Executors 与 Transformers 内使用框架上下文机制，显式记录关键事件（如 Agent 响应、分支选择结果），便于 DevUI 或日志聚合平台检索。
- OTLP 集成： 将
tracing_enabled设为True，并配置 OpenTelemetry Protocol (OTLP) exporter。这样可将完整执行调用链（Trace）导出到 APM/Trace 平台（如 Azure Monitor、Jaeger）。 - 示例代码：https://github.com/microsoft/Agent-Framework-Samples/tree/main/08.EvaluationAndTracing/python/multi_workflow_aifoundry_devui
将 DevUI 的可视执行路径与 APM trace 数据结合后，你可以快速定位延迟瓶颈、精准诊断故障，并对复杂 AI 系统保持全局可控。
下一步：面向 Agent 架构师的学习资源
多 Agent 编排代表了复杂 AI 架构的未来。建议你继续深入 Microsoft Agent Framework，系统掌握这些关键能力。
下面是一组精选资源，帮助你加速成长为 Agent Architect：
-
Microsoft Agent Framework GitHub 仓库： https://github.com/microsoft/agent-framework
-
Microsoft Agent Framework Workflow 官方示例： https://github.com/microsoft/agent-framework/tree/main/python/samples/getting_started/workflows
分类
作者
高级云倡导者
Kinfey Lo 是微软高级云倡导者，专注于 Edge AI 生态中 Small Language Models（SLMs）的开发与工程化落地。他是 “Phi Cookbook” 的作者，该资源面向 Phi 系列 SLM 的实践应用。其专长在于构建适配 Edge AI 独特需求的 GenAIOps 策略。
