首页
/ AutoGen（Python）消息广播核心概念完全指南：Topic、Subscription 与 Type-Based Subscription 的深入解析

AutoGen（Python）消息广播核心概念完全指南：Topic、Subscription 与 Type-Based Subscription 的深入解析

2026-09-07 16:16:22作者：贡沫苏Truman
autogen
A programming framework for agentic AI
项目地址：https://gitcode.com/GitHub_Trending/au/autogen

本指南围绕仓库 python/docs/src/user-guide/core-user-guide/core-concepts/topic-and-subscription.md 展开，系统讲解 AutoGen Agent 运行时（agent runtime）中的广播消息模型——Topic（主题）与 Subscription（订阅）。读完本文，你将理解 Topic 的类型/来源二元结构与命名规则、Subscription 到 Agent 的映射机制、为何基于类型的订阅是推荐做法，以及如何将单租户单主题、单租户多主题、多租户等典型场景落地为可运行的订阅与发布代码。

两种消息投递方式：点对点（Direct Messaging）与广播（Broadcast）

AutoGen 的 Agent 运行时向 Agent 投递消息有两种方式，它们在运行时抽象 _agent_runtime.py中分别对应两类核心 API：

  • 直接消息（Direct Messaging）：一对一投递。发送方必须明确给出接收方的 Agent ID，对应运行时抽象中的 send_message 类接口；
  • 广播（Broadcast）：一对多投递。发送方不需要指定任何接收方 Agent ID，只需要指明广播到哪个 Topic，由运行时根据既有的订阅关系决定消息送到谁手上。
Topic 本质上是 Agent ID 之上的一层"间接寻址"（indirection）：广播让生产者和消费者完全解耦，发送方不感知、也不依赖接收方的具体身份。

许多场景天然适合广播。典型的是事件驱动的工作流：Agent 之间往往互不感知、互不依赖，谁会对某类消息感兴趣、谁会来消费，是由发布者所不知道的订阅关系动态决定的。因此，理解广播模型，核心就是理解 Topic（广播的边界） 与 Subscription（边界到 Agent 的映射） 这两个概念。

Topic：一次广播消息的作用范围

Topic 的二元构成

根据关联文档的定义：

Topic = (Topic Type, Topic Source)

一个 Topic 由两部分组成：

  • Topic Type（主题类型）：通常由应用代码定义，用来标记该主题承载的是哪一类消息。例如，一个处理 GitHub 事件的 Agent，在发布"新 issue"消息时可以用 "GitHub_Issues" 作为 topic type；
  • Topic Source（主题来源）：同一 topic type 下唯一标识一个 Topic 的值，通常由应用数据决定。例如该 GitHub Agent 可以用 "github.com/{repo_name}/issues/{issue_number}" 这样的字符串作为 topic source，从而唯一锁定"某一个仓库的某一个 issue"。

Topic Source 是发布者划定消息边界、制造数据隔离（silo）的关键手段：同一个 topic type 下不同的 source 之间天然互相隔离，互不串扰。

Topic 的字符串表示与命名约束

Topic ID 可以与字符串互相转换，格式为：

Topic_Type/Topic_Source

关联文档给出的可读性规则是：

  • Topic Type：必须是 UTF-8 文本，仅允许字母（a-z）、数字（0-9）和下划线（_）；不能以数字开头，不能包含任何空格；
  • Topic Source：必须是 UTF-8 文本，字符范围须落在 ASCII 32（空格）到 126（~）之间（含端点）。

在此基础上，Python 端实际的校验规则可以在 TopicId 实现中得到印证。autogen_core._topic.TopicId 是一个 @dataclass(eq=True, frozen=True)，其 type 字段在构造时用正则 ^[\w\-\.\:\=]+\Z 校验，命中正则之外的内容会抛出 ValueError。可以看到 Python 实现实际允许的字符比上文描述更宽（额外允许 -、.、:、=，且允许数字开头）。从字段 docstring 看，type 与 source 均对齐 CloudEvents 规范 的 type 与 source 语义。实际开发时以当前仓库实现为准。

同时，TopicId 提供两个双向转换方法：

str(TopicId(type="github_issues", source="github.com/example/autogen/issues/1"))
# "github_issues/github.com/example/autogen/issues/1"

TopicId.from_str("github_issues/github.com/example/autogen/issues/1")
# TopicId(type="github_issues", source="github.com/example/autogen/issues/1")

实现见 python/packages/autogen-core/src/autogen_core/_topic.py#L37-L47：__str__ 以 type/source 拼接，from_str 则用 split("/", maxsplit=1) 仅切分第一段，因此 source 内部即使包含 / 也不会破坏解析。

Subscription：把 Topic 映射到 Agent

如果说 Topic 回答"这条广播是关于什么的"，那么 Subscription 回答"这条广播应该送到谁"——它把 Topic 映射到 Agent ID：

Topic 与 Subscription 的关系示意

运行时维护着一张订阅表，并依据这张表把广播消息投递给各 Agent：

  • 某个 Topic 没有任何订阅：发布到该 Topic 的消息不会送达任何 Agent（消息对当前应用而言无消费者）；
  • 某个 Topic 有多个订阅：消息会依据全部匹配的订阅被投递，且同一接收方只会收到一次；
  • 应用可以通过运行时 API 动态添加或移除订阅。

在 Python 抽象层 _subscription.py中，Subscription 被定义为一个可运行时检查的协议（@runtime_checkable Protocol），它描述任何订阅实现都必须具备的契约：

成员 职责
id 订阅的唯一标识，通常是一个 UUID，用于去重与移除
is_match(topic_id) 判断某个 Topic 是否与该订阅匹配
map_to_agent(topic_id) 把匹配的 Topic 映射到应处理它的 AgentId；仅应在 is_match 返回 True 时调用，否则实现可抛 CantHandleException
__eq__ 默认按 id 判定两个订阅是否相等

运行时是如何维护与匹配订阅的

在单进程实现的内部，SubscriptionManager 承担订阅簿记工作，可以从源码看到三条关键逻辑：

  • add_subscription：若已存在相同订阅则抛 ValueError，否则追加并重建"Topic → 接收方 AgentId"索引（源码 L42-L48）；
  • remove_subscription(id)：按订阅 id 移除，不存在时抛 ValueError，移除后同样重建索引（源码 L50-L61）；
  • 懒构建索引：当一个从未见过的 Topic 第一次被查询时，_build_for_new_topic 会遍历全部订阅，凡 is_match(topic) 为真，就把 map_to_agent(topic) 的结果追加进该 Topic 的接收方列表（源码 L63-L78）。

也就是说，"多订阅命中同一 Topic、每个订阅各映射出一个接收方"的汇总正是在这里完成的。对应用开发者而言，只需面向运行时抽象 API 操作订阅即可，例如 AgentRuntime 接口中声明的 add_subscription(subscription) 与 remove_subscription(id)（见 python/packages/autogen-core/src/autogen_core/_agent_runtime.py#L268-L285）。

Type-Based Subscription：基于类型的订阅

上述 Subscription 只是协议，框架为绝大多数场景准备了一个推荐使用的具体实现：TypeSubscription（在 Python 端位于 autogen_core 顶层命名空间，即 from autogen_core import TypeSubscription，源码见 python/packages/autogen-core/src/autogen_core/_type_subscription.py）。

它的声明形式是：

Type-Based Subscription = Topic Type --> Agent Type

即把一个 topic type 映射到一个 agent type。它是一种"无界"的声明式映射——无需预先知道任何具体的 topic source 或 agent key：

  • 凡是 topic type 与该订阅匹配的 Topic，都会被映射到一个 Agent；
  • 这个 Agent 的类型取订阅声明的 agent type；
  • 这个 Agent 的 key（agent key）取该 Topic 的 source 值。

关联文档中还同步给出了相关概念 Agent ID 的说明（Agent ID 同样由 type 与 key 两部分组成，其字符串形态为 AgentType/Key，详见 Agent 标识与生命周期文档）。

从 TypeSubscription 的实现（python/packages/autogen-core/src/autogen_core/_type_subscription.py#L53-L60）可以清晰看到这一机制的两个核心方法：

def is_match(self, topic_id: TopicId) -> bool:
    return topic_id.type == self._topic_type

def map_to_agent(self, topic_id: TopicId) -> AgentId:
    if not self.is_match(topic_id):
        raise CantHandleException("TopicId does not match the subscription")
    return AgentId(type=self._agent_type, key=topic_id.source)
  • is_match 只比较 topic type，与 source 无关；
  • map_to_agent 直接构造 AgentId(type=订阅的agent_type, key=Topic的source)——topic source 原样下沉成为 agent key。

类文档字符串里的例子更直观地展示了这一点：TypeSubscription(topic_type="t1", agent_type="a1") 一旦注册，source 为 s1 的 Topic ("t1","s1") 会被 a1/s1 处理，source 为 s2 的 Topic ("t1","s2") 会被 a1/s2 处理——每个 source 都拥有自己独立的 Agent 实例，天然形成数据隔离。

细节：TypeSubscription.__init__(topic_type, agent_type, id=None) 中 id 缺省时自动生成 UUID；agent_type 同时接受字符串或 AgentType 对象。两个订阅的相等判定为"id 相同，或 agent_type 与 topic_type 都相同"（源码 L62-L66）。

为什么它更受推荐？因为它可移植且与具体数据解耦：开发者无需编写依赖特定 Agent ID 的应用代码，即使 topic source 千变万化，一条声明即可覆盖无穷多个具体 Topic。

典型应用场景：基于类型订阅的落地模式

把 TypeSubscription 用在实际工程中，通常会从两个维度去拆解需求：

  1. 是否多租户（single-tenant / multi-tenant）；
  2. 每个租户下是单个 Topic 还是多个 Topic（single / multiple topics per tenant）。

这里的"租户（tenant）"通常指负责某一特定用户会话或某一特定请求的一组 Agent。

场景一：单租户 + 单 Topic

这是最简单、也最常见的场景，适合命令行工具、单用户应用等只有一套会话的场合。做法是：

  • 为每一个 agent type 各创建一个 TypeSubscription；
  • 所有订阅使用同一个 topic type；
  • 发布时始终使用同一个 Topic（即相同的 topic type 与 topic source）。

延续关联文档的例子：假设有 "triage_agent"、"coder_agent"、"reviewer_agent" 三个 agent type，topic type 取 "default"，则订阅如下：

# Type-based Subscriptions for single-tenant, single topic scenario
TypeSubscription(topic_type="default", agent_type="triage_agent")
TypeSubscription(topic_type="default", agent_type="coder_agent")
TypeSubscription(topic_type="default", agent_type="reviewer_agent")

发布时所有消息都用 source "default"，于是 Topic 恒为 ("default", "default")。发到该 Topic 的消息会送达上面三种类型的全部 Agent，即产生以下 Agent ID：

# The agent IDs created based on the topic source
AgentID("triage_agent", "default")
AgentID("coder_agent", "default")
AgentID("reviewer_agent", "default")

单租户单 Topic 场景下基于类型订阅的工作方式

一种常见的落地形态是：三个 Agent 各自承担三阶段工作（分流 → 编码 → 审查），全部订阅同一 Topic，任何发布到 ("default", "default") 的消息都会以广播形式同时进入这三个 Agent 的消息流。若目标 Agent ID 尚不存在，运行时会为其创建实例（前提是该 agent type 已在运行时注册了相应的工厂与处理逻辑，相关约定见 Agent 标识与生命周期文档）。

场景二：单租户 + 多 Topic

当同一个应用里希望控制不同 Agent 各管一类 Topic、制造订阅层面的"分工与隔离"时，就用多 Topic 模式。做法是：

  • 为每个 agent type 创建订阅，但使用不同的 topic type；
  • 若希望若干 agent type 共享同一类 Topic，就把同一个 topic type 映射给多个 agent type；
  • topic source 在发布时依旧统一使用同一个值。

继续沿用三个 agent type 的例子：

# Type-based Subscriptions for single-tenant, multiple topics scenario
TypeSubscription(topic_type="triage", agent_type="triage_agent")
TypeSubscription(topic_type="coding", agent_type="coder_agent")
TypeSubscription(topic_type="coding", agent_type="reviewer_agent")

于是：

  • 发布到 ("triage", "default") 的消息 → 只投递给 triage_agent；
  • 发布到 ("coding", "default") 的消息 → 同时投递给 coder_agent 和 reviewer_agent。

单租户多 Topic 场景下基于类型订阅的工作方式

场景三：多租户场景

单租户场景里 topic source 是"写死在代码里"的常量（如 "default"）；一旦进入多租户场景，topic source 就变成了随数据而变的变量。

一个判断你是否处于多租户场景的可靠信号：你需要同一 agent type 的多个实例。例如用不同 Agent 实例分别处理不同用户会话以隔离私有数据，或让同一 agent type 的多个实例并行分担重负载。

以 GitHub issue 处理为例。假如只为 "triage_agent" 声明了订阅：

TypeSubscription(topic_type="github_issues", agent_type="triage_agent")

那么：

  • 消息发布到 Topic ("github_issues", "github.com/microsoft/autogen/issues/1") → 投递给 Agent ID ("triage_agent", "github.com/microsoft/autogen/issues/1")；
  • 消息发布到 Topic ("github_issues", "github.com/microsoft/autogen/issues/9") → 投递给 Agent ID ("triage_agent", "github.com/microsoft/autogen/issues/9")。

多租户场景下基于类型订阅的工作方式

这里的 Agent ID 完全由数据决定：不同 issue 的广播被路由给不同的 triage_agent 实例，运行时会为尚不存在的目标实例创建新 Agent，从而做到按 issue 粒度的处理隔离——这正是"topic source 下沉为 agent key"这一设计在多租户场景下的威力。

如果每个租户内部还想进一步细分（即"每个租户多个 Topic"），做法与单租户多 Topic 一致：使用不同的 topic type 即可。例如每个 issue 租户下同时有 "github_issues" 与 "github_discussions" 两个 topic type，分别路由给负责 issue 与 discussion 的 Agent。

发布与订阅：一个端到端的组合示例

把上述概念串起来，一次完整的广播消息消费大致长这样（方法签名与语义均取自当前仓库的 AgentRuntime 抽象与 TypeSubscription 实现）：

from autogen_core import AgentId, TopicId, TypeSubscription

# 1. 向运行时注册"类型化订阅"：github_issues 类消息一律交给 triage_agent
await runtime.add_subscription(
    TypeSubscription(topic_type="github_issues", agent_type="triage_agent")
)

# 2. 此后，向任何以 github_issues 为 type 的 Topic 广播
topic = TopicId(type="github_issues", source="github.com/microsoft/autogen/issues/1")
await runtime.publish_message(message, topic_id=topic, sender=AgentId("poller_agent", "default"))

publish_message 是无响应的广播（不期待任何接收方回复），其可传关键字参数还包括 sender、cancellation_token 与 message_id。运行时会遍历所有匹配订阅，把消息逐一送入由 TypeSubscription 推导出的目标 Agent。当业务不再需要某条订阅时，用注册时记录下来的订阅 id（默认为自动生成的 UUID）调用 runtime.remove_subscription(id) 即可摘除该路由。

上述四种场景可以归纳为一张速查表，方便在不同需求下快速选型：

场景 订阅策略 发布使用的 Topic 实际投递到的 Agent ID
单租户 + 单 Topic 每种 agent type 各一订阅，共用一个 topic type 固定，如 ("default", "default") 各 agent type 的 (agent_type, "default")
单租户 + 多 Topic 各 agent type 用不同 topic type（可多对一共享） 按类别切换 topic type，source 固定 (agent_type, "default")
多租户 每个要并发的 agent type 一条订阅 topic type 固定，source 取租户唯一标识 (agent_type, <tenant/数据标识>)
多租户 + 多 Topic 每租户按需注册多个 topic type 的订阅 同时切换 topic type 与租户 source (agent_type, <tenant/数据标识>)

进一步阅读

autogen
A programming framework for agentic AI
项目地址：https://gitcode.com/GitHub_Trending/au/autogen
登录后查看全文

相关内容推荐

1 AutoGen 消息路由设计解析：TopicId、订阅模型与 Well-Known Topic 类型2 如何用 AutoGen Core 的发布订阅消息实现多智能体顺序工作流？3 AutoGen 编程模型解析：以发布-订阅事件流驱动多智能体协作4 AutoGen 应用分层架构解析：Messaging Runtime、消息行为契约与 Multi-Agent 设计模式5 AutoGen 开源贡献指南：uv 本地 CI 检查、版本号规则与发布流程深度解析6 Autogen多租户聊天应用中的会话管理方案解析7 AutoGen .NET 框架入门：用事件驱动模型构建可扩展的多智能体 AI 系统8 AutoGen Services 架构：Agent Worker、Gateway 与跨语言分布式消息的底层实现9 AutoGen 中 Agent ID 与 Topic ID 规范详解：分布式 Agent 寻址与发布订阅的双标识符体系10 AutoGen（Python）包体系全景：Core、AgentChat、Extensions 与 Studio 的分层架构与选型指南
热门项目推荐
相关项目推荐

热门内容推荐

1 编程实践项目探索指南：从零构建技术能力体系2 技术解构式学习：从0到1构建你的编程知识体系3 构建自己的技术世界：build-your-own-x项目的实践探索指南4 解锁编程技能的实践之旅：从零构建你的技术世界5 技术实践探索：从零开始构建核心系统的实践指南6 亲手锻造技术引擎：从0到1构建核心系统的实践指南

最新内容推荐

OpenClaw Meta Provider 插件完整参考：Muse Spark 推理模型接入、认证与 Responses API 实战Web-Dev-For-Beginners 太空游戏第三部分实践：如何为 app.js 编写高质量注释并整理代码结构CPython 新特性解析：memoryview.cast 新增 order 参数，实现零拷贝 Fortran 连续视图计划模式上下文压缩蒸馏：oh-my-pi "批准并压缩上下文"流程的设计与实现LlamaIndex Tavily 研究工具（TavilyToolSpec）实战指南：让 Agent 拥有联网搜索与网页内容提取能力Backstage 数据库配置实战：从本地 PostgreSQL 到云上免密认证CopilotKit × AG2 的 Agentic Chat 端到端 QA 指南：聊天闭环、前端工具、渲染工具与 Agent 上下文全验证PyTorch Activation Sparsifier 实战指南：基于前向预钩子的激活稀疏化三阶段流程ECC Coding Standards 技能实战：跨项目 TypeScript/React 编码规范、不可变性与代码质量基线OpenMontage 构建性能优化：规避 Barrel File 导入，让 Remotion 合成器显著提速

项目优选

收起
kernelkernel
deepin linux kernel
C
33
18
ops-transformerops-transformer
本项目是CANN提供的transformer类大模型算子库，实现网络在NPU上加速计算。
C++
1.14 K
2.76 K
pytorchpytorch
作为 Ascend for PyTorch 社区的核心组件，TorchNPU 是昇腾专为 PyTorch 打造的深度学习适配插件，使 PyTorch 框架能够直接调用昇腾 NPU，为开发者提供昇腾 AI 处理器的超强算力。
Python
860
1.35 K
docsdocs
暂无描述
Markdown
899
5.83 K
ops-nnops-nn
本项目是CANN提供的神经网络类计算算子库，实现网络在NPU上加速计算。
C++
925
1.85 K
jiuwenswarmjiuwenswarm
JiuwenSwarm 是一款基于openJiuwen开发的智能AI Agent，它能够将大语言模型的强大能力，通过你日常使用的各类通讯应用，直接延伸至你的指尖。
Python
3.84 K
1.02 K
kernelkernel
openEuler内核是openEuler操作系统的核心，既是系统性能与稳定性的基石，也是连接处理器、设备与服务的桥梁。
C
533
601
cann-learning-hubcann-learning-hub
CANN 学习中心仓，支持在线互动运行、边学边练，提供教程、示例与优化方案，一站式助力昇腾开发者快速上手。
Jupyter Notebook
1.03 K
525
ops-mathops-math
本项目是CANN提供的数学类基础计算算子库，实现网络在NPU上加速计算。
C++
1.37 K
1.46 K
AscendNPU-IRAscendNPU-IR
AscendNPU-IR是基于MLIR（Multi-Level Intermediate Representation）构建的，面向昇腾亲和算子编译时使用的中间表示，提供昇腾完备表达能力，通过编译优化提升昇腾AI处理器计算效率，支持通过生态框架使能昇腾AI处理器与深度调优
C++
548
395