课程首页 › Module 7 · 框架与场景:MAF / CUA / 本地 Agent › L16

L16 Microsoft Agent Framework 深入:企业级 Agent 的标准答案

课程目标:深入理解 MAF 的核心抽象(Agents、Tools、Threads、Middleware、Workflows),掌握五种编排模式,能用 MAF 构建生产级 Agent。


一、从一个抱怨开始#

老王是某大型企业的架构师。他想统一公司的 Agent 技术栈:

结果——每个 Agent 的代码风格不一样、监控不一样、记忆机制不一样、错误处理不一样。

老王崩溃:"就没有一个统一标准吗?"

有——这就是 Microsoft Agent Framework 想解决的问题。

MAF(Microsoft Agent Framework)——微软在 2025 年推出的统一 Agent SDK,一个框架覆盖企业 Agent 开发的所有需求。


二、用一个比喻:MAF = 企业的"Office 全家桶"#

把 Agent 框架比作办公软件:

场景工具
写文档Word
做表格Excel
做演示PowerPoint
发邮件Outlook
构建 AgentMAF(一个 SDK 全包)

MAF 的野心:让企业只学一个框架,就能搞定:


三、MAF 的核心抽象#

Microsoft Agent Framework 列出了 MAF 的六大核心概念。

3.1 Agents(Agent 对象)#

创建方式——选一个 chat client,定义名字和 instructions:

from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
    instructions="You are good at recommending trips to customers based on their preferences.",
    name="TripRecommender"
)

可选的 chat client:

# 用 MiniMax(兼容 OpenAI)
from agent_framework.openai import OpenAIChatClient

agent = OpenAIChatClient(
    base_url="https://api.minimax.io/v1",
    api_key=os.environ["MINIMAX_API_KEY"],
    model_id="MiniMax-M3"
).create_agent(
    name="HelpfulAssistant",
    instructions="You are a helpful assistant."
)

3.2 Tools(工具)#

注册工具:

from pydantic import Field
from typing import Annotated

def get_attractions(
    location: Annotated[str, Field(description="The location to get the top tourist attractions for")],
) -> str:
    """Get the top tourist attractions for a given location."""
    return f"The top attractions for {location} are Eiffel Tower, Louvre, Notre Dame."

agent = ChatAgent(
    chat_client=OpenAIChatClient(),
    instructions="You are a helpful assistant",
    tools=[get_attractions]
)

两种注册时机:

3.3 Threads(对话线程)#

创建并复用:

thread = agent.get_new_thread()

# 第一轮
await agent.run("Hello, I am here to help you book travel.", thread=thread)

# 第二轮(同一个 thread,记忆延续)
await agent.run("Where would you like to go?", thread=thread)

# 序列化和恢复(持久化场景)
serialized = await thread.serialize()
restored = await agent.deserialize_thread(serialized)

3.4 Middleware(中间件)#

中间件让你能在 LLM 调用前后插入自定义逻辑:

async def logging_function_middleware(
    context,  # FunctionInvocationContext
    next
):
    print(f"[Function] Calling {context.function.name}")
    await next(context)
    print(f"[Function] {context.function.name} completed")

agent = ChatAgent(
    chat_client=OpenAIChatClient(),
    instructions="You are helpful",
    middleware=[logging_function_middleware]
)

两种中间件:

3.5 Memory(记忆)#

短期记忆(in-memory):默认在 thread 里

长期记忆(persistent):用 chat_message_store_factory:

from agent_framework import ChatMessageStore

def create_message_store():
    return ChatMessageStore()

agent = ChatAgent(
    chat_client=OpenAIChatClient(),
    chat_message_store_factory=create_message_store  # 跨 session 持久化
)

动态记忆(Mem0 等外部服务):

from agent_framework.mem0 import Mem0Provider

memory_provider = Mem0Provider(
    api_key="...",
    user_id="ben"
)

agent = ChatAgent(
    chat_client=OpenAIChatClient(),
    context_providers=memory_provider  # 自动注入相关记忆到 context
)

3.6 Observability(可观测性)#

MAF 原生集成 OpenTelemetry:

from agent_framework.observability import get_tracer, get_meter

tracer = get_tracer()
meter = get_meter()

with tracer.start_as_current_span("my_custom_span"):
    # 自动追踪 agent 执行
    counter = meter.create_counter("my_custom_counter")
    counter.add(1, {"key": "value"})

每个 LLM 调用、工具调用、orchestration 步骤都会变成 span——直接导出到 Foundry 仪表板。


四、Workflows:五大编排模式#

Microsoft Agent Framework 列出了 MAF 的五大编排模式:

模式何时用
Sequential流水线任务,前一步的输出是后一步的输入
Concurrent多个任务并行执行
Group Chat多个 Agent 讨论一个话题
Handoff一个 Agent 完成子任务后交给另一个
MagneticManager Agent 维护任务列表,分给 subagents

4.1 Sequential(顺序)#

from agent_framework import WorkflowBuilder

builder = WorkflowBuilder()
builder.add_edge(planner, executor)
builder.set_start_executor(planner)
workflow = builder.build()

# planner -> executor 顺序执行
result = await workflow.run("Plan and execute a trip to Paris")

4.2 Concurrent(并发)#

# 多个 researcher 并行查资料
builder = WorkflowBuilder()
builder.add_fan_out_edge(planner, [researcher1, researcher2, researcher3])
builder.add_fan_in_edge([researcher1, researcher2, researcher3], summarizer)
builder.set_start_executor(planner)

4.3 Handoff(交接)#

# 客服 Agent 处理不了时交接给技术支持
builder = WorkflowBuilder()
builder.add_edge(customer_service, tech_support, condition=needs_tech_help)
builder.set_start_executor(customer_service)

4.4 Magnetic(磁性编排)#

最强模式——Manager Agent 动态维护任务列表:

manager = provider.as_agent(
    name="manager",
    instructions="""You manage a task list. Delegate tasks to subagents as needed.
    Update the task list as work progresses."""
)

researcher = provider.as_agent(name="researcher", tools=[search_tool])
writer = provider.as_agent(name="writer", tools=[write_tool])

magnetic_workflow = create_magnetic_workflow(
    manager=manager,
    subagents=[researcher, writer]
)

五、LangGraph Agent 托管到 Foundry#

Microsoft Agent Framework 最后一个亮点——MAF 可以托管 LangGraph Agent:

# 用 LangGraph 写 Agent
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_azure_ai.agents.hosting import ResponsesHostServer

graph = create_agent(ChatOpenAI(model="gpt-5-mini"), tools=[])

# 暴露成 Foundry 标准的 Responses 端点
ResponsesHostServer(graph).run(port=8088)

意义:你继续用 LangGraph 写 Agent,但 Foundry 帮你处理 runtime、session、扩展性、身份管理、协议端点。

这意味着——MAF 不是替代 LangGraph,而是给 LangGraph 加上企业级运行时。


六、动手:用 MAF 写一个差旅 Agent#

import asyncio
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework import ChatAgent
from azure.identity import AzureCliCredential

def search_flight(origin: str, destination: str) -> str:
    """Search available flights."""
    return f"Found 5 flights from {origin} to {destination}"

def book_hotel(city: str) -> str:
    """Book a hotel in the city."""
    return f"Booked a hotel in {city}"

# 创建 Agent
agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
    instructions="You are a travel assistant. Help users plan travel.",
    name="TravelAgent",
    tools=[search_flight, book_hotel]
)

# 对话
async def main():
    thread = agent.get_new_thread()
    response1 = await agent.run(
        "I want to plan a trip to Paris from Shanghai",
        thread=thread
    )
    print(response1)

    response2 = await agent.run(
        "Find me a flight and book a hotel",
        thread=thread
    )
    print(response2)

asyncio.run(main())

七、MAF vs LangGraph vs AutoGen:怎么选#

维度MAFLangGraphAutoGen
抽象层级高(封装好)中(显式图)中(对话驱动)
多 Agent 编排5 种内置模式自定义图内置 Group Chat
企业级特性强(原生 Foundry 集成)弱(需要自己集成)中
学习曲线中高中
灵活性中高中

选 MAF 的场景:

继续用 LangGraph 的场景:


八、关键要点回顾#

概念用途
ChatAgent创建 Agent 的统一入口
Tools注册工具函数
Threads维护多轮对话状态
Middleware插入自定义逻辑(日志、认证、限流)
Memory短期 / 长期 / 外部服务三种模式
WorkflowsSequential / Concurrent / Group Chat / Handoff / Magnetic
ObservabilityOpenTelemetry 原生集成

MAF 的核心价值:一个 SDK 覆盖企业 Agent 开发的所有需求。


九、参考资源#


十、下一课预告#

L17 浏览器 Agent(CUA):让 Agent 像人一样操作浏览器——打开网页、点击按钮、提取信息。

为什么重要:现实世界 99% 的网站没有 API,Agent 要工作就得能操作浏览器。

自测 · 检验是否真的掌握了

1. MAF 的五种编排模式中,哪一种是由 Manager Agent 维护任务列表并动态分派给子 Agent 的?

Magnetic 编排引入一个 manager 角色,它维护任务列表并按需把任务分给各个 subagent,任务完成后还能动态调整后续安排。Sequential 是固定顺序的流水线,Concurrent 是并行扇出后汇聚,Handoff 则是按条件把控制权交给下一个 Agent,它们的流程结构在定义时就固定了。Magnetic 最灵活,也最适合任务边界事先无法完全确定的场景。

2. MAF 中的 thread 承担什么职责?

thread 是对话线程,负责承载多轮对话的状态:同一个 thread 上连续 run,模型才知道上一轮说了什么;thread 还能被序列化和反序列化,用于持久化场景下的恢复。工具注册属于 Agent 构造阶段的事情,中间件则负责在调用前后插入横切逻辑。状态与执行逻辑分离,正是 MAF 让 Agent 易于扩展的原因。

3. MAF 可以托管用 LangGraph 编写的 Agent,这说明二者是什么关系?

MAF 托管 LangGraph Agent 的意义在于:你继续用熟悉的框架写控制流,由 Foundry/MAF 提供运行时、会话管理、扩缩容、身份与协议端点这些企业级能力。二者不是互斥的竞争者,框架层与运行时层可以叠加。完全替代或完全并列的说法都忽略了分层协作这一实际形态。