跳到主要内容

Microsoft Agent Framework — 架构与原理

30 秒导读: Microsoft Agent Framework(MAF)是微软给「把 AI agent 做到能上生产」用的框架,同时提供 .NET 和 Python 两套一致的 API。它有两个层次:下层是单个 agent(一个会调工具、能被中间件包裹、带会话记忆的 LLM 循环);上层是多 agent 图工作流(把多个 agent/函数连成一张有向图,按数据流一步步跑)。两层站在同一套「类型化消息」抽象上。它最值钱的部分是工作流引擎:用 Pregel 超步(把整张图切成一轮一轮离散的计算步)驱动执行,并在每个超步边界落检查点,于是整条多 agent 流程可以持久化、崩溃后恢复、暂停等人回话、甚至回到某一步重跑。

1. 这是什么(零基础也能懂)​

  • 一句话定义: 一个用来搭建并运行 AI agent、以及把多个 agent 编排成工作流的生产级框架,.NET 与 Python 双语言、API 对齐。

  • 解决什么问题 / 给谁用: 假设你已经能让一个 LLM「聊天 + 调几个工具」,但真要上线时会撞到一堆硬问题——多个 agent 怎么协作、流程跑一半崩了怎么接着跑、要不要人来审一下再继续、换个模型供应商要不要重写。MAF 就是替这些「从原型到生产」的活儿铺底座的,给的是要把 agent 系统运营起来的团队用。

  • 它能做什么(功能):

    • 建单个 agent:接各家模型(Foundry / Azure OpenAI / OpenAI / Anthropic / Gemini / Ollama…),挂工具、挂中间件、带会话记忆。
    • 建多 agent 工作流:把 agent 和普通函数连成图,支持顺序、并发、交接(handoff)、群聊等模式。
    • 生产能力:检查点与断点续跑、流式事件、人在环路(human-in-the-loop)、时间旅行(回到某个检查点重跑)、OpenTelemetry 可观测性、YAML 声明式定义。

  • 用起来什么样: 最小的一个 agent(Python,来自 README):

    # 示意,源自 README 快速上手
    from agent_framework import Agent
    from agent_framework.foundry import FoundryChatClient
    from azure.identity import AzureCliCredential

    agent = Agent(                                  # 一个 agent = 客户端 + 指令 + (工具)
        client=FoundryChatClient(credential=AzureCliCredential()),
        name="HaikuAgent",
        instructions="You are an upbeat assistant that writes beautifully.",
    )
    print(await agent.run("Write a haiku about Microsoft Agent Framework."))

    把两个 agent 串成一条最小工作流也只要一行 builder:

    # 示意,源自 python/samples/03-workflows/_start-here/step2_agents_in_a_workflow.py
    from agent_framework import WorkflowBuilder
    workflow = WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()
    events = await workflow.run("Create a slogan for a new electric SUV.")

  • 一句话直觉/类比: 把它想成 agent 世界的主板 + 流水线控制器。单个 agent 是插在主板上的芯片;工作流引擎是流水线控制器,它不关心每颗芯片内部怎么算,只按「谁产出什么类型的数据 → 该送给谁」把料一节一节往下传,每传完一节存一次档。

2. 顶层全景(它大概怎么转)​

怎么读下面这张图: 从上往下是「你的输入 → 图引擎按超步推进 → 图里每个节点可以是一个 agent」;右侧岔出去的是「需要人回话时,流程停在这里等」。核心记住一件事:图引擎和 agent 是两层,靠「带类型的消息」对接。

              你的应用 / DevUI / 声明式 YAML
                        │  workflow.run(输入)
                        ▼
   ┌──────────────  Workflow 图引擎  ──────────────┐
   │                                                │
   │   Runner:Pregel 超步循环(一轮 = 一个超步)    │
   │     每轮:跑本轮活跃的 Executor,再沿 Edge 把   │
   │           输出「按类型」路由给下游 Executor     │
   │                                                │
   │     超步边界:提交状态 + 落一个 Checkpoint      │
   │              (于是可恢复 / 可回退重跑)         │
   └───────┬───────────────────────────┬────────────┘
           │ 每个 Executor 内部可以是    │ 遇到 request_info
           ▼                            ▼
   ┌─ Agent(run 循环)──────┐     RequestInfo:人在环路
   │  ChatClient ↔  LLM      │     流程挂起,等外部回话后
   │  Tools / Middleware     │     从检查点继续
   │  Session(会话记忆)     │
   └─────────────────────────┘

部件一句话职责:

部件干什么在哪个文件(python)
Agent / BaseAgent单个 agent 的 run 循环:组消息、调 LLM、跑工具、回结果_agents.py
ChatClient(各实现)把统一的消息抽象翻译成某家模型 API 的调用_clients.py + 各 provider 包
工具 / 中间件 / 技能给 agent 加「手脚」:函数工具、请求/响应拦截、领域知识库_tools.py / _middleware.py / _skills.py
Executor + @handler工作流图里的一个节点;按入参类型决定哪个 handler 接_workflows/_executor.py
Edge / EdgeGroup节点间的有向边;可带条件、扇出、扇入、switch-case_workflows/_edge.py
RunnerPregel 超步循环的发动机,驱动整张图跑到收敛_workflows/_runner.py
WorkflowCheckpoint + 存储每个超步边界的完整快照,支持恢复与时间旅行_workflows/_checkpoint.py
编排 BuilderSequential / Concurrent / Handoff / Magentic / GroupChat 高层模式orchestrations/…

主线走一遍(高层、不进代码):

输入 → WorkflowBuilder 把 agent/函数连成图并 build() 出 Workflow
     → Runner 进入第 1 个超步:跑起点 Executor
     → 起点输出一个「带类型的消息」→ 沿匹配的 Edge 路由给下游 Executor
     → 超步结束:提交状态、落检查点 → 进入下一超步
     → 直到没有待投递的消息(收敛)→ 收集 yield 的输出返回

单个 agent 那一层的主线(第 1 章细讲):agent.run(消息) → 组装历史 + 系统指令 → 调 ChatClient → 若模型要求调工具则执行工具、把结果回灌再调一次 → 收敛后返回 AgentResponse。

3. 阅读地图(建议顺序)​

从「单个 agent」往「多 agent 编排」由浅入深读。前两章讲清楚一个 agent 是什么,第三、四章是全书重点——工作流引擎与它的生产能力,第五章把前面拼成高层模式。

  1. Agent 抽象层:run 循环、ChatClient 与会话 —— 一个 agent 到底由什么组成:BaseAgent / RawAgent / Agent 三层、run 循环怎么和模型来回、会话记忆怎么存。先读这章,它是后面一切的原子。
  2. 工具、中间件与技能:让 agent 长出手脚 —— @tool 怎么把普通函数变成工具、工具调用循环与审批(approval)、三种中间件(Agent/Function/Chat)的洋葱包裹、Agent Skills 领域知识库。
  3. Workflow 图引擎:类型路由的 Pregel 超步 —— 全书核心:Executor + @handler 的类型路由、四种边(单边/扇出/扇入/switch-case)、Runner 的超步循环与收敛判定。
  4. 持久化、人在环路与时间旅行 —— 超步边界为什么是天然的检查点位置、WorkflowCheckpoint 存了什么、request_info 如何让流程挂起等人、恢复与回退重跑怎么做。
  5. 高层编排模式:把 agent 编进工作流 —— Sequential / Concurrent / Handoff / Magentic / GroupChat 五种 Builder 各自把图搭成什么形状、分别解决什么协作问题。

4. 巧妙之处(可带走的技术)​

这几处是「读完值得记住、能借鉴到自己系统里」的设计决策。

  • 两层统一在同一套消息抽象上。 「单 agent」和「多 agent 图」不是两套无关的东西:图里的一个 Executor 内部可以直接包一个 agent,agent 的输出就是流经边的消息。于是学会了单 agent,就自动会用工作流的节点。依据:_workflows/_workflow_builder.py:53(WorkflowBuilder 直接接受 SupportsAgentRun)。

  • 用「消息类型」而不是「显式跳转」来路由。 工作流节点用 @handler 声明自己吃哪种类型的消息;引擎按运行时类型匹配把消息投给能接的 handler。这让图的连线像强类型函数签名一样有约束,连错类型在构图期就能查出来。依据:_workflows/_executor.py:530(handler 装饰器)、_workflows/_executor.py:411(input_types)。

  • Pregel 超步:把「一张图并发跑」化简成「一轮一轮离散推进」。 不追求真并行,而是把执行切成离散超步——每个超步内投递本轮消息、跑活跃节点,超步之间有清晰边界。好处是状态在边界处是干净、可快照的。依据:_workflows/_runner.py:45(Runner,注释即写「run a workflow in Pregel supersteps」)、_workflows/_runner.py:107(run_until_convergence)。

  • 超步边界 = 免费的检查点位置。 正因为超步边界状态干净,框架就在每个超步结束时提交状态并落一个检查点;检查点之间用 previous_checkpoint_id 串成链。这一个设计同时买到了三件事:崩溃续跑、暂停等人、回到任意历史步重跑(时间旅行)。依据:_workflows/_runner.py:123(超步循环里 create_checkpoint_if_enabled)、_workflows/_checkpoint.py:30(WorkflowCheckpoint 快照结构)。

  • 人在环路复用同一条恢复通道。 节点调 ctx.request_info(...) 就发出一个「我需要外部输入」的请求事件、把流程挂起;外部回话后用 workflow.run(responses=...) 从挂起点继续——和「从检查点恢复」走的是同一套机制。依据:_workflows/_workflow_context.py:393(request_info)、_workflows/_workflow.py:695(run 的 responses 参数)。

  • 中间件按关注点分三种、层层包裹。 Agent 级(包住整次 run)、Function 级(包住一次工具调用)、Chat 级(包住一次模型请求)各是一层洋葱,call_next() 决定要不要往里走。依据:_middleware.py:82(MiddlewareType 枚举 AGENT/FUNCTION/CHAT)、_middleware.py:465(AgentMiddleware)。

5. 代码地图(导航索引)​

按符号名 grep 比按行号更抗上游漂移。下表是「想读某个机制该打开哪个文件、找哪个符号」的跳转表(路径相对克隆根,Python 侧为主)。

主题文件路径关键符号
Agent 三层与 run 循环python/packages/core/agent_framework/_agents.pyBaseAgent · RawAgent · Agent
ChatClient 抽象python/packages/core/agent_framework/_clients.pyBaseChatClient · SupportsChatGetResponse
会话 / 记忆python/packages/core/agent_framework/_sessions.pyAgentSession · HistoryProvider · InMemoryHistoryProvider
函数工具与 @toolpython/packages/core/agent_framework/_tools.pyFunctionTool · tool · FunctionInvocationLayer
三种中间件python/packages/core/agent_framework/_middleware.pyMiddlewareType · AgentMiddleware · FunctionInvocationContext · ChatContext
Agent Skillspython/packages/core/agent_framework/_skills.pySkill · InlineSkill · ClassSkill · FileSkill
工作流节点与类型路由python/packages/core/agent_framework/_workflows/_executor.pyExecutor · handler · input_types
边与边组python/packages/core/agent_framework/_workflows/_edge.pyEdge · FanOutEdgeGroup · FanInEdgeGroup · SwitchCaseEdgeGroup
构图 APIpython/packages/core/agent_framework/_workflows/_workflow_builder.pyWorkflowBuilder · add_edge · add_fan_out_edges · add_switch_case_edge_group · add_fan_in_edges · add_chain
超步执行引擎python/packages/core/agent_framework/_workflows/_runner.pyRunner · run_until_convergence · _run_iteration · create_checkpoint_if_enabled
运行上下文(发消息/产出/请求)python/packages/core/agent_framework/_workflows/_workflow_context.pyWorkflowContext · send_message · yield_output · request_info
检查点与存储python/packages/core/agent_framework/_workflows/_checkpoint.pyWorkflowCheckpoint · CheckpointStorage · InMemoryCheckpointStorage · FileCheckpointStorage
工作流对外入口python/packages/core/agent_framework/_workflows/_workflow.pyWorkflow · Workflow.run · WorkflowRunResult
事件流python/packages/core/agent_framework/_workflows/_events.pyWorkflowEvent · superstep_started · superstep_completed · request_info
编排模式 Builderpython/packages/orchestrations/agent_framework_orchestrations/SequentialBuilder(_sequential.py) · ConcurrentBuilder(_concurrent.py) · HandoffBuilder(_handoff.py) · MagenticBuilder(_magentic.py) · GroupChatBuilder(_group_chat.py)
.NET 对应实现dotnet/src/Microsoft.Agents.AI(NuGet 包根)