企业官网建设流程全解析

• 人工智能
• AI 应用
• AI Agent

本文是《Tutorial-Codebase-Knowledge》仓库中 AutoGen Core 系列教程 的第 2 章，主题是 AutoGen Core 的消息系统（Messaging System）——即TopicId、Subscription与发布（Publish）/订阅（Subscribe）路由机制。在 第 1 章：Agent 中我们了解了 Agent 是独立的工作单元，但多个 Agent 之间如何在不互相知道对方存在的前提下协调配合？答案就是本文的发布订阅模型。读完本文，你将掌握TopicId的type/source双字段结构、Subscription协议（is_match/map_to_agent）、TypeSubscription与@default_subscription装饰器的用法，并理解消息发布后从注册表匹配到目标 Agent 投递的完整底层链路。

说明：本仓库是一个"代码库转教程"的 AI 项目（参见 README.md 与 系统设计文档），docs/AutoGen Core/下的章节正是由该流水线从 microsoft/autogen 的 autogen-core 源码自动分析生成的。因此文中代码注释标注的_topic.py、_subscription.py、_type_subscription.py等均指上游 autogen-core 源码文件；而本文对应章节、第 1 章 Agent 与 第 3 章 AgentRuntime 则是本仓库内可直接查阅的配套文档。

为什么需要消息系统：广播信息的动机

在第 1 章中，我们把 Agent 视作独立的"员工"。但一个关键问题是：当 Researcher（研究员）发现了事实后，它并不知道谁需要这些信息。也许 Writer（撰稿人）需要，也许以后还会加入 Fact-Checker（事实核查）或 Summary（摘要）Agent。Researcher 如何做到"我宣布：事实在这里！"而不需要一个固定的收件人清单？

这就是消息系统（Messaging System），具体来说是Topics（主题）与Subscriptions（订阅）要解决的问题：它允许 Agent 向任何感兴趣的对象广播消息，就像在公司公告栏上张贴通知一样。

把博客示例细化一下，整个协作流程是这样的：

1. ResearcherAgent 找到了关于 "AutoGen Agents" 的事实；
2. Researcher不再直接把消息发给Writer，而是将事实**发布（publish）**到一个通用的 "research-results"Topic（主题）上；
3. WriterAgent 事先告诉系统它**订阅（subscribed）**了 "research-results" 这个 Topic；
4. 系统发现 Topic 上出现了新消息，就把它投递给Writer（以及其他所有订阅者）。

这样一来，Researcher不需要知道Writer是谁，甚至不需要知道Writer是否存在！它只需要广播结果。如果以后新增一个同样需要结果的FactCheckerAgent，它只需订阅同一个 Topic 即可，Researcher一行代码都不用改。

核心概念：Topic 与 Subscription 的四个组成部分

1. Topic（TopicId）：公告栏

一个TopicId代表一个具体的消息频道或类别，可以把它想象成公告栏的名称（如 "项目更新"、"综合通知"）。它由两个主要部分组成：

• type：这是什么种类的事件或信息？（例如"research.completed"、"user.request"）。它用于对消息分类。
• source：这个事件从哪来 / 因何而起？通常关联到具体的任务或上下文（例如正在研究的博客主题"autogen-agents-blog-post"，或产生事件的团队"research-team"）。

# From: _topic.py (Simplified) from dataclasses import dataclass @dataclass(frozen=True) # Immutable: can't change after creation class TopicId: type: str source: str def __str__(self) -> str: # Creates an id like "research.completed/autogen-agents-blog-post" return f"{self.type}/{self.source}"

注意这里使用了@dataclass(frozen=True)，即创建后不可变的不可变对象。这种type/source结构为灵活的过滤提供了可能：Agent 可以订阅某个type下的全部主题（无论source是什么），也可以只订阅特定source的主题。此外，TopicId还提供了from_str类方法，用于把"type/source"这样的字符串反向解析成TopicId对象，方便从消息头等序列化形式还原主题。

2. Publishing：张贴公告

当 Agent 有信息要广泛分享时，它会向特定的TopicId发布（publish）一条消息。这就像把一张便条钉到指定的公告栏上——发布者不需要知道谁会来读它。

3. Subscription：登记订阅意向

一个Subscription是 Agent 声明自己对某些TopicId感兴趣的方式。它就像一条规则："如果有一条消息被发布到匹配这个模式的 Topic，请把它投递给这类Agent"。Subscription把TopicId模式（例如 "所有 type 为research.completed的主题"）与一个AgentId（或确定AgentId的方式）关联起来。

4. Routing：投递邮件

AgentRuntime（我们将在 第 3 章：AgentRuntime 中遇到的系统管理者）负责跟踪所有激活的Subscription。当消息被发布到某个TopicId时，AgentRuntime检查哪些Subscription与该TopicId匹配；对每个匹配项，它根据该Subscription的规则计算出应该由哪个具体的AgentId接收消息并完成投递。

实战用例：Researcher 发布，Writer 订阅

下面完整复现 Researcher 与 Writer 的协作场景。

目标：Researcher 把事实发布到一个 Topic 上，Writer 通过订阅接收这些事实。

第 1 步：定义 Topic

我们需要一个承载研究结果的TopicId。设type为"research.facts.available"，source标识具体的研究任务（例如"blog-post-autogen"）。

# From: _topic.py from autogen_core import TopicId # Define the topic for this specific research task research_topic_id = TopicId(type="research.facts.available", source="blog-post-autogen") print(f"Topic ID: {research_topic_id}") # Output: Topic ID: research.facts.available/blog-post-autogen

这就定义好了我们要使用的"公告栏"。

第 2 步：Researcher 发布消息

ResearcherAgent 在找到事实后，通过运行时提供的agent_context把ResearchFacts消息发布到该主题。注意：agent_context.publish_message调用中，Researcher 没有指定任何收件人，只指定了主题。

# Simplified concept - Researcher agent logic # Assume 'agent_context' and 'message' (ResearchTopic) are provided # Define the facts message (from Chapter 1) @dataclass class ResearchFacts: topic: str facts: list[str] async def researcher_publish_logic(agent_context, message: ResearchTopic, msg_context): print(f"Researcher working on: {message.topic}") facts_data = ResearchFacts( topic=message.topic, facts=[f"Fact A about {message.topic}", f"Fact B about {message.topic}"] ) # Define the specific topic for this task's results results_topic = TopicId(type="research.facts.available", source=message.topic) # Use message topic as source # Publish the facts to the topic await agent_context.publish_message(message=facts_data, topic_id=results_topic) print(f"Researcher published facts to topic: {results_topic}") # No direct reply needed return None

补充一点：如果 Agent 继承自BaseAgent（参见 第 1 章 与 第 3 章），也可以直接调用self.publish_message(...)便捷方法，它内部同样是借助 runtime 完成发布，效果等价。

第 3 步：Writer 订阅

WriterAgent 需要告诉系统自己对"research.facts.available"这类主题感兴趣。这里使用预定义的订阅类型TypeSubscription。它的语义是："我对所有精确匹配这个 type的主题感兴趣。当消息到达时，创建/复用我的 type的 Agent，其key与主题的source一致。"

# From: _type_subscription.py (Simplified Concept) from autogen_core import TypeSubscription, BaseAgent class WriterAgent(BaseAgent): # ... agent implementation ... async def on_message_impl(self, message: ResearchFacts, ctx): # This method gets called when a subscribed message arrives print(f"Writer ({self.id}) received facts via subscription: {message.facts}") # ... process facts and write draft ... # How the Writer subscribes (usually done during runtime setup - Chapter 3) # This tells the runtime: "Messages on topics with type 'research.facts.available' # should go to a 'writer' agent whose key matches the topic source." writer_subscription = TypeSubscription( topic_type="research.facts.available", agent_type="writer" # The type of agent that should handle this ) print(f"Writer subscription created for topic type: {writer_subscription.topic_type}") # Output: Writer subscription created for topic type: research.facts.available

当 Researcher 向TopicId(type="research.facts.available", source="blog-post-autogen")发布消息时，AgentRuntime会发现writer_subscription匹配了topic_type，于是执行规则："查找（或创建）一个AgentId(type='writer', key='blog-post-autogen')的 Agent 并投递消息"。

收益：解耦（Decoupling）！Researcher 只管广播，Writer 只管监听相关广播。我们可以随时增加更多监听者（比如让FactChecker也订阅同一个topic_type），而完全不需要修改 Researcher 的任何代码。

底层原理：一次发布消息的完整旅程

概念流程

下图展示了一条已发布消息从发布到投递的完整旅程（Mermaid 时序图）：

对应到AgentRuntime的具体实现，整个流程可以拆解为五步：

1. 发布（Publish）：Agent 调用agent_context.publish_message(message, topic_id)，内部实际调用AgentRuntime的 publish 方法。发布调用通常是异步不阻塞的，返回值为None（这也是发布与send_message直发的重要区别：直发会等待并返回对端的返回值）。
2. 查找（Lookup）：AgentRuntime拿到topic_id后，查询其内部的Subscription Registry（订阅注册表）。
3. 匹配（Match）：注册表遍历所有已注册的Subscription对象。每个Subscription都有一个is_match(topic_id)方法，注册表找出所有is_match返回True的订阅。
4. 映射（Map）：对每个匹配的Subscription，Runtime 调用其map_to_agent(topic_id)方法，根据订阅规则和主题细节返回应该处理这条消息的具体AgentId。
5. 投递（Deliver）：AgentRuntime根据返回的AgentId找到对应的 Agent 实例（如果尚不存在则创建，尤其是TypeSubscription场景下），然后调用该 Agent 的on_message方法，把原始发布的message投递进去。订阅者收到的MessageContext中会携带topic_id字段（在 第 3 章 的WriterAgent示例中通过ctx.topic_id打印出来），订阅者可以据此知道消息来自哪个主题。

代码剖析：消息系统的四个关键实现

TopicId（_topic.py）：如前所示，一个持有type和source的简单 dataclass，并包含校验逻辑以确保type符合一定的命名约定。同时提供from_str类方法作为"type/source"字符串解析的辅助工具。

# From: _topic.py @dataclass(eq=True, frozen=True) class TopicId: type: str source: str # ... validation and __str__ ... @classmethod def from_str(cls, topic_id: str) -> Self: # Helper to parse "type/source" string # ... implementation ...

Subscription协议（_subscription.py）：定义了任何订阅规则都必须遵守的契约（contract）。

# From: _subscription.py (Simplified Protocol) from typing import Protocol # ... other imports class Subscription(Protocol): @property def id(self) -> str: ... # Unique ID for this subscription instance def is_match(self, topic_id: TopicId) -> bool: """Check if a topic matches this subscription's rule.""" ... def map_to_agent(self, topic_id: TopicId) -> AgentId: """Determine the target AgentId if is_match was True.""" ...

任何实现了这三个成员的类都可以充当订阅规则：id为每个订阅实例提供唯一标识（用于注册与去重）；is_match决定"这条主题我是否感兴趣"；map_to_agent在匹配成功时决定"消息该送给谁"。

TypeSubscription（_type_subscription.py）：Subscription协议最常见的实现，它提供了"特定主题类型对应一个按 source 区分的 Agent 实例"的行为。

# From: _type_subscription.py (Simplified) class TypeSubscription(Subscription): def __init__(self, topic_type: str, agent_type: str, ...): self._topic_type = topic_type self._agent_type = agent_type # ... generates a unique self._id ... def is_match(self, topic_id: TopicId) -> bool: # Matches if the topic's type is exactly the one we want return topic_id.type == self._topic_type def map_to_agent(self, topic_id: TopicId) -> AgentId: # Maps to an agent of the specified type, using the # topic's source as the agent's unique key. if not self.is_match(topic_id): raise CantHandleException(...) # Should not happen if used correctly return AgentId(type=self._agent_type, key=topic_id.source) # ... id property ...

注意两个值得留意的细节：

• is_match是精确匹配topic_id.type == self._topic_type，不做前缀或模糊匹配；
• map_to_agent用主题的source作为目标 Agent 的key，即"每个 source 一个 Agent 实例"。这意味着向research.facts.available/blog-post-autogen与research.facts.available/another-post两个主题发布，会分别路由到writer/blog-post-autogen与writer/another-post两个独立的 Writer 实例。若想了解更多AgentId中type与key的含义，可回看 第 1 章。

DefaultSubscription（_default_subscription.py）：通常通过装饰器@default_subscription使用，提供了一种便捷方式来创建TypeSubscription：agent_type从被装饰的 Agent 类自动推断，topic_type默认为"default"（但可以覆盖）。它简化了最常见的订阅场景。

# From: _default_subscription.py (Conceptual Usage) from autogen_core import BaseAgent, default_subscription, ResearchFacts @default_subscription # Uses 'default' topic type, infers agent type 'writer' class WriterAgent(BaseAgent): # Agent logic here... async def on_message_impl(self, message: ResearchFacts, ctx): ... # Or specify the topic type @default_subscription(topic_type="research.facts.available") class SpecificWriterAgent(BaseAgent): # Agent logic here... async def on_message_impl(self, message: ResearchFacts, ctx): ...

实际的消息发送（publish_message）与路由逻辑位于AgentRuntime内部。在 第 3 章：AgentRuntime 中可以看到完整的运行时实现：SingleThreadedAgentRuntime内部通过_message_queue队列与后台任务处理PublishMessageEnvelope，再交给SubscriptionManager.get_subscribed_recipients(topic_id)遍历_subscriptions列表，对每个is_match通过的订阅调用map_to_agent得到收件人列表，最后逐个定位/创建 Agent 并调用其on_message。订阅通过runtime.add_subscription(subscription)注册，这通常发生在运行时设置阶段（参见 第 3 章 中完整的可运行示例与预期输出）。

总结与下一步

AutoGen Core 使用发布/订阅系统（TopicId、Subscription）让 Agent 之间无需直接耦合即可通信，这是构建灵活、可扩展的多 Agent 应用的关键基础。

• Topic（TopicId）：用于广播消息的具名频道（type/source双字段）；
• Publish：向某个 Topic 发送消息，无需指定收件人；
• Subscription：Agent 对某些 Topic 上消息的兴趣声明，本质是一条路由规则；
• Routing：AgentRuntime借助订阅注册表完成is_match匹配 →map_to_agent映射 →on_message投递的完整链路。

接下来，可以继续阅读 第 3 章：AgentRuntime，了解负责创建、运行与连接 Agent 的编排者——它正是实现消息发布与订阅路由的引擎。完整的概念关系图（Agent 生命周期管理、消息路由、LLM 客户端、Tool、Memory 等模块间的联系）可参考 AutoGen Core 教程首页。

• 人工智能
• AI 应用
• AI Agent