人工智能AI 应用AI Agent【免费下载链接】Tutorial-Codebase-KnowledgePocket Flow: Codebase to Tutorial项目地址https://gitcode.com/gh_mirrors/tu/Tutorial-Codebase-Knowledge点击查看免费下载本文是《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 向任何感兴趣的对象广播消息就像在公司公告栏上张贴通知一样。把博客示例细化一下整个协作流程是这样的ResearcherAgent 找到了关于 AutoGen Agents 的事实Researcher不再直接把消息发给Writer而是将事实**发布publish**到一个通用的 research-resultsTopic主题上WriterAgent 事先告诉系统它**订阅subscribed**了 research-results 这个 Topic系统发现 Topic 上出现了新消息就把它投递给Writer以及其他所有订阅者。这样一来Researcher不需要知道Writer是谁甚至不需要知道Writer是否存在它只需要广播结果。如果以后新增一个同样需要结果的FactCheckerAgent它只需订阅同一个 Topic 即可Researcher一行代码都不用改。核心概念Topic 与 Subscription 的四个组成部分1. TopicTopicId公告栏一个TopicId代表一个具体的消息频道或类别可以把它想象成公告栏的名称如 项目更新、综合通知。它由两个主要部分组成type这是什么种类的事件或信息例如research.completed、user.request。它用于对消息分类。source这个事件从哪来 / 因何而起通常关联到具体的任务或上下文例如正在研究的博客主题autogen-agents-blog-post或产生事件的团队research-team。# From: _topic.py (Simplified) from dataclasses import dataclass dataclass(frozenTrue) # Immutable: cant 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(frozenTrue)即创建后不可变的不可变对象。这种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.availablesource标识具体的研究任务例如blog-post-autogen。# From: _topic.py from autogen_core import TopicId # Define the topic for this specific research task research_topic_id TopicId(typeresearch.facts.available, sourceblog-post-autogen) print(fTopic 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(fResearcher working on: {message.topic}) facts_data ResearchFacts( topicmessage.topic, facts[fFact A about {message.topic}, fFact B about {message.topic}] ) # Define the specific topic for this tasks results results_topic TopicId(typeresearch.facts.available, sourcemessage.topic) # Use message topic as source # Publish the facts to the topic await agent_context.publish_message(messagefacts_data, topic_idresults_topic) print(fResearcher 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(fWriter ({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_typeresearch.facts.available, agent_typewriter # The type of agent that should handle this ) print(fWriter subscription created for topic type: {writer_subscription.topic_type}) # Output: Writer subscription created for topic type: research.facts.available当 Researcher 向TopicId(typeresearch.facts.available, sourceblog-post-autogen)发布消息时AgentRuntime会发现writer_subscription匹配了topic_type于是执行规则查找或创建一个AgentId(typewriter, keyblog-post-autogen)的 Agent 并投递消息。收益解耦DecouplingResearcher 只管广播Writer 只管监听相关广播。我们可以随时增加更多监听者比如让FactChecker也订阅同一个topic_type而完全不需要修改 Researcher 的任何代码。底层原理一次发布消息的完整旅程概念流程下图展示了一条已发布消息从发布到投递的完整旅程Mermaid 时序图对应到AgentRuntime的具体实现整个流程可以拆解为五步发布PublishAgent 调用agent_context.publish_message(message, topic_id)内部实际调用AgentRuntime的 publish 方法。发布调用通常是异步不阻塞的返回值为None这也是发布与send_message直发的重要区别直发会等待并返回对端的返回值。查找LookupAgentRuntime拿到topic_id后查询其内部的Subscription Registry订阅注册表。匹配Match注册表遍历所有已注册的Subscription对象。每个Subscription都有一个is_match(topic_id)方法注册表找出所有is_match返回True的订阅。映射Map对每个匹配的SubscriptionRuntime 调用其map_to_agent(topic_id)方法根据订阅规则和主题细节返回应该处理这条消息的具体AgentId。投递DeliverAgentRuntime根据返回的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(eqTrue, frozenTrue) 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 subscriptions 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.pySubscription协议最常见的实现它提供了特定主题类型对应一个按 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 topics 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 # topics source as the agents unique key. if not self.is_match(topic_id): raise CantHandleException(...) # Should not happen if used correctly return AgentId(typeself._agent_type, keytopic_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使用提供了一种便捷方式来创建TypeSubscriptionagent_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_typeresearch.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 应用的关键基础。TopicTopicId用于广播消息的具名频道type/source双字段Publish向某个 Topic 发送消息无需指定收件人SubscriptionAgent 对某些 Topic 上消息的兴趣声明本质是一条路由规则RoutingAgentRuntime借助订阅注册表完成is_match匹配 →map_to_agent映射 →on_message投递的完整链路。接下来可以继续阅读 第 3 章AgentRuntime了解负责创建、运行与连接 Agent 的编排者——它正是实现消息发布与订阅路由的引擎。完整的概念关系图Agent 生命周期管理、消息路由、LLM 客户端、Tool、Memory 等模块间的联系可参考 AutoGen Core 教程首页。赞分享人工智能AI 应用AI Agent【免费下载链接】Tutorial-Codebase-KnowledgePocket Flow: Codebase to Tutorial项目地址https://gitcode.com/gh_mirrors/tu/Tutorial-Codebase-Knowledge点击查看免费下载相关推荐AutoGenPython消息广播核心概念完全指南Topic、Subscription 与 Type-Based Subscription 的深入解析AutoGenPython消息广播核心概念完全指南Topic、Subscription 与 Type Based Subscription 的深入解析 本人工智能AI AgentAgent 框架多智能体大模型工具调用TDengine 原生数据订阅Native Subscription完整实战指南Topic 创建、Consumer 参数与消息消费TDengine 原生数据订阅Native Subscription完整实战指南Topic 创建、Consumer 参数与消息消费 TDengine TS数据库时序数据库大数据物联网云原生如何用 iii queue worker 以 durable:subscriber 订阅 topic 处理发布/订阅消息如何用 iii queue worker 以 durable:subscriber 订阅 topic 处理发布/订阅消息 当你需要多个独立消费者可靠地收到同一后端流程编排任务调度可观测性创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 AutoGen Core 消息系统深度解析：Topic 与 Subscription 发布订阅机制实战
人工智能AI 应用AI Agent【免费下载链接】Tutorial-Codebase-KnowledgePocket Flow: Codebase to Tutorial项目地址https://gitcode.com/gh_mirrors/tu/Tutorial-Codebase-Knowledge点击查看免费下载本文是《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 向任何感兴趣的对象广播消息就像在公司公告栏上张贴通知一样。把博客示例细化一下整个协作流程是这样的ResearcherAgent 找到了关于 AutoGen Agents 的事实Researcher不再直接把消息发给Writer而是将事实**发布publish**到一个通用的 research-resultsTopic主题上WriterAgent 事先告诉系统它**订阅subscribed**了 research-results 这个 Topic系统发现 Topic 上出现了新消息就把它投递给Writer以及其他所有订阅者。这样一来Researcher不需要知道Writer是谁甚至不需要知道Writer是否存在它只需要广播结果。如果以后新增一个同样需要结果的FactCheckerAgent它只需订阅同一个 Topic 即可Researcher一行代码都不用改。核心概念Topic 与 Subscription 的四个组成部分1. TopicTopicId公告栏一个TopicId代表一个具体的消息频道或类别可以把它想象成公告栏的名称如 项目更新、综合通知。它由两个主要部分组成type这是什么种类的事件或信息例如research.completed、user.request。它用于对消息分类。source这个事件从哪来 / 因何而起通常关联到具体的任务或上下文例如正在研究的博客主题autogen-agents-blog-post或产生事件的团队research-team。# From: _topic.py (Simplified) from dataclasses import dataclass dataclass(frozenTrue) # Immutable: cant 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(frozenTrue)即创建后不可变的不可变对象。这种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.availablesource标识具体的研究任务例如blog-post-autogen。# From: _topic.py from autogen_core import TopicId # Define the topic for this specific research task research_topic_id TopicId(typeresearch.facts.available, sourceblog-post-autogen) print(fTopic 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(fResearcher working on: {message.topic}) facts_data ResearchFacts( topicmessage.topic, facts[fFact A about {message.topic}, fFact B about {message.topic}] ) # Define the specific topic for this tasks results results_topic TopicId(typeresearch.facts.available, sourcemessage.topic) # Use message topic as source # Publish the facts to the topic await agent_context.publish_message(messagefacts_data, topic_idresults_topic) print(fResearcher 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(fWriter ({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_typeresearch.facts.available, agent_typewriter # The type of agent that should handle this ) print(fWriter subscription created for topic type: {writer_subscription.topic_type}) # Output: Writer subscription created for topic type: research.facts.available当 Researcher 向TopicId(typeresearch.facts.available, sourceblog-post-autogen)发布消息时AgentRuntime会发现writer_subscription匹配了topic_type于是执行规则查找或创建一个AgentId(typewriter, keyblog-post-autogen)的 Agent 并投递消息。收益解耦DecouplingResearcher 只管广播Writer 只管监听相关广播。我们可以随时增加更多监听者比如让FactChecker也订阅同一个topic_type而完全不需要修改 Researcher 的任何代码。底层原理一次发布消息的完整旅程概念流程下图展示了一条已发布消息从发布到投递的完整旅程Mermaid 时序图对应到AgentRuntime的具体实现整个流程可以拆解为五步发布PublishAgent 调用agent_context.publish_message(message, topic_id)内部实际调用AgentRuntime的 publish 方法。发布调用通常是异步不阻塞的返回值为None这也是发布与send_message直发的重要区别直发会等待并返回对端的返回值。查找LookupAgentRuntime拿到topic_id后查询其内部的Subscription Registry订阅注册表。匹配Match注册表遍历所有已注册的Subscription对象。每个Subscription都有一个is_match(topic_id)方法注册表找出所有is_match返回True的订阅。映射Map对每个匹配的SubscriptionRuntime 调用其map_to_agent(topic_id)方法根据订阅规则和主题细节返回应该处理这条消息的具体AgentId。投递DeliverAgentRuntime根据返回的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(eqTrue, frozenTrue) 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 subscriptions 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.pySubscription协议最常见的实现它提供了特定主题类型对应一个按 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 topics 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 # topics source as the agents unique key. if not self.is_match(topic_id): raise CantHandleException(...) # Should not happen if used correctly return AgentId(typeself._agent_type, keytopic_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使用提供了一种便捷方式来创建TypeSubscriptionagent_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_typeresearch.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 应用的关键基础。TopicTopicId用于广播消息的具名频道type/source双字段Publish向某个 Topic 发送消息无需指定收件人SubscriptionAgent 对某些 Topic 上消息的兴趣声明本质是一条路由规则RoutingAgentRuntime借助订阅注册表完成is_match匹配 →map_to_agent映射 →on_message投递的完整链路。接下来可以继续阅读 第 3 章AgentRuntime了解负责创建、运行与连接 Agent 的编排者——它正是实现消息发布与订阅路由的引擎。完整的概念关系图Agent 生命周期管理、消息路由、LLM 客户端、Tool、Memory 等模块间的联系可参考 AutoGen Core 教程首页。赞分享人工智能AI 应用AI Agent【免费下载链接】Tutorial-Codebase-KnowledgePocket Flow: Codebase to Tutorial项目地址https://gitcode.com/gh_mirrors/tu/Tutorial-Codebase-Knowledge点击查看免费下载相关推荐AutoGenPython消息广播核心概念完全指南Topic、Subscription 与 Type-Based Subscription 的深入解析AutoGenPython消息广播核心概念完全指南Topic、Subscription 与 Type Based Subscription 的深入解析 本人工智能AI AgentAgent 框架多智能体大模型工具调用TDengine 原生数据订阅Native Subscription完整实战指南Topic 创建、Consumer 参数与消息消费TDengine 原生数据订阅Native Subscription完整实战指南Topic 创建、Consumer 参数与消息消费 TDengine TS数据库时序数据库大数据物联网云原生如何用 iii queue worker 以 durable:subscriber 订阅 topic 处理发布/订阅消息如何用 iii queue worker 以 durable:subscriber 订阅 topic 处理发布/订阅消息 当你需要多个独立消费者可靠地收到同一后端流程编排任务调度可观测性创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考