Microsoft Agent Framework 入门

这篇适合谁，读完能做什么

这篇适合已经会写一点 Python、想用微软技术栈做智能体的开发者。读完你应该能分清 Microsoft Agent Framework（MAF）的几个核心概念：智能体、线程、中间件、记忆和工作流，并能照着写出第一个能运行的智能体。

它是什么，和同类有什么不同

Microsoft Agent Framework（MAF）是微软统一的智能体开发框架，用来构建生产和研究场景中的 AI 智能体^[1]。它支持顺序、并发、群聊、接力（handoff）和磁吸式（magnetic）等多种编排方式。磁吸式编排是指由一个管理智能体创建并修改任务列表，再协调子智能体完成任务^[1]。

MAF 还为生产环境准备了几项能力。它用 OpenTelemetry 做可观测性，工具调用、编排步骤、推理流程和性能指标都可以在 Microsoft Foundry 的仪表盘里看到^[1]。智能体线程和工作流可以暂停、恢复，并从错误中恢复，这让长时间运行的流程成为可能^[1]。对于需要人工审批的任务，它支持人在回路（human-in-the-loop）的工作流^[1]。

在互通性上，MAF 的智能体可以运行在容器、本地机房和多个云上。创建智能体时可以用自己习惯的 SDK，包括 Azure OpenAI 和 OpenAI^[1]。智能体还能通过 Agent-to-Agent（A2A）和模型上下文协议（MCP）发现并调用其他智能体和工具^[1]。

准备工作

你需要一个 Python 环境和一个推理服务，可以是 Azure OpenAI，也可以是 OpenAI。要把 LangGraph 智能体托管到 Foundry，还需要 Foundry Project Manager 角色，并在本机执行 az login，让 DefaultAzureCredential 能完成认证^[1]。

第一步：创建智能体

创建智能体需要三样东西：推理服务（也就是大模型提供方）、一组指令，以及一个 name^[1]。下面的例子用 Azure OpenAI 创建一个旅行推荐智能体：

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

第二步：运行智能体

智能体用 .run 做非流式调用，用 .run_stream 做流式调用^[1]。下面的例子是非流式的：

result = await agent.run("What are good places to visit in Amsterdam?")
print(result.text)

每次运行还可以传入 max_tokens、tools 和 model 等选项。某个任务需要特定模型或工具时，可以只在这一次运行中指定，不必改动智能体本身^[1]。

第三步：多轮对话与线程

多轮对话靠线程（thread）维持。调用 get_new_thread() 创建的线程可以保存下来，以后继续用；如果不创建，框架会在一次运行中自动建一个线程，运行结束就失效^[1]。

需要长期保存时，可以把线程序列化成数据存起来，之后再恢复。序列化用 await thread.serialize()，恢复用 await agent.deserialize_thread(serialized_thread)^[1]。

中间件：在智能体和工具、大模型之间插入逻辑

中间件用来在智能体执行任务的过程中插入动作，比如记录日志。MAF 提供两种中间件^[1]。

函数中间件位于智能体和它调用的函数或工具之间，可以在调用前后做处理，例如记录函数调用。中间件里的 next 决定是交给下一层中间件，还是直接执行函数^[1]。

聊天中间件位于智能体和发给大模型的请求之间，它的上下文里包含发送给 AI 服务的 messages，因此可以用来记录每一次模型调用^[1]。

记忆的三种形式

MAF 的智能体记忆分为三类^[1]：

内存存储：保存在线程里，只在程序运行期间有效。
持久化消息：通过 chat_message_store_factory 指定，用来跨会话保存对话历史。
动态记忆：在每次运行前加入上下文，可以放在 mem0 等外部服务里，通过 context_providers 接入。

工作流：把多个智能体和逻辑串起来

工作流是预先定义好的一组步骤，其中可以包含 AI 智能体。工作流支持多智能体编排，也支持检查点（checkpointing），把工作流的状态保存下来^[1]。

工作流由几个核心部件组成。执行器（executor）接收输入消息，完成自己的任务，再输出一条消息，推动工作流往下走。执行器可以是 AI 智能体，也可以是自定义逻辑^[1]。

执行器之间的连接叫边（edge），常见的有直连边、条件边、switch-case 边、扇出边（fan-out）和扇入边（fan-in）^[1]。最简单的直连边可以这样写：

from agent_framework import WorkflowBuilder

builder = WorkflowBuilder()
builder.add_edge(source_executor, target_executor)
builder.set_start_executor(source_executor)
workflow = builder.build()

工作流运行时会产生事件，便于观察执行过程。常见的事件有 WorkflowStartedEvent（开始）、WorkflowOutputEvent（产出结果）、WorkflowErrorEvent（出错）、ExecutorInvokeEvent（执行器开始处理）、ExecutorCompleteEvent（执行器处理完成）和 RequestInfoEvent（发出请求）^[1]。

把 LangGraph 智能体托管到 Foundry

MAF 不限于自己写的智能体。如果你已经用 LangChain 或 LangGraph 写好了智能体，可以借助 langchain_azure_ai.agents.hosting 包，把编译后的 LangGraph 图作为 Foundry 托管智能体运行。运行时、会话、伸缩、身份和协议端点由 Foundry 管理，智能体逻辑仍然留在 LangGraph 里^[1]。

先安装带 hosting 扩展的包，命令是 pip install -U "langchain-azure-ai[hosting]>=1.2.4" azure-identity^[1]。

托管有两种协议。ResponsesHostServer 提供 /responses 端点，适合大多数对话型智能体，也是官方建议的默认选择。InvocationsHostServer 提供 /invocations 端点，适合自定义 JSON 结构或非对话类的处理^[1]。

运行前设置两个环境变量。FOUNDRY_PROJECT_ENDPOINT 填项目地址，格式是 https://<resource>.services.ai.azure.com/api/projects/<project>；FOUNDRY_MODEL_NAME 填模型部署名，例如 gpt-5-mini^[1]。

示例代码保存为 main.py，本地用 python main.py 启动，然后向 http://localhost:8088/responses 发送 Responses 请求即可测试^[1]。

多轮对话怎么保存状态

客户端续接对话时，可以传 previous_response_id，也可以传 conversation ID。如果图是带 LangGraph 检查点保存器（checkpointer）编译的，对话状态就和检查点绑定。MemorySaver 适合本地测试，正式环境要换成持久化的检查点保存器^[1]。

人工审批怎么接入

如果图里用了 LangGraph 的 interrupt()，ResponsesHostServer 会把挂起的中断以 function_call 或 mcp_approval_request 条目返回给客户端。客户端再用对应的 function_call_output 或 mcp_approval_response 继续执行^[1]。

用 azd 部署到 Foundry

部署使用 Azure Developer CLI。依次执行 azd ext install azure.ai.agents 安装扩展，azd ai agent init -m <manifest> 初始化，azd ai agent run 在本地运行（需要 Docker），最后执行 azd provision 和 azd deploy。部署托管智能体需要 Foundry Project Manager 角色^[1]。