AutoGen 应用分层架构解析：Messaging Runtime、消息行为契约与 Multi-Agent 设计模式

AutoGen（本仓库为 au/autogen）的 Python 核心包 autogen_core 定位为一套**无观点（unopinionated）**的 Agent 编程框架：它不绑定任何具体的 Agent 抽象或多 Agent 协作模式，而是通过"运行时消息基础设施 + 开发者自定义消息类型（行为契约）"的分层设计，支撑从单进程脚本到跨语言分布式系统的各类多 Agent 应用。本文基于 application-stack.md，完整讲解应用分层的上下两端——底层 Agent Runtime 的消息与路由能力、顶层消息类型构成的行为契约，并用代码生成（Coder / Executor / Reviewer）示例串联 reflection（反思）这一经典多 Agent 模式，帮助读者理解并着手构建自己的多 Agent 应用。

为什么需要"分层"：Runtime 与行为契约的分离

AutoGen 的整套设计可以浓缩为一张应用分层图：

这张图表达了两个核心分层观点：

底层：消息与路由基础设施（Messaging & Routing Facilities） 位于栈底的是让 Agent 之间能够互相通信的基础消息与路由能力，它们由 Agent Runtime（Agent 运行时） 统一管理。大多数应用开发者无需接触底层细节，只与 Runtime 提供的高层 API 交互即可（详见 Agent and Agent Runtime）。

顶层：消息类型与行为契约（Behavior Contract / Message Protocol） 位于栈顶的是开发者自己定义的、Agent 之间互相交换的消息类型。这些消息类型的集合形成了一份 Agent 必须遵守的行为契约（Behavior Contract，也被称为消息协议 Message Protocol）。契约的具体实现（即每个 Agent 如何"处理"某类消息）决定了多 Agent 应用的协作逻辑，而这正是开发者的职责所在。多 Agent 设计模式正是从这些行为契约中涌现出来的（详见 Multi-Agent Design Patterns）。

一句话概括这一分层思想：Runtime 解决"消息如何送达"，行为契约解决"消息如何被理解与响应"——前者由框架负责，后者由你的应用逻辑负责。

底层基础设施：Agent Runtime 的消息与路由能力

"应用栈底层由 Agent Runtime 管理"并不是一句空话。在 Python 源码中，这一定义体现在 autogen_core/_agent_runtime.py 的 AgentRuntime Protocol 中，它定义了所有 Runtime 实现都必须提供的高层 API，其中最核心的两类消息能力是：

• send_message（直连消息）：点对点地把消息发送给指定 recipient（一个 AgentId），并等待并返回 Agent 的响应。调用方可选传入 sender、cancellation_token（取消令牌）与 message_id。若收件方无法处理该消息会抛出 CantHandleException，无法投递会抛出 UndeliverableException。
• publish_message（广播消息）：把消息发布到某个 TopicId 上，由 Runtime 根据订阅（Subscription） 关系投递给该主题下所有感兴趣的 Agent；广播不期待响应。

对应地，Agent Runtime 还提供 register_factory 等注册能力，用于把某个 Agent 类型与工厂函数绑定（代码注释中明确说明这是底层 API，通常应使用 Agent 类的 register 方法，因为它会同时处理订阅注册）。

在 Python API 中，最典型的单进程实现是 autogen_core.SingleThreadedAgentRuntime（见 architecture.md），它适合所有 Agent 运行在同一进程内的场景。对于跨进程、跨语言、跨机器的场景，AutoGen 还提供由 host servicer（宿主机服务端）与多个 worker（工作端）构成的分布式 Runtime，但其对 Agent 暴露的 API 与单机版一致，因此开发者可以在不修改任何 Agent 实现的前提下切换 Runtime 类型。

从源码结构看，autogen_core 通过 topic 与订阅机制（_topic.py、_subscription.py、_type_subscription.py、_type_prefix_subscription.py 等）实现了 publish-subscribe 的投递语义，而 _routed_agent.py 则提供了最常用的 Agent 基类 RoutedAgent——它根据消息类型（必要时叠加 match 谓词）把收到的消息自动路由到对应的方法处理器上。

Agent 侧的"契约实现"载体：RoutedAgent 与事件/RPC 处理器

行为契约最终落实为 Agent 如何处理某类消息。在源码层面，这份实现落在 autogen_core/_routed_agent.py 的 RoutedAgent 上。它的工作方式是：Agent 类上凡是带有 @event 或 @rpc 装饰器的 async 方法，都会在构造时通过 _discover_handlers() 被自动收集进 _handlers 字典；当消息到达时，on_message_impl() 会取出 type(message) 对应的处理器列表，按处理器名称的字母顺序依次调用其 router，命中第一个返回 True 的处理器执行（_routed_agent.py）。

两个装饰器对应两种消息语义：

• @event：声明"事件"型处理器。方法签名固定为 (self, message, ctx)，必须返回 None，且 ctx.is_rpc 必须为假（即只处理广播/事件类消息）。它适合响应型、无需应答的协作，例如记录状态、触发下一步流程。
• @rpc：声明"远程过程调用"型处理器。同样要求 async 方法，但可以返回响应消息，且只处理 ctx.is_rpc 为真的点对点请求（_routed_agent.py 的 rpc 实现会把 router 包装为 ctx.is_rpc and match(...)）。

这两个装饰器都会对消息入参类型与返回类型做严格校验（strict=True 时类型不匹配直接抛 CantHandleException/ValueError，strict=False 时降级为 logger.warning）。换句话说：Agent 能收什么消息、会回什么消息，在代码里就是显式、类型化的声明——这正是"行为契约"从设计层落到代码层的具体体现。

顶层行为契约：一个代码生成应用示例

为了让分层概念落地，原文档给出了一个具体的多 Agent 应用例子：代码生成应用（Code Generation Application）。它由三个 Agent 组成：

• Coder Agent（编码 Agent）：负责根据任务生成代码；
• Executor Agent（执行 Agent）：负责运行生成的代码；
• Reviewer Agent（评审 Agent）：负责评估执行结果，决定通过或打回。

三个 Agent 之间交换的消息与数据流如下：

在这个例子中，行为契约由五类消息构成：

消息类型	发送方	接收方	语义
CodingTaskMsg	应用（Application）	Coder Agent	下发编码任务
CodeGenMsg	Coder Agent	Executor Agent	携带生成的代码，请求执行
ExecutionResultMsg	Executor Agent	Reviewer Agent	报告代码执行结果
ReviewMsg	Reviewer Agent	Coder Agent	评审未通过，要求重新生成/修改
CodingResultMsg	Reviewer Agent	应用（Application）	评审通过，向应用返回最终结果

这份契约并不存在于任何框架内置代码中，而是由各 Agent 对消息的处理逻辑来实现的。以 Reviewer Agent 为例，其核心决策逻辑可描述为：

Reviewer Agent 监听 ExecutionResultMsg，评估代码执行结果并决定通过（approve）或拒绝（reject）。若通过，则向应用发送 CodingResultMsg 作为最终交付；若未通过，则向 Coder Agent 发送 ReviewMsg，触发新一轮代码生成。

把这段逻辑对应到 RoutedAgent 的实现方式，大致是给 Reviewer Agent 定义一个以 ExecutionResultMsg 为入参类型的处理器，并在处理器内根据执行结果分支：通过时 send_message(CodingResultMsg, application)，否则 send_message(ReviewMsg, coder_agent)。由于处理器入参/返回类型均被 @rpc/@event 严格声明，这套流转在编译/导入期就是自描述的——即所谓"行为契约"。

从契约到模式：Reflection（反思）与更多多 Agent 协作

上述这个 Coder–Executor–Reviewer 协作流程，正是多 Agent 设计模式中 reflection（反思） 的一个实例："让生成的结果经过另一轮生成来评审，从而提升整体质量"。该模式在仓库中对应两个实教学资源：

• Reflection 模式教程；
• 代码执行 + GroupChat 教程，其中也包含 Coder–Reviewer 之间的数据流图 coder-reviewer-data-flow.svg，可作为该示例的补充图示。

正如 Multi-Agent Design Patterns 所强调的：多 Agent 设计模式本质上是消息协议/行为契约的涌现结构。只要开发者用 AutoGen 定义好 Agent 与消息类型，就可以自由实现任意模式——包括但不限于：

• Reflection（反思/评审循环）：如上例，生成 + 评审 + 再生成的反馈环；
• Group Chat（群聊/任务分解）：多个 Agent 围绕共享话题协作，将复杂任务拆解处理；
• Sequential Workflow（顺序工作流）：消息沿固定链路的串行流转；
• Handoffs（交接）、Mixture of Agents、Multi-agent Debate 等，均可在 design-patterns 目录 中找到对应 notebook 教程。

从 agent-and-multi-agent-application.md 的定义看，Agent 是一种"通过消息通信、维护自身状态、并针对收到的消息或状态变化执行动作"的软件实体；多 Agent 应用中各 Agent 既可以在同一进程内协作，也可以跨机器、跨组织边界、跨语言实现，它们作为可独立开发/测试/部署、天然可组合的自包含单元，被复用到不同场景并装配成更复杂的系统。这与 application-stack 一文"Runtime 管通信、开发者管契约"的定位一脉相承。

动手实践路径与文档索引

要亲手验证上面的分层与契约概念，推荐按照仓库内文档的依赖顺序推进：

1. 理解 Agent 与多 Agent 应用的基本定义：Agent and Multi-Agent Applications；
2. 深入 Agent 标识、生命周期与 Runtime 环境：Agent Identity and Lifecycle、Agent Runtime Environments，以及 Agent and Agent Runtime；
3. 学习消息如何送达（直连 vs 广播）：Topic and Subscription 与 Message and Communication——这两篇与"应用栈底层"直接对应；
4. 跟随动手教程：Quickstart 之后，实现一个 Coder–Reviewer 式的 reflection 应用来落实本文的行为契约示例。

如果想要在源码层面确认本文所述的分层机制，可按以下路径查阅：

• Runtime 高层 API 定义：autogen_core/_agent_runtime.py（AgentRuntime Protocol、send_message、publish_message）；
• 单进程 Runtime 实现：autogen_core/_single_threaded_agent_runtime.py；
• 消息处理与契约声明：autogen_core/_routed_agent.py（message_handler、event、rpc 装饰器）与 RoutedAgent 类定义（同文件 #L415 起）；
• 广播与订阅模型：autogen_core/_topic.py、autogen_core/_subscription.py、autogen_core/_type_subscription.py。

小结

AutoGen 应用分层架构的精髓在于清晰的职责切分：底层由 Agent Runtime 统一承载消息路由、Agent 生命周期与订阅投递等通用机制，保证上层应用的可移植与可伸缩（单进程 ↔ 分布式切换不修改 Agent 代码）；顶层则由开发者以类型化的消息集（行为契约） 定义 Agent 间的交互协议，并借由 RoutedAgent + @event/@rpc 将协议落实为确定的处理逻辑。二者结合，任何多 Agent 协作模式——包括示例中的 reflection——都可以像"定义消息 + 实现处理器 + 交给 Runtime"一样自然地构造出来。