微软 AI Agent 入门课第 8 课的多智能体：角色、消息与终止条件怎么写

看 08-multi-agent/README.md 的时候很容易产生一种错觉：多智能体好像是个纯设计问题。那一章确实是这么写的，它列了实现这个模式的几块积木——Agent Communication、Coordination Mechanisms、Agent Architecture、Visibility into Multi-Agent Interactions、Multi-Agent Patterns、Human in the loop，然后用订机票订酒店租车那个例子讲了一遍谁该跟谁通气。

问题是，等你打开 08-multi-agent/code_samples/08-python-agent-framework.ipynb 开始照着敲，卡住的往往不是”要不要拆 agent”，而是三个很具体的位置：角色写在哪个参数里、消息沿哪条边走、什么时候算跑完。这三处在这一课的代码里各有各的写法，而且散在不同文件中，值得一次性理清楚。

角色：as_agent 的 name 不只是标签

主 notebook 里先建客户端，再由客户端派生 agent：

client = FoundryChatClient(
    project_endpoint=endpoint,
    model=deployment_name,
    credential=DefaultAzureCredential()
)

endpoint 与 deployment_name 分别读自 AZURE_AI_PROJECT_ENDPOINT 与 AZURE_AI_MODEL_DEPLOYMENT_NAME，这两个变量名在仓库根的 .env.example 里也有对应条目。notebook 的 setup cell 还做了一件值得抄的事：把这两个变量收进一个 missing 列表，缺哪个就直接 raise ValueError 报出缺的是哪一个，而不是等到调用时才炸。

角色本身只有两个参数：

planner_agent = client.as_agent(
    name="TravelPlanner",
    instructions="You are a travel planning specialist. Create detailed trip itineraries based on the traveler's preferences. Include daily schedules, must-see attractions, and logistical tips.",
)

name 这个字段容易被当成随手起的标签，但它在后面是有用的。流式循环里判断”现在是谁在说话”，靠的就是 update.author_name：

async for event in events:
    if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
        update = event.data
        author = update.author_name

也就是说，name 是你在输出流里区分角色的唯一线索。同一课的 code_samples/workflows-agent-framework/python/ 下几个 notebook 干脆把名字和指令都提到常量里（SalesAgentName / SalesAgentInstructions 这种成对写法），角色定义和图的搭建就彻底分开了，改 prompt 不用碰拓扑。

消息：边是 add_edge 加出来的

两个 agent 串起来是这样：

workflow = WorkflowBuilder(start_executor=planner_agent) \
    .add_edge(planner_agent, concierge_agent) \
    .build()

加第三个角色时，notebook 只多写了一条边：

extended_workflow = WorkflowBuilder(start_executor=planner_agent) \
    .add_edge(planner_agent, concierge_agent) \
    .add_edge(concierge_agent, budget_agent) \
    .build()

这一课的 Summary 里自述，这样做的好处是”扩展工作流时不必修改已有的 agent”（仓库文档自述）。从代码看确实如此：budget_agent 的加入没有动前两个 agent 的任何参数。

但顺序图只是最简单的一种。同目录 03.python-agent-framework-workflow-ghmodel-concurrent.ipynb 里出现了消息接口的真身——一个自定义 Executor：

class InputDispatcher(Executor):
    """Forward the user input unchanged to all participating agents."""

    @handler
    async def forward(self, text: str, ctx: WorkflowContext[str]) -> None:
        await ctx.send_message(text)

这里有三个东西第一次露面：@handler 装饰器标出这个方法是消息入口；WorkflowContext[str] 的类型参数标明这条边上往下传的是什么；ctx.send_message(...) 才是真正把消息推给下游的动作。图的搭法也换了：

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

add_fan_out_edges 把一份输入广播给列表里的每个 agent。注意 output_executors 这个参数——它是下一节的关键。

还有个细节值得单独说：主 notebook 的流式循环里维护了一个 last_author 变量，每次 update.author_name 变了才打一条分隔线并把新名字打出来，否则就用 print(update.text, end="", flush=True) 不换行地往下追加。这段代码本身没什么技巧，但它揭示了一件事——流式事件是按 token 粒度到达的，“哪个 agent 在说话”这个信息挂在每一个 update 上，不是一段一段成块给你的。你要做自己的界面或者日志，切分逻辑得自己写，判据就是 author_name 变没变。第 8 课 README 讲 Logging and Monitoring Tools 时说日志条目应该记下”是哪个 agent 执行的动作”，落到代码里，可取的那个字段就是它。

终止：三个完全不同的位置

这是我认为这一课最容易看漏的地方。“什么时候算跑完”在这套代码里不是一个概念，而是散在三处。

第一处，在 build 的时候声明。 上面那个并发例子显式传了 output_executors=agents，取结果时的注释写明 outputs 是一个 AgentResponse 列表，“one per output executor，顺序按传给 output_executors 的顺序”。而 02.python-agent-framework-workflow-ghmodel-sequential.ipynb 的注释说得更直接：当前的 Workflow 返回一个结果对象，get_outputs() 给出每个 output executor 的 AgentResponse，“最后一级（quote_agent）是这里唯一的 output”。

第二处，在读结果的时候收口。

events = await workflow.run(message)
outputs = events.get_outputs()
result = outputs[0].text if outputs else ""

顺序图没有条件分支，跑到出边为空的节点就结束，get_outputs() 拿到的就是终点的输出。

第三处，写在 executor 内部。 真正意义上的”终止条件”出现在 04.python-agent-framework-workflow-aifoundry-condition.ipynb，那里有一对语义完全不同的调用：

@executor(id="handle_review")
async def handle_review(review: ReviewResult, ctx: WorkflowContext[str]) -> None:
    if review.review_result == "No":
        await ctx.yield_output(f"Review failed: {review.reason}, please revise the draft.")
    else:
        await ctx.send_message(
            AgentExecutorRequest(messages=[ChatMessage(Role.USER, text=review.draft_content)], should_respond=True)
        )

ctx.yield_output(...) 是”到此为止，这就是结果”，ctx.send_message(...) 是”继续往下走”。判定终止的代码就写在这一个 if 里，它不在框架里，是你自己写的。

而路由本身又是另一个函数。同一个 notebook 里，分支靠 selection_func 决定：

def select_targets(review: ReviewResult, target_ids: list[str]) -> list[str]:
        handle_review_id, save_draft_id = target_ids
        if review.review_result == "Yes":
            return [save_draft_id]
        else:
            return [handle_review_id]

它挂在这条边上（上下两段分别抄自该 notebook 的两个 cell）：

.add_multi_selection_edge_group(
    to_reviewer_result,
    [handle_review, save_draft],
    selection_func=select_targets,
)

有个连锁条件不能忽略：select_targets 读的是 ReviewResult.review_result，而这个字段来自把审稿 agent 的文本用 pydantic 解析出来——ReviewAgent.model_validate_json(response.agent_run_response.text)，其中 review_result 被声明为 Literal["Yes", "No"]。换句话说，终止判定的可靠性一路依赖到”agent 是否老老实实输出了可解析的 JSON”，而这一点在这份代码里是靠 prompt 里那句”return result as JSON with fields …”来保证的，没有额外兜底。这层耦合是仓库代码本身的结构，值得在自己改造时先想清楚。

几处写法不一致，别照着上一个文件套

同一课下面这几个 notebook 是不同时期的产物，接口写法并不统一，这是最容易踩的坑：

• 主 notebook 用 workflow.run(..., stream=True) 再 async for 遍历；02 的注释明写 Workflow.run_stream 已不属于公开 API；而 04 里仍然是 async for event in workflow.run_stream(task)。
• 起点的声明有两种：WorkflowBuilder(start_executor=...) 与 WorkflowBuilder().set_start_executor(...)。
• 客户端也分两路：主 notebook 与 02/03 用 agent_framework.foundry 的 FoundryChatClient，04 用 agent_framework.azure 的 AzureAIAgentClient，安装的包也不同（前者 agent-framework，后者 agent-framework-azure-ai）。
• 文件名里的 ghmodel 已经和内容对不上了：02 的 migration note 写明这些示例原先引用 GitHub Models，因其退役而改用 Microsoft Foundry，但文件名没改。

我们没有跑过这些代码，所以不去判断哪种写法当前有效。能确定的只是：照着其中一个文件的写法去改另一个，很可能对不上，动手前先看你打开的那个文件自己是怎么写的。

至于第 8 课 README 反复强调的”对交互过程的可见性”，代码侧对应的东西是 WorkflowViz：viz.to_mermaid() 与 viz.to_digraph() 把图导成文本，viz.export(format="svg") 导成图片。02 和 03 都给这一步套了 except ImportError 回退，注释说明 SVG 导出需要额外的 graphviz 依赖，缺了就退回上面的文本输出。

Windows 侧要注意的两点

一是凭据类型不统一：主 notebook 用 DefaultAzureCredential，02/03/04 用 AzureCliCredential，02 的说明里写明要先用 Azure CLI 登录（az login）AzureCliCredential 才能完成认证。在 Windows 上这一步同样是先在 PowerShell 里登录，再启动 notebook 内核，顺序反了当前会话可能读不到登录状态。

二是 .NET 示例的运行方式。08-dotnet-agent-framework.md 给了两条路径，其中 chmod +x 那条明确标注是 Linux/macOS，Windows 上没有这一步，直接用 .NET CLI：

dotnet run 08-dotnet-agent-framework.cs

顺带说一下 .NET 那份的三点差异（它是独立实现，别和 Python 混着看）。它的图搭法是 new WorkflowBuilder(frontdeskAgent).AddEdge(frontdeskAgent, reviewerAgent).Build()；执行走 InProcessExecution.RunStreamingAsync(...)，随后还要 await run.TrySendMessageAsync(new TurnToken(emitEvents: true)) 才拿得到事件；事件里区分角色用的是 AgentResponseUpdateEvent 的 ExecutorId，而不是 Python 侧的 author_name。另外一个细节：角色文本在这份 C# 里是赋给 ChatClientAgentOptions 的 Description 属性，而 Python 侧是 as_agent(instructions=...)——这是仓库里白纸黑字的两种写法，我们没有实际运行，不对哪种更合适下判断。

.NET 这份同样没有代码层面的终止条件，两个 agent 跑完图就结束；Concierge 的角色指令里那句”如果满意就声明 approved”是写在 prompt 里的约定，代码里并没有与之对应的判定分支。想要”审核不过就打回重写”，还是得回到 Python 04 那种 yield_output / send_message 的写法自己实现。

最后提醒一处课文与代码之间的空档。第 8 课 README 把 Human in the loop 列为实现这个模式的积木之一，明确说大多数场景里都会有人参与，你需要告诉 agent 什么时候该请人介入（比如订之前先要一次确认）。但翻遍这一课 code_samples/ 下的 Python 与 .NET 示例，我们没有找到任何一处对应的实现——04 里的分支判定完全由 select_targets 这个纯函数做掉，没有停下来等人回话的环节。要做人工确认，那个位置只能落在 executor 内部：在 yield_output 与 send_message 之间插入你自己的等待逻辑。这属于课文写到了、示例代码没覆盖的部分，别指望照抄。

真要落地，我的建议是先把这三处分开确认：角色在 as_agent 的参数里、边在 add_edge / add_fan_out_edges / add_multi_selection_edge_group 里、终止在 output_executors 与 yield_output 里。这三处对不上，图画得再漂亮也跑不出你要的东西。

---

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