Microsoft Agent Framework 从哪抄起:微软 AI Agent 入门课第 14 课

2026-08-18

翻到 ai-agents-for-beginners 的 14-microsoft-agent-framework/ 时,很多人会先打开 README.md 从头读,读完仍然不知道该抄哪一段代码。更快的路子是直接进 code-samples/:这一课的示例文件里,「查酒店有没有房 → 有房去下单、没房换个城市」这同一个场景被写了好几遍,每一遍只在骨架上多加一层能力。把这条骨架看懂,剩下几份就是变奏。

下面按「一次请求在代码里经过哪几层」走一遍。先说清楚范围:这一课的 code-samples/ 目录里我们只看到 Python 侧的 .ipynb.py没有 .NET 版本——课程里的 .NET 实现在别的课,例如 04-tool-use/code_samples/04-dotnet-agent-framework.cs,两边的类名与写法不要混着抄。

骨架:一次订房请求经过的五层

最干净的一份是 14-microsoft-agent-framework/code-samples/14-conditional-workflow.ipynb。它的链路是这样的:

第一层,工具。@tool 装饰一个普通函数,参数用 Annotated 带上描述:

@tool(description="Check hotel room availability for a destination city")
def hotel_booking(destination: Annotated[str, "The destination city to check for hotel rooms"]) -> str:

函数体里是一个写死的城市名单,把结果 json.dumps 成字符串返回——它是示例数据,不是真的查房。

第二层,agent。 provider.as_agent(...) 拿到 agent,tools=[hotel_booking] 把工具挂上去,同时用 default_options={"response_format": BookingCheckResult} 指定结构化输出,BookingCheckResult 是一个 pydantic.BaseModel,字段是 destination / has_availability / message

第三层,executor。 agent 外面再套一层 AgentExecutor(..., id="availability_agent")。notebook 的说明里写得很直白:这一层是「wraps agent for workflow use」。WorkflowBuilder 认的是 executor,不是 agent 本身。

第四层,边与条件函数。 路由发生在这里:

workflow = (
    WorkflowBuilder(
        start_executor=availability_agent,
        output_executors=[display_result],
    )
    # NO AVAILABILITY PATH
    .add_edge(availability_agent, alternative_agent, condition=no_availability_condition)
    .add_edge(alternative_agent, display_result)
    # HAS AVAILABILITY PATH
    .add_edge(availability_agent, booking_agent, condition=has_availability_condition)
    .add_edge(booking_agent, display_result)
    .build()
)

condition 的只有从 availability_agent 出去的那两条;两个分支各自还有一条无条件边通到 display_result,别抄漏了,否则输出那一层收不到东西。

条件函数收到的是上游的 AgentExecutorResponse,里面做的事是把文本反序列化回模型再取字段:

result = BookingCheckResult.model_validate_json(message.agent_run_response.text)
return result.has_availability

这两处摆在一起看就清楚了:分支是 Python 代码判的,不是模型自己「决定」的;而条件函数能判,前提是上一层的 response_format 保证了这段文本能被反序列化。第二层那个看起来可选的 response_format,其实是第四层能成立的条件。

第五层,输出。 一个 @executor(id="display_result") 装饰的协程,里面 await ctx.yield_output(response.agent_run_response.text);外面用 events = await workflow.run(request)events.get_outputs() 取回来。入参是 AgentExecutorRequest(messages=[...], should_respond=True)

这五层是这一课所有 workflow 样例的公共部分。后面几份示例改的都是其中某一层。

其余示例各加了哪一层

14-middleware.ipynb——在工具的返回值上动手。 它定义了一个 priority_check_middleware,签名是 (context: FunctionInvocationContext, next: Callable[[FunctionInvocationContext], Awaitable[None]]),通过 middleware=[priority_check_middleware] 挂在 agent 上。关键在执行顺序:函数体里先 await next(context) 让原函数跑完,然后才去读写 context.result——把 has_availability 改成 True、补一个 priority_override 字段,再 json.dumps 写回去。

于是同一套边、同一套条件函数,走向了另一条分支。这个示例的落点不是「middleware 能记日志」,而是 middleware 改的是工具返回值,不是图的结构。README 里另外给了一种 ChatContext 的 chat middleware,拦的是发往模型的 messages,是另一个位置。

14-human-loop.ipynb——把一次 run 拆成两次。 它写了一个继承 ExecutorDecisionManager@handler 标记的方法里调 await ctx.request_info(request_data=HumanFeedbackRequest(...), response_type=str) 把流程挂起,人的回复由另一个 @response_handler 标记的方法接住。调用侧也跟着变形——第一次是 workflow.run(request_paris, stream=True),拿到挂起事件后,第二次用 workflow.run(stream=True, responses=responses) 续跑。这一层值得单独记:加了人工确认之后,你的调用代码不再是「await 一次拿结果」。

14-concurrent.ipynb——扇出。 它自己写了个 InputDispatcher(Executor)@handler 方法里只做一件事 await ctx.send_message(text),然后:

workflow = (
    WorkflowBuilder(start_executor=dispatcher, output_executors=agents)
    .add_fan_out_edges(dispatcher, agents)
    .build()
)

值得注意的是,这一份里三个 agent(attractions / dining / history)是直接 provider.as_agent(...) 建的,没有传 response_format,让模型吐 JSON 靠的是 instructions 里那句「Return structured JSON matching the AttractionsRecommendation schema」。模型与 agent 的对应关系落在下游:代码里有一个 sections 列表,把 agent 名、Pydantic 模型、渲染函数三者按顺序配成一行行。notebook 的注释写明 get_outputs() 返回的顺序与 output_executors 一致,所以下游是按下标 outputs[i].text 去对应模型 model_validate_json 的——顺序错了就解析失败,这是照抄时容易踩的点。

14-sequential.ipynb 就是最朴素的 .add_edge(front_desk_agent, concierge_agent),两个 agent 都列进 output_executorsoutputs[0]outputs[1] 分别是两段。

14-handoff.ipynb——换了一套 builder。 这份不用 WorkflowBuilder,而是从 agent_framework.orchestrations 导入 HandoffBuilder,用 participants=[...].with_start_agent(...).add_handoff(customer_support_agent, [...]) 组装,终止条件是传进去的一个 lambda(termination_condition=lambda conv: ...,按对话里 user 消息条数截断)。这份文件里有一句注释很值钱:Workflow runs are NOT isolated - state is preserved across calls to run(),所以它每个测试用例都重新 build_workflow() 一次。拿这份代码改自己的东西时,别把一个 workflow 实例反复复用还指望它是干净的。

同一个目录里两份写法不一致,抄哪份

code-samples/hotel_booking_workflow_sample.py 是这一课的独立脚本版,场景和 14-conditional-workflow.ipynb 一模一样,但写法对不上:

位置notebook 里独立脚本里
工具装饰器@tool(description=...)@ai_function(description=...)
agent 创建provider.as_agent(...)chat_client.create_agent(...)
结构化输出default_options={"response_format": BookingCheckResult}response_format=BookingCheckResult
用户消息Message(role="user", contents=[...])ChatMessage(Role.USER, text=...)
起点设置WorkflowBuilder(start_executor=...) 构造参数WorkflowBuilder() 之后 .set_start_executor(...)
providerFoundryChatClientagent_framework.foundryOpenAIChatClientagent_framework.openai),按环境变量三选一

更能说明问题的是同一个 notebook 内部也对不上:14-conditional-workflow.ipynb 的 Step 2 说明文字写的是「We use the @ai_function decorator」,紧接着的代码单元用的却是 @tool;末尾的 Key Takeaways 也仍写着 @ai_function。说明文字没跟上代码的改动,这在照着 markdown 抄而不看代码时最容易中招。

两份都在仓库里,都没有标 deprecated。我们能确认的只有这一层事实:这套 API 在演进,同一功能存在两种符号名。所以照抄之前,先确认你装的那一版 agent_framework 到底导出了 tool 还是 ai_function——ImportError 比运行到一半报错要好排查得多。该课程与框架都在持续更新,以仓库最新代码为准。

跑之前要配的东西

仓库根目录的 .env.example 列了变量名,这一课的 notebook 需要的是 AZURE_AI_PROJECT_ENDPOINTAZURE_AI_MODEL_DEPLOYMENT_NAME 两个,配合 azure.identityAzureCliCredential 做免密钥认证——也就是说本机得先用 Azure CLI 登录过。notebook 里靠 load_dotenv().env下面这段是通用做法、不是课程仓的官方内容:Windows 下用 .env 文件通常比设进程环境变量省事,因为 PowerShell 的 $env:XXX="..." 只对当前会话有效,换个终端起 Jupyter 就丢了;Linux/macOS 侧用 export 也是同样的临时性问题。写 .env 文件时记得它别进版本库。

另外,.env.example 里逐字写明:GitHub Models 已废弃(retiring July 2026)且不支持 Responses API,示例已改用 Azure OpenAI 的 Responses API。所以网上老教程里那套 GITHUB_TOKEN 的配法,在这一课已经不适用了。AZURE_AI_MODEL_DEPLOYMENT_NAME.env.example 里给了一个示例部署名,那只是仓库当前的示例值,随版本变动,按你自己 Foundry 项目里的部署名填。

这几个文件里我们没有看到 experimental / preview 之类的标记;能查到的废弃说明只有上面那条关于 GitHub Models 的。示例会调用云端模型服务、会产生费用并把你的输入送出去,密钥与数据边界请自己评估。

以上代码片段均原样取自仓库文件;涉及多处组合时,以上为按仓库代码中的接口语义组合的示例,未经实测,以仓库最新代码为准。

还有一份不属于 workflow 的示例

code-samples/14-langchain-hosted-agent.py 与上面几份不是一路货:它讲的是把一个已经用 LangGraph 写好的 agent 放到 Microsoft Foundry 上托管。README 写明这条路走 langchain_azure_ai.agents.hosting 包,并给了两个宿主类的分工——ResponsesHostServer 暴露 /responses 端点(OpenAI 兼容的对话、流式与会话续接),InvocationsHostServer 暴露 /invocations 端点(自定义 JSON 结构、webhook 式或非对话型处理)。README 建议对话型 agent 从前者起步,理由是仓库文档自述 Responses API 是 Foundry 里 agent 式开发的主 API。

这一份的代码形态也完全不同:graph = create_agent(build_chat_model(), tools=[]) 拿到编译后的图,然后 ResponsesHostServer(graph).run(port=port) 起服务。README 另外交代了两处行为:客户端靠 previous_response_id 或一个 conversation ID 续接对话,会话状态与 LangGraph 的 checkpointer 挂钩;如果图里用了 LangGraph 的 interrupt(),宿主会把挂起的中断转成 Responses 的 function_call / mcp_approval_request 条目,客户端用对应的输出项恢复。也就是说,前面 ctx.request_info() 那套人工介入,在托管形态下换了一种协议表达。

顺带说一句 handoff 那份里的 require_per_service_call_history_persistence=True:它是建 agent 时传的参数,和 HandoffBuilder 的终止条件、以及 README 里列出的 WorkflowStartedEvent / ExecutorInvokeEvent / RequestInfoEvent 这些内置事件一样,属于「读代码时顺手记下来、真出问题才用得上」的那类细节。

一处需要留神的对照

14-concurrent.ipynb 末尾有一个并发与顺序的对比单元。它搭「顺序版」用的是 .add_chain([attractions_agent, dining_agent, history_agent]),而并发版是 dispatcher 把同一段文本广播给三个 agent。把这两处放在一起看:链式结构下,第二个 agent 收到的是第一个 agent 的输出,不是原始的用户输入——14-sequential.ipynb 的说明里就是这么写的(「Front Desk Output → Concierge Agent」)。也就是说,两个 workflow 的差别不只在调度方式,喂给第二、第三个 agent 的输入本身就不一样。这个单元会打印墙钟时间,我们没有跑过,也不引用其中任何数值;只提醒一句:读这个对比时,先看清两边喂进去的东西是不是同一份。


本文依据 github.com/microsoft/ai-agents-for-beginners 仓库于 2026-08-18 的公开内容整理, 事实来自仓库内的课程正文与代码示例。我们没有跑过文中涉及的代码, 因此不涉及运行结果、耗时与 Agent 实际表现的任何描述。 该课程持续更新,文中涉及的文件路径、依赖与接口写法随版本变动,请以仓库最新内容为准。 文中涉及的云端服务调用会产生费用并可能上传数据,请自行评估密钥与数据边界。

想系统学会用 AI?报名体系课或加入会员,照着学、照着用。