数据截至 2026-08-18(上游 commit 1f738cdeb7f5)

高层编排模式:把 agent 编进工作流

30 秒导读: 前面几章讲了单个 agent 怎么跑(第 01 章)、怎么长手脚(第 02 章),以及底层那台"类型路由的 Pregel 图引擎"怎么转(第 03 章)。本章讲的是最上面那层:框架怎么把"多个 agent 协作"这件事,变成五种开箱即用的拓扑(顺序、并行、去中心路由、编排者主导、Magentic 自规划)。核心洞察只有一句——这五种模式不是各写一套引擎,而是各自在同一台图引擎上"生成一张特定形状的图"。理解了这句,五个 Builder 就都通了。

本章覆盖两块内容:

桥接层——让 agent 和 workflow 能互相包裹的三个适配器(AgentExecutor、WorkflowAgent、WorkflowExecutor)。
编排包——agent_framework_orchestrations 里的五个高层 Builder,以及压轴的 Magentic 台账机制。

1. 先搞清楚一件事:两个世界要打通

框架里有两个"世界",词汇不一样:

世界	基本单位	怎么调用	输出
Agent 世界	`Agent` / 任何 `SupportsAgentRun`	`agent.run(messages)`	`AgentResponse`
Workflow 世界	`Executor`(图里的节点)	引擎按边把消息路由给它	靠 `ctx.send_message` / `ctx.yield_output`

编排的本质,就是把 agent 塞进 workflow 的图里当节点跑。但 agent 的接口(run)和节点的接口(收消息、发消息)对不上。所以框架先造了一层适配器,把两个世界的接口互相翻译。

桥接层一共三个适配器,方向各不同:

                     桥接层三件套(谁包谁)

   Agent  ──包成──►  AgentExecutor          （agent 当图里一个节点)
                     └ 收 AgentExecutorRequest，发 AgentExecutorResponse

   Workflow ──包成──►  WorkflowAgent          （整张图反过来当一个 agent)
                     └ 对外暴露 .run()，内部把 workflow 事件翻成 AgentResponse

   Workflow ──包成──►  WorkflowExecutor       （子工作流当父图里一个节点)
                     └ 图套图,支持嵌套

先把这三个适配器讲透,后面五个 Builder 才有地基。

2. 桥接件一:AgentExecutor —— 把 agent 包成节点

它解决的小问题: 图引擎只认 Executor。你有一个 agent,想让它在图里当一个节点,谁来收发消息、谁来维护对话上下文?

思路: 写一个 Executor 子类,内部持有 agent;收到消息就攒进缓存,该回复时调 agent.run(),把结果打包成一个标准信封发给下游。这个包装类就是 AgentExecutor(python/packages/core/agent_framework/_workflows/_agent_executor.py:119,class AgentExecutor)。

2.1 两个标准信封

整个编排包的节点之间,传的都是这两个 dataclass:

信封	方向	关键字段	源码
`AgentExecutorRequest`	发给 agent 节点	`messages`、`should_respond`(是否要它真的回复)	`_agent_executor.py:32`(`class AgentExecutorRequest`)
`AgentExecutorResponse`	agent 节点发出	`executor_id`、`agent_response`、`full_conversation`(到此为止的完整对话)	`_agent_executor.py:46`(`class AgentExecutorResponse`)

should_respond=False 是个关键设计:它让编排者可以只把消息灌进某个 agent 的上下文缓存、但不让它现在开口(见 run handler,_agent_executor.py:197)。后面群聊/交接的"广播同步"全靠这个开关。

full_conversation 也不是摆设。它保证下游 agent 拿到的是完整对话历史而不是只有上一个 agent 的最后一句——否则链条越长,前面的用户提问越容易丢。

2.2 无缝链接:三种输入都能接

AgentExecutor 定义了一组 handler,靠输入类型自动分派(这正是第 03 章讲的类型路由):

收到的类型	走哪个 handler	行为
`AgentExecutorRequest`	`run`	标准路径,攒缓存后按需回复
`AgentExecutorResponse`	`from_response`	上一个 agent 的输出直接喂进来,继续对话
`str`	`from_str`	裸字符串当新用户输入
`Message` / `list[Message]`	`from_message` / `from_messages`	单条/多条消息

from_response(_agent_executor.py:213)里藏着一个上下文策略开关 context_mode:

# 示意,非源码:from_response 里怎么决定"把多少历史喂给下一个 agent"
if context_mode == "full":          # 默认:全量历史都带上
    cache.extend(prior.full_conversation)
elif context_mode == "last_agent":  # 只带上一个 agent 的回复
    cache.extend(prior.agent_response.messages)
else:                               # custom:用户给的过滤函数说了算
    cache.extend(context_filter(prior.full_conversation))

一个容易踩的坑(源码里专门警告了): 如果你写自定义 executor,想改写 agent 的输出文本,别直接 send_message 一个裸 str——那会命中下游的 from_str,把完整对话历史丢光。要用 AgentExecutorResponse.with_text(...)(_agent_executor.py:62),它保持信封类型不变,于是走 from_response,历史得以保留。这个坑在 from_str 的 docstring(_agent_executor.py:244)里被明确点名。

3. 桥接件二:WorkflowAgent —— 把整张图反过来当 agent

它解决的小问题: 你辛辛苦苦编排了一张多 agent 的图,现在想把它当成一个普通 agent 塞进别人的系统(或者再嵌进另一张图)。可是图的接口是"跑起来吐一串事件",不是 run() -> AgentResponse。

思路: 反向包装。WorkflowAgent(python/packages/core/agent_framework/_workflows/_agent.py:52,class WorkflowAgent)继承 BaseAgent,对外长得就是个 agent——有 .run();内部把 workflow 跑出来的事件流,翻译回 AgentResponse / AgentResponseUpdate。

它在构造时会做一个类型校验:workflow 的起始节点必须能吃 list[Message],否则拒绝包装(_agent.py:117)——因为 agent 的输入就是消息列表,图的入口得对得上。

翻译规则很清晰,只放行两类事件(_agent.py:_convert_workflow_events_to_agent_response,起于 :483):

workflow 事件类型	翻成什么
`output`(终态输出)	追加进 `AgentResponse.messages`
`request_info`(要人介入)	翻成一个"函数审批请求"内容,交给上层处理
其它(生命周期、诊断、编排内部事件如 `group_chat`/`handoff_sent`/`magentic_orchestrator`)	一律丢弃

这就是"图套 agent 套图"能无限嵌套的原因:每一层只暴露干净的 AgentResponse,内部噪音全被这层滤掉。

4. 桥接件三:WorkflowExecutor —— 子工作流嵌套

它解决的小问题: 上面 WorkflowAgent 是"图 → agent"。但如果我想直接把一张子图当成父图里的一个节点(不经过 agent 这层皮),怎么办?

思路: WorkflowExecutor(python/packages/core/agent_framework/_workflows/_workflow_executor.py:106,class WorkflowExecutor)把一整张 workflow 包成一个 Executor。父图给它一条消息,它就在内部跑完子图,再把子图的输出转发回父图。

两个要点:

输出转发有开关(allow_direct_output,_workflow_executor.py:525):默认把子图输出当普通消息 send_message 给父图的下游节点;开成 True 则直接 yield_output,让子图的输出就是父图的输出。
请求/响应会跨层协调:子图中途需要外部输入(比如人在环路),WorkflowExecutor 会把请求包成 SubWorkflowRequestMessage 冒泡给父图,父图应答后再喂回子图恢复执行(_workflow_executor.py:_process_workflow_result,起于 :537)。每次子图调用都有独立的 ExecutionContext 做隔离,支持并发多次调用。

一句话记住三件套的分工:

AgentExecutor      : agent  → 节点        （最常用,五个 Builder 的地基)
WorkflowAgent      : 图     → agent       （对外封装 / 无限嵌套)
WorkflowExecutor   : 图     → 节点        （图套图,子工作流)

5. 编排包全景:五个 Builder,五种拓扑

有了 AgentExecutor 这块地基,agent_framework_orchestrations 包提供了五个高层 Builder。它们的共同套路是:

   participants=[agent1, agent2, ...]
            │
            ▼
   XxxBuilder.build()
            │
            ├─ 1. 把每个 agent 包成 AgentExecutor(或其特化子类)
            ├─ 2. 造若干"内部节点"(分发器/聚合器/编排者)
            ├─ 3. 按这个模式的拓扑,在 WorkflowBuilder 上连边
            ▼
        一张 Workflow（回到第 03 章那台图引擎)

关键认知:五个 Builder 本身不含执行逻辑,它们只是"图的生成器"。真正跑的还是第 03 章那台超步引擎。区别只在连边的形状:

Builder	拓扑一句话	谁决定下一个谁说话	中心化?
`SequentialBuilder`	链:A→B→C	固定顺序	——
`ConcurrentBuilder`	扇出并行再扇入	全体并行,无先后	——
`HandoffBuilder`	网状:agent 自己交接	agent 自己(调交接工具)	去中心
`GroupChatBuilder`	星形:编排者居中	编排者(选择函数/agent)	中心化
`MagenticBuilder`	星形 + 自规划循环	manager(进度台账)	中心化

拓扑对比图(方向统一从左到右 / 居中):

Sequential:   IN → A → B → C → OUT           （一条链)

Concurrent:        ┌→ A ┐
              IN → ┤  B ├ → 聚合器 → OUT       （扇出/扇入)
                   └→ C ┘

Handoff:      A ⇄ B                            （全连通网,agent 自己跳)
              ⇅ ╳ ⇅
              C ⇄ D

GroupChat /   ┌─────编排者─────┐               （星形:所有话都过中心)
Magentic:     A      B       C
              └──────┴────────┘

下面逐个拆。每个都按"要解决什么 → 拓扑 → 关键源码 → 巧妙点"讲。

6. SequentialBuilder —— 链式,最简单的那个

要解决什么: 让几个 agent 按固定顺序接力,共享同一条对话。典型场景:草稿 agent → 审校 agent → 摘要 agent。

拓扑: 一条直链。开头加一个内部节点 _InputToConversation(_sequential.py:49)负责把各种输入(str / Message / list)归一化成 list[Message],然后逐个 add_edge 串起来。

连边逻辑短到可以直接看(_sequential.py:267):

# 示意,非源码:build() 尾部就是一个 for 循环把参与者串成链
prior = input_conv
for p in participants:          # participants 已被包成 AgentExecutor
    builder.add_edge(prior, p)
    prior = p

默认输出: 只有最后一个参与者的 yield_output 会被当成 workflow 的终态输出(default_output_from=[participants[-1]],_sequential.py:258)。

巧妙点 / 可配置项:

chain_only_agent_responses=True → 把上面讲的 context_mode 设成 "last_agent",链上只传上一个 agent 的回复而非全量历史(_sequential.py:203)。
.with_request_info(agents=[...]) → 开人在环路:每个 agent 说完暂停,发 request_info 事件让人审阅、可注入引导(_sequential.py:156)。开了这个的 agent 会被包成 AgentApprovalExecutor 而非普通 AgentExecutor。

7. ConcurrentBuilder —— 扇出并行,再聚合

要解决什么: 同一个问题,让多个 agent 同时从不同角度回答,最后汇总。典型场景:多专家并行会诊。

拓扑: 分发器 → 扇出到所有 agent → 扇入聚合器。两个内部节点:

_DispatchToAllParticipants(_concurrent.py:55):把输入原样广播给所有参与者(靠图的扇出边,不点名目标)。
_AggregateAgentConversations(_concurrent.py:83):等所有 agent 都回复后,从每个 agent 各取最后一条 assistant 消息,拼成一个 AgentResponse(aggregate handler,_concurrent.py:96)。

连边就是一次扇出加一次扇入(_concurrent.py:430):

# 示意,非源码:并行拓扑的两条关键连边
builder.add_fan_out_edges(dispatcher, participants)   # 一散多
builder.add_fan_in_edges(participants, aggregator)    # 多聚一

扇入边天生就是第 03 章讲的"barrier 屏障":聚合器要等齐所有上游才触发。这就是"并行后汇总"的语义来源,不用 Builder 自己写等待逻辑。

巧妙点: 聚合器可换。.with_aggregator(cb)(_concurrent.py:269)接受一个 Executor,或一个普通回调 (results) -> Any;回调会被 _CallbackAggregator(_concurrent.py:139)包起来,同步回调自动丢进线程池跑,避免阻塞事件循环。

8. HandoffBuilder —— 去中心化,agent 自己交接

要解决什么: 客服式路由。分诊 agent 判断后把对话交给退款 agent 或账单 agent,由后者自己再决定要不要转交。没有中央调度,谁接棒谁自己说了算。

这也是本章第一个"谁下一个说话"由 agent 自身决定的模式。它和群聊的分野,源码 docstring 说得很干脆(_handoff.py:20):

Group Chat : centralized orchestration of multiple agents（中心编排者拍板)
Handoff    : decentralized routing by agents themselves（agent 自己调工具跳转)

8.1 交接怎么实现:合成工具 + 中间件短路

这是整个 handoff 最巧的地方。agent 怎么"表达"要交接给谁?答案:给它装一把合成工具,再用中间件拦截。

流程(_handoff.py):

1. 克隆每个 agent,给它注入若干"交接工具" handoff_to_<目标>
       └ _apply_auto_tools（:303) / _create_handoff_tool（:332)
2. agent 想交接 → 它就"调用"某把 handoff_to_X 工具
3. _AutoHandoffMiddleware（:130) 拦截这次工具调用,不真跑函数体,
   而是短路返回一个合成结果 {"handoff_to": "X"}
       └ process()（:137）里 raise MiddlewareTermination
4. HandoffAgentExecutor 从 agent 回复的最后一条消息里读出这个结果,
   知道要跳去 X（_is_handoff_requested，:484)
5. 定向发消息给 X，并 add_event("handoff_sent")（:408-414)

工具函数体是空的、永不执行(_create_handoff_tool 里那个 _handoff_tool 就 pass),它纯粹是给 LLM 的一个"信号槽"。真正的路由逻辑全在中间件短路那一下。这个设计的好处:交接对 LLM 就是一次普通工具调用,不需要教模型任何新协议。

8.2 全连通拓扑 + 广播同步

handoff 的图是全连通的(build(),_handoff.py:1003):每个 agent 和其它所有 agent 都有边。为什么要全连通?因为一个 agent 说完话,要广播给所有其它 agent(用 should_respond=False 的信封,只更新它们的上下文、不让它们抢答),这样谁被交接到时,历史都是同步的(_broadcast_messages,_handoff.py:474)。

默认拓扑是 mesh(所有 agent 互相可交接,_resolve_handoffs 的 else 分支,_handoff.py:1061);也可以用 .add_handoff(source, [targets]) 精确指定路由图。

巧妙点 / 坑:

_run_agent_and_emit 被重写(_handoff.py:352),交接前会用 clean_conversation_for_handoff(_orchestrator_helpers.py:16)把对话里的工具调用/结果洗成纯文本——否则下一个 agent 收到不配对的 tool_call 状态,provider 会报错。
require_per_service_call_history_persistence=True 是 handoff 的硬性要求(build() 里会校验并抛错,_handoff.py:957)。原因正是中间件短路了工具调用,服务端根本没见过这些工具结果,本地历史必须单独持久化才不会对不上。
自主模式 .with_autonomous_mode():agent 回复后若没交接,默认会停下来等用户输入;开了自主模式则自动灌一句"继续"提示接着干,直到交接或到轮次上限(_handoff.py:428)。

9. GroupChatBuilder —— 中心编排者主导对话

要解决什么: 一群 agent 围坐讨论,由一个编排者决定每一轮谁发言、何时结束。典型场景:研究员 + 写手轮流,由协调者控节奏。

拓扑: 星形。编排者居中,和每个参与者建双向边(build(),_group_chat.py:1035):

# 示意,非源码:群聊就是编排者与每个参与者的双向连边
for participant in participants:
    builder.add_edge(orchestrator, participant)   # 派活
    builder.add_edge(participant, orchestrator)   # 交活

9.1 编排循环(编排者干的活)

编排者是 BaseGroupChatOrchestrator(_base_group_chat_orchestrator.py:130)的子类,它维护完整对话历史,循环做四步:

收到任务/上一个参与者的回复
   → 存进历史
   → 检查终止条件 / 轮次上限（到了就 yield 完成消息)
   → 选出下一个发言者
   → 广播给其它人 + 定向请求发言者回复
   → 轮次 +1，回到开头

选发言者这一步有两种实现,决定了两种"编排者":

编排者	怎么选下一个发言者	源码
`GroupChatOrchestrator`	你给的选择函数 `GroupChatState -> str`(可写轮询、条件路由等)	`_group_chat.py:98`,`_get_next_speaker`(`:237`)
`AgentBasedGroupChatOrchestrator`	一个 LLM agent 读对话,输出结构化的 `{terminate, next_speaker, ...}`	`_group_chat.py:284`,`_invoke_agent`(`:487`)

GroupChatBuilder 构造时三选一:selection_func / orchestrator_agent / 自定义 orchestrator(_group_chat.py:_set_orchestrator,:693)。

9.2 双信封:agent 和自定义节点区别对待

参与者可以是 agent,也可以是自定义 Executor。编排者派活时用两种信封(_send_request_to_participant,_base_group_chat_orchestrator.py:443):

是 agent → 发 AgentExecutorRequest(只给消息)。
是自定义 executor → 发 GroupChatRequestMessage(带完整上下文和额外指令)。

谁是 agent 由 ParticipantRegistry(_base_group_chat_orchestrator.py:86)在建图时登记。这层区分让"自定义逻辑节点"也能平等参与群聊。

10. MagenticBuilder —— 压轴:会自我纠错的自规划编排

Magentic 是最复杂、也最精彩的一个。它复刻了微软 Magentic-One 的编排范式:一个 LLM manager 先立计划,再逐轮监控进度,卡住了就重新规划。它复用了群聊的星形拓扑和 BaseGroupChatOrchestrator 基类,但把"选发言者"升级成了一套双台账机制。

先看它和普通群聊的差别在哪:

维度	普通 GroupChat	Magentic
开场	直接选人发言	先让 manager 立计划(事实 + 步骤)
每轮决策	选择函数/agent 只选下一个人	manager 产出进度台账:是否完成 / 是否死循环 / 是否有进展 / 下一个谁 / 给什么指令
卡住时	到轮次上限就硬停	检测到停滞 → 重置 + 重规划,再来一遍

10.1 两本台账(Magentic 的核心数据结构)

Magentic 的聪明之处,是把"协作状态"显式建模成两本可序列化的台账:

① 任务台账 _MagenticTaskLedger(_magentic.py:271)——开场时立好,只有两块:

facts:对任务做的"事实调查表"(已知事实 / 需查证 / 需推导 / 有根据的猜测)。用 ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT(_magentic.py:108)让 manager 填。
plan:基于团队构成和事实,列出的要点式计划(ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT,:137)。

② 进度台账 MagenticProgressLedger(_magentic.py:307)——每一轮都重算一次,五个字段:

字段	含义
`is_request_satisfied`	任务完成了吗?(完成就去合成最终答案)
`is_in_loop`	在原地打转吗?
`is_progress_being_made`	最近几轮有推进吗?
`next_speaker`	下一个该谁说话
`instruction_or_question`	给这个人的具体指令

manager 靠 ORCHESTRATOR_PROGRESS_LEDGER_PROMPT(_magentic.py:194)让 LLM 以纯 JSON 吐出这五项,create_progress_ledger(_magentic.py:699)带重试解析。

10.2 内外两层循环(编排的骨架)

MagenticOrchestrator(_magentic.py:864)的循环结构:

外层循环（规划阶段）
  └ plan()：立任务台账（事实+计划)
  └ 进入内层循环 ──────────────────────────┐
                                            │
内层循环（协调阶段，_run_inner_loop_helper) │
  每一轮:                                    │
    1. 查限额（轮次/重置上限,到了就终止)    │
    2. create_progress_ledger()：算进度台账  │
    3. 任务完成? → prepare_final_answer()，结束
    4. 停滞或死循环? → stall_count++          │
         stall_count 超上限 → 跳出去重规划 ──┘（reset + replan)
    5. 否则:把指令发给 next_speaker，等它回复后回到第 1 步

关键代码位置:

停滞计数与阈值判断:_magentic.py:1116(is_progress_being_made 为假或 is_in_loop 为真则 stall_count += 1,反之衰减)。
超阈值触发重置重规划:_reset_and_replan(_magentic.py:1156)——清空上下文、广播 MagenticResetSignal 让所有参与者重置、调 manager.replan() 用更新的事实和新计划重来。
参与者收到重置信号后清空自己的缓存和会话:MagenticAgentExecutor.handle_magentic_reset(_magentic.py:1355)。

10.3 重规划:失败后不硬撑,而是学到东西再来

这是 Magentic 最像"人"的一环。replan(_magentic.py:654)不是简单重跑,而是:

用 ..._FACTS_UPDATE_PROMPT(_magentic.py:169)让 manager 改写事实表——把新学到的东西加进去,甚至要求"至少更新一条有根据的猜测并解释理由"。
用 ..._PLAN_UPDATE_PROMPT(_magentic.py:186)让 manager 先解释上次为什么失败(根因),再给一份避免重蹈覆辙的新计划。

于是"卡住 → 反思根因 → 更新认知 → 换计划"成了一个闭环。MagenticContext.reset(_magentic.py:386)只清对话历史和停滞计数,保留任务、轮次、参与者,所以重来不是从零开始,而是带着教训重来。

10.4 Builder 与 manager 的分工

MagenticBuilder(_magentic.py:1381)负责建图(星形双向边,build() 于 :1772,和群聊同构);planning 逻辑全在 manager 里。manager 是可插拔的:

MagenticManagerBase(_magentic.py:469):抽象基类,四个抽象方法 plan / replan / create_progress_ledger / prepare_final_answer。
StandardMagenticManager(_magentic.py:514):默认实现,用一个 agent 做真实 LLM 调用。你也可以传 manager_agent=... 让 Builder 自动帮你造一个(_set_manager,:1606)。

还支持 enable_plan_review=True:manager 立完计划先暂停,发 MagenticPlanReviewRequest(_magentic.py:834)让人审批/改,批了才往下跑——这是第 04 章人在环路机制在 Magentic 里的具体落地。

11. 巧妙之处小结(可借鉴的技术)

"Builder = 图生成器"这一层抽象。五个模式共用一台引擎(第 03 章),差异被收敛到"连边形状"这一件事上。想加第六种拓扑?写个新 Builder 连不同的边即可,不碰引擎。
should_respond=False 当"只灌上下文不发言"的开关(_agent_executor.py:32)。交接和群聊的"广播同步"全靠它,没有它就得给引擎加特殊消息类型。
交接 = 合成工具 + 中间件短路(_handoff.py:132)。把"路由决策"伪装成一次普通工具调用,LLM 零学习成本,路由逻辑完全在框架侧。
Magentic 把协作状态显式建成两本台账(_magentic.py:271、:306)。"是否停滞""是否死循环"这些平时藏在提示词里的隐性判断,被拉出来做成结构化字段 + 可序列化对象,于是能检测、能重试、能 checkpoint、能让人审。
重规划带"根因反思"(_magentic.py:186)。失败不是重试同一件事,而是先让模型说清为什么失败再换招——这是把"从错误中学习"编码进了编排循环。

12. 边界与局限(诚实)

本章绝大多数代码只有 Python 实现。 编排包 agent_framework_orchestrations 是 Python 包;跨语言(.NET 等)是否有对等的五个 Builder,克隆里的这些文件看不出来,不要假设一一对应。
无限循环风险要自己防。 GroupChatOrchestrator 的 docstring 明说:若既不设 max_rounds 也不设 termination_condition,对话会永远跑下去(_group_chat.py:142)。Magentic 靠 max_stall_count/max_reset_count/max_round_count 兜底,但默认 max_round_count=None 也是无上限。
Magentic 的 JSON 解析是"尽力而为"。 进度台账依赖 LLM 吐出合法 JSON,源码里带重试和退避(create_progress_ledger,_magentic.py:699);群聊 agent 编排者甚至有针对"多个 JSON 拼在一起"的临时补丁(_parse_last_json_object,_group_chat.py:419,注释自称 stop-gap)。这些是对模型不稳定输出的防御,不是稳态设计,上游可能变。
handoff 对参与者类型要求最严。 必须是 Agent 实例(不能只是 SupportsAgentRun),因为要克隆、注入工具、加中间件(_handoff.py:569 的类校验)。其它 Builder 大多接受更宽的 SupportsAgentRun。
clean_conversation_for_handoff 的工具内容识别是粗粒度的。 源码里有 TODO(_orchestrator_helpers.py:39)承认"目前把任何非文本内容都当工具调用"是简化处理,未来要精细化。

13. 代码地图(导航索引)

按符号名 grep 最稳(行号会随上游漂移)。所有路径相对克隆根 microsoft/agent-framework。

主题	文件路径	符号名
agent 包成节点	`python/packages/core/agent_framework/_workflows/_agent_executor.py`	`AgentExecutor`、`AgentExecutorRequest`、`AgentExecutorResponse`、`AgentExecutorResponse.with_text`
图反包成 agent	`python/packages/core/agent_framework/_workflows/_agent.py`	`WorkflowAgent`、`_convert_workflow_events_to_agent_response`
子工作流嵌套	`python/packages/core/agent_framework/_workflows/_workflow_executor.py`	`WorkflowExecutor`、`SubWorkflowRequestMessage`、`_process_workflow_result`
agent id 解析	`python/packages/core/agent_framework/_workflows/_agent_utils.py`	`resolve_agent_id`
顺序链	`python/packages/orchestrations/agent_framework_orchestrations/_sequential.py`	`SequentialBuilder`、`_InputToConversation`
并行扇出/扇入	`python/packages/orchestrations/agent_framework_orchestrations/_concurrent.py`	`ConcurrentBuilder`、`_DispatchToAllParticipants`、`_AggregateAgentConversations`、`_CallbackAggregator`
去中心交接	`python/packages/orchestrations/agent_framework_orchestrations/_handoff.py`	`HandoffBuilder`、`HandoffAgentExecutor`、`_AutoHandoffMiddleware`、`get_handoff_tool_name`
群聊编排基类	`python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py`	`BaseGroupChatOrchestrator`、`ParticipantRegistry`、`_send_request_to_participant`
群聊 Builder/编排者	`python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py`	`GroupChatBuilder`、`GroupChatOrchestrator`、`AgentBasedGroupChatOrchestrator`、`GroupChatState`
Magentic	`python/packages/orchestrations/agent_framework_orchestrations/_magentic.py`	`MagenticBuilder`、`MagenticOrchestrator`、`StandardMagenticManager`、`MagenticManagerBase`、`_MagenticTaskLedger`、`MagenticProgressLedger`、`MagenticContext`
交接对话清洗	`python/packages/orchestrations/agent_framework_orchestrations/_orchestrator_helpers.py`	`clean_conversation_for_handoff`
公开导出清单	`python/packages/orchestrations/agent_framework_orchestrations/__init__.py`	`__all__`

上一章: 04 · 持久化、人在环路与时间旅行 — 本章的 .with_request_info / enable_plan_review / with_checkpointing 都建立在那一章的机制上。 回总览: index

1. 先搞清楚一件事:两个世界要打通​

2. 桥接件一:AgentExecutor —— 把 agent 包成节点​

2.1 两个标准信封​

2.2 无缝链接:三种输入都能接​

3. 桥接件二:WorkflowAgent —— 把整张图反过来当 agent​

4. 桥接件三:WorkflowExecutor —— 子工作流嵌套​

5. 编排包全景:五个 Builder,五种拓扑​

6. SequentialBuilder —— 链式,最简单的那个​

7. ConcurrentBuilder —— 扇出并行,再聚合​

8. HandoffBuilder —— 去中心化,agent 自己交接​

8.1 交接怎么实现:合成工具 + 中间件短路​

8.2 全连通拓扑 + 广播同步​

9. GroupChatBuilder —— 中心编排者主导对话​

9.1 编排循环(编排者干的活)​

9.2 双信封:agent 和自定义节点区别对待​

10. MagenticBuilder —— 压轴:会自我纠错的自规划编排​

10.1 两本台账(Magentic 的核心数据结构)​

10.2 内外两层循环(编排的骨架)​

10.3 重规划:失败后不硬撑,而是学到东西再来​

10.4 Builder 与 manager 的分工​

11. 巧妙之处小结(可借鉴的技术)​

12. 边界与局限(诚实)​

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