如何用 AutoGen Core 的发布订阅消息实现多智能体顺序工作流【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen你有一串任务必须按固定顺序执行——比如先提取产品卖点再写营销文案然后校对润色最后交付用户——每个智能体只做一件事做完把结果交给下一位。AutoGen Core 的 Sequential Workflow 设计模式 就是为这种场景准备的不写任何调用下一个智能体的胶水代码而是让每个智能体把产物发布到下一个智能体订阅的主题topic上由运行时Agent Runtime负责投递从而串起整条流水线。本文以一个官方文档中的四智能体营销文案流水线为例讲清如何用发布订阅消息搭出这条顺序工作流定义消息协议、声明每个智能体订阅的主题、把前一步的输出发布到后一步的主题、启动运行时并验证控制台输出。发布订阅如何驱动顺序流程先理解两个概念代码里的命名都从它们来。细节见 Topic and Subscription。Topic由两部分组成Topic (Topic Type, Topic Source)。Topic Type 通常由应用代码定义标记消息的类型Topic Source 是该类型下某个具体主题的标识常由数据决定用来限定消息的作用范围、形成信息孤岛silos。Type-based subscription类型订阅把 Topic Type 映射到 Agent Type任何匹配该 Topic Type 的主题都会投递给agent key 等于 topic source的那个智能体实例。顺序工作流正是靠它实现发给下一位——每个智能体声明自己订阅的主题类型然后把完成的工作发布到下一个智能体的主题类型上且source沿用self.id.key保证消息始终落在同一组实例上。顺序工作流 notebook 用的具体机制是每个智能体用一个以下一位命名的 Topic Type 接收输入处理完后publish_message到再下一个的主题类型。链头由你从外部发布第一条消息触发链条自然逐级推进直到最后一个智能体只输出结果、不再发布。准备条件按 Installation 的要求安装依赖需要 Python 3.10 或更高版本pip install autogen-core pip install autogen-ext[openai]第二条是为模型客户端OpenAIChatCompletionClient准备扩展。如果环境中没有设置OPENAI_API_KEY环境变量需要在创建客户端时传入api_key参数官方 notebook 中对应的注释形式是# api_keyYOUR_API_KEY。编写顺序工作流代码完整可运行的脚本如下结构来自 Sequential Workflow 官方示例消息协议、四个主题类型、四个智能体、注册并启动运行时。import asyncio from dataclasses import dataclass from autogen_core import ( MessageContext, RoutedAgent, SingleThreadedAgentRuntime, TopicId, message_handler, type_subscription, ) from autogen_core.models import ChatCompletionClient, SystemMessage, UserMessage from autogen_ext.models.openai import OpenAIChatCompletionClient # 消息协议智能体之间传递的纯数据 dataclass class Message: content: str # 主题类型每个智能体订阅一个以自己命名的 topic type # 并把处理结果发布到下一个智能体的 topic type concept_extractor_topic_type ConceptExtractorAgent writer_topic_type WriterAgent format_proof_topic_type FormatProofAgent user_topic_type User type_subscription(topic_typeconcept_extractor_topic_type) class ConceptExtractorAgent(RoutedAgent): def __init__(self, model_client: ChatCompletionClient) - None: super().__init__(A concept extractor agent.) self._system_message SystemMessage( content( You are a marketing analyst. Given a product description, identify:\n - Key features\n - Target audience\n - Unique selling points\n\n ) ) self._model_client model_client message_handler async def handle_user_description(self, message: Message, ctx: MessageContext) - None: prompt fProduct description: {message.content} llm_result await self._model_client.create( messages[self._system_message, UserMessage(contentprompt, sourceself.id.key)], cancellation_tokenctx.cancellation_token, ) response llm_result.content assert isinstance(response, str) print(f{-*80}\n{self.id.type}:\n{response}) # 发布到 WriterAgent 订阅的主题source 沿用当前实例 key await self.publish_message(Message(response), topic_idTopicId(writer_topic_type, sourceself.id.key)) type_subscription(topic_typewriter_topic_type) class WriterAgent(RoutedAgent): def __init__(self, model_client: ChatCompletionClient) - None: super().__init__(A writer agent.) self._system_message SystemMessage( content( You are a marketing copywriter. Given a block of text describing features, audience, and USPs, compose a compelling marketing copy (like a newsletter section) that highlights these points. Output should be short (around 150 words), output just the copy as a single text block. ) ) self._model_client model_client message_handler async def handle_intermediate_text(self, message: Message, ctx: MessageContext) - None: prompt fBelow is the info about the product:\n\n{message.content} llm_result await self._model_client.create( messages[self._system_message, UserMessage(contentprompt, sourceself.id.key)], cancellation_tokenctx.cancellation_token, ) response llm_result.content assert isinstance(response, str) print(f{-*80}\n{self.id.type}:\n{response}) await self.publish_message(Message(response), topic_idTopicId(format_proof_topic_type, sourceself.id.key)) type_subscription(topic_typeformat_proof_topic_type) class FormatProofAgent(RoutedAgent): def __init__(self, model_client: ChatCompletionClient) - None: super().__init__(A format proof agent.) self._system_message SystemMessage( content( You are an editor. Given the draft copy, correct grammar, improve clarity, ensure consistent tone, give format and make it polished. Output the final improved copy as a single text block. ) ) self._model_client model_client message_handler async def handle_intermediate_text(self, message: Message, ctx: MessageContext) - None: prompt fDraft copy:\n{message.content}. llm_result await self._model_client.create( messages[self._system_message, UserMessage(contentprompt, sourceself.id.key)], cancellation_tokenctx.cancellation_token, ) response llm_result.content assert isinstance(response, str) print(f{-*80}\n{self.id.type}:\n{response}) await self.publish_message(Message(response), topic_idTopicId(user_topic_type, sourceself.id.key)) type_subscription(topic_typeuser_topic_type) class UserAgent(RoutedAgent): def __init__(self) - None: super().__init__(A user agent that outputs the final copy to the user.) message_handler async def handle_final_copy(self, message: Message, ctx: MessageContext) - None: print(f\n{-*80}\n{self.id.type} received final copy:\n{message.content}) async def main() - None: model_client OpenAIChatCompletionClient( modelgpt-4o-mini, # 如环境未设置 OPENAI_API_KEY可改为 api_keyYOUR_API_KEY ) runtime SingleThreadedAgentRuntime() await ConceptExtractorAgent.register( runtime, typeconcept_extractor_topic_type, factorylambda: ConceptExtractorAgent(model_clientmodel_client) ) await WriterAgent.register(runtime, typewriter_topic_type, factorylambda: WriterAgent(model_clientmodel_client)) await FormatProofAgent.register( runtime, typeformat_proof_topic_type, factorylambda: FormatProofAgent(model_clientmodel_client) ) await UserAgent.register(runtime, typeuser_topic_type, factorylambda: UserAgent()) runtime.start() # 向链条头第一个智能体订阅的主题发布首条消息source 用 default await runtime.publish_message( Message(contentAn eco-friendly stainless steel water bottle that keeps drinks cold for 24 hours), topic_idTopicId(concept_extractor_topic_type, sourcedefault), ) await runtime.stop_when_idle() await model_client.close() asyncio.run(main())几个关键点值得注意每个智能体类上的type_subscription(topic_type...)装饰器声明了它订阅的主题类型。用装饰器时register阶段运行时会自动把对应的TypeSubscription注册进去不需要再手写订阅代码。不想用装饰器时也可以用运行时 API 显式声明await runtime.add_subscription(TypeSubscription(topic_type..., agent_type...))。发布时sourceself.id.key是维持同一组实例的关键消息投到主题(WriterAgent, key)时实际收件人是 agent key 等于该 source 的那个WriterAgent实例运行时会在实例不存在时自动创建它。首条消息从外部发布sourcedefault因此整条链路的实例 key 都是default。代码中全部使用了await所以按 Quick Start 的提示在非 Jupyter 环境如 VSCode 脚本中需要用async def main()包裹并用asyncio.run(main())运行上面脚本已包含这一处理。运行并判断结果直接运行脚本观察控制台输出。官方 notebook 中的示例输出文档示例文案内容会因模型而异是分四个段落、以 80 个-分隔的打印-------------------------------------------------------------------------------- ConceptExtractorAgent: **Key Features:** - Made from eco-friendly stainless steel - Can keep drinks cold for up to 24 hours ... -------------------------------------------------------------------------------- WriterAgent: Stay Hydrated, Stay Sustainable! ... -------------------------------------------------------------------------------- FormatProofAgent: ... -------------------------------------------------------------------------------- User received final copy: ...可以对照的判断点是四个主题类型依次出现ConceptExtractorAgent→WriterAgent→FormatProofAgent→User每段内容都是上一段的加工结果且以User received final copy:收尾。所有消息处理完后await runtime.stop_when_idle()返回程序退出说明整条链已跑完。链断掉或行为异常时检查什么结合 Topic and Subscription 和 Message and Communication 中的说明顺序工作流有几条文档明确给出的边界恰好也是最常见的断链原因某一级没有输出如果某个 topic 没有任何订阅发布到该 topic 的消息不会投递给任何智能体链会在这里静默终止。检查相邻两级代码里的 topic type 字符串是否拼写一致上一级publish_message的TopicId(type...)必须等于下一级type_subscription(topic_type...)的值。Topic Type 的命名有约束只允许字母、数字和下划线不能以数字开头、不能含空格Topic Source 允许 ASCII 32–126 的字符。用中文或带空格的 topic type 会不合法。广播是单向的发布订阅不能用于请求/响应即使某智能体的 handler 返回了值该返回值也会被丢弃。所以链的推进只能靠处理完再发布下一条不能靠返回值串联。自己发布、自己订阅不会收到如果某智能体发布的消息类型正是它自己订阅的它不会收到自己发布的消息这是运行时为防止死循环做的处理。下游 handler 抛异常不会传回发布者某智能体在处理发布消息时抛出的异常会被记录日志但不会传播回发布方。所以上游正常、下游不输出时去看运行时的日志而不是期待异常被抛出。如果希望所有智能体共享同一个发布/订阅范围单主题、单作用域文档给出了简化写法DefaultTopicId()与default_subscription但顺序工作流需要每一级用不同的 topic type 来区分发给谁所以本文主路径使用的是显式的TopicIdtype_subscription。下一步搭通这条链之后可以按需调整修改各智能体的系统提示词替换任务示例中的四个 prompt 分别对应分析、写作、校对、交付把UserAgent中的print换成存库、发邮件等动作——官方文档说明这在实际应用中是可以直接替换的。涉及多租户同一 agent type 需要多个实例并行处理不同会话时topic source 需要改成数据相关的唯一标识这部分用法在 Topic and Subscription 文档的 Multi-Tenant 一节有完整示例。【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
如何用 AutoGen Core 的发布订阅消息实现多智能体顺序工作流？
如何用 AutoGen Core 的发布订阅消息实现多智能体顺序工作流【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen你有一串任务必须按固定顺序执行——比如先提取产品卖点再写营销文案然后校对润色最后交付用户——每个智能体只做一件事做完把结果交给下一位。AutoGen Core 的 Sequential Workflow 设计模式 就是为这种场景准备的不写任何调用下一个智能体的胶水代码而是让每个智能体把产物发布到下一个智能体订阅的主题topic上由运行时Agent Runtime负责投递从而串起整条流水线。本文以一个官方文档中的四智能体营销文案流水线为例讲清如何用发布订阅消息搭出这条顺序工作流定义消息协议、声明每个智能体订阅的主题、把前一步的输出发布到后一步的主题、启动运行时并验证控制台输出。发布订阅如何驱动顺序流程先理解两个概念代码里的命名都从它们来。细节见 Topic and Subscription。Topic由两部分组成Topic (Topic Type, Topic Source)。Topic Type 通常由应用代码定义标记消息的类型Topic Source 是该类型下某个具体主题的标识常由数据决定用来限定消息的作用范围、形成信息孤岛silos。Type-based subscription类型订阅把 Topic Type 映射到 Agent Type任何匹配该 Topic Type 的主题都会投递给agent key 等于 topic source的那个智能体实例。顺序工作流正是靠它实现发给下一位——每个智能体声明自己订阅的主题类型然后把完成的工作发布到下一个智能体的主题类型上且source沿用self.id.key保证消息始终落在同一组实例上。顺序工作流 notebook 用的具体机制是每个智能体用一个以下一位命名的 Topic Type 接收输入处理完后publish_message到再下一个的主题类型。链头由你从外部发布第一条消息触发链条自然逐级推进直到最后一个智能体只输出结果、不再发布。准备条件按 Installation 的要求安装依赖需要 Python 3.10 或更高版本pip install autogen-core pip install autogen-ext[openai]第二条是为模型客户端OpenAIChatCompletionClient准备扩展。如果环境中没有设置OPENAI_API_KEY环境变量需要在创建客户端时传入api_key参数官方 notebook 中对应的注释形式是# api_keyYOUR_API_KEY。编写顺序工作流代码完整可运行的脚本如下结构来自 Sequential Workflow 官方示例消息协议、四个主题类型、四个智能体、注册并启动运行时。import asyncio from dataclasses import dataclass from autogen_core import ( MessageContext, RoutedAgent, SingleThreadedAgentRuntime, TopicId, message_handler, type_subscription, ) from autogen_core.models import ChatCompletionClient, SystemMessage, UserMessage from autogen_ext.models.openai import OpenAIChatCompletionClient # 消息协议智能体之间传递的纯数据 dataclass class Message: content: str # 主题类型每个智能体订阅一个以自己命名的 topic type # 并把处理结果发布到下一个智能体的 topic type concept_extractor_topic_type ConceptExtractorAgent writer_topic_type WriterAgent format_proof_topic_type FormatProofAgent user_topic_type User type_subscription(topic_typeconcept_extractor_topic_type) class ConceptExtractorAgent(RoutedAgent): def __init__(self, model_client: ChatCompletionClient) - None: super().__init__(A concept extractor agent.) self._system_message SystemMessage( content( You are a marketing analyst. Given a product description, identify:\n - Key features\n - Target audience\n - Unique selling points\n\n ) ) self._model_client model_client message_handler async def handle_user_description(self, message: Message, ctx: MessageContext) - None: prompt fProduct description: {message.content} llm_result await self._model_client.create( messages[self._system_message, UserMessage(contentprompt, sourceself.id.key)], cancellation_tokenctx.cancellation_token, ) response llm_result.content assert isinstance(response, str) print(f{-*80}\n{self.id.type}:\n{response}) # 发布到 WriterAgent 订阅的主题source 沿用当前实例 key await self.publish_message(Message(response), topic_idTopicId(writer_topic_type, sourceself.id.key)) type_subscription(topic_typewriter_topic_type) class WriterAgent(RoutedAgent): def __init__(self, model_client: ChatCompletionClient) - None: super().__init__(A writer agent.) self._system_message SystemMessage( content( You are a marketing copywriter. Given a block of text describing features, audience, and USPs, compose a compelling marketing copy (like a newsletter section) that highlights these points. Output should be short (around 150 words), output just the copy as a single text block. ) ) self._model_client model_client message_handler async def handle_intermediate_text(self, message: Message, ctx: MessageContext) - None: prompt fBelow is the info about the product:\n\n{message.content} llm_result await self._model_client.create( messages[self._system_message, UserMessage(contentprompt, sourceself.id.key)], cancellation_tokenctx.cancellation_token, ) response llm_result.content assert isinstance(response, str) print(f{-*80}\n{self.id.type}:\n{response}) await self.publish_message(Message(response), topic_idTopicId(format_proof_topic_type, sourceself.id.key)) type_subscription(topic_typeformat_proof_topic_type) class FormatProofAgent(RoutedAgent): def __init__(self, model_client: ChatCompletionClient) - None: super().__init__(A format proof agent.) self._system_message SystemMessage( content( You are an editor. Given the draft copy, correct grammar, improve clarity, ensure consistent tone, give format and make it polished. Output the final improved copy as a single text block. ) ) self._model_client model_client message_handler async def handle_intermediate_text(self, message: Message, ctx: MessageContext) - None: prompt fDraft copy:\n{message.content}. llm_result await self._model_client.create( messages[self._system_message, UserMessage(contentprompt, sourceself.id.key)], cancellation_tokenctx.cancellation_token, ) response llm_result.content assert isinstance(response, str) print(f{-*80}\n{self.id.type}:\n{response}) await self.publish_message(Message(response), topic_idTopicId(user_topic_type, sourceself.id.key)) type_subscription(topic_typeuser_topic_type) class UserAgent(RoutedAgent): def __init__(self) - None: super().__init__(A user agent that outputs the final copy to the user.) message_handler async def handle_final_copy(self, message: Message, ctx: MessageContext) - None: print(f\n{-*80}\n{self.id.type} received final copy:\n{message.content}) async def main() - None: model_client OpenAIChatCompletionClient( modelgpt-4o-mini, # 如环境未设置 OPENAI_API_KEY可改为 api_keyYOUR_API_KEY ) runtime SingleThreadedAgentRuntime() await ConceptExtractorAgent.register( runtime, typeconcept_extractor_topic_type, factorylambda: ConceptExtractorAgent(model_clientmodel_client) ) await WriterAgent.register(runtime, typewriter_topic_type, factorylambda: WriterAgent(model_clientmodel_client)) await FormatProofAgent.register( runtime, typeformat_proof_topic_type, factorylambda: FormatProofAgent(model_clientmodel_client) ) await UserAgent.register(runtime, typeuser_topic_type, factorylambda: UserAgent()) runtime.start() # 向链条头第一个智能体订阅的主题发布首条消息source 用 default await runtime.publish_message( Message(contentAn eco-friendly stainless steel water bottle that keeps drinks cold for 24 hours), topic_idTopicId(concept_extractor_topic_type, sourcedefault), ) await runtime.stop_when_idle() await model_client.close() asyncio.run(main())几个关键点值得注意每个智能体类上的type_subscription(topic_type...)装饰器声明了它订阅的主题类型。用装饰器时register阶段运行时会自动把对应的TypeSubscription注册进去不需要再手写订阅代码。不想用装饰器时也可以用运行时 API 显式声明await runtime.add_subscription(TypeSubscription(topic_type..., agent_type...))。发布时sourceself.id.key是维持同一组实例的关键消息投到主题(WriterAgent, key)时实际收件人是 agent key 等于该 source 的那个WriterAgent实例运行时会在实例不存在时自动创建它。首条消息从外部发布sourcedefault因此整条链路的实例 key 都是default。代码中全部使用了await所以按 Quick Start 的提示在非 Jupyter 环境如 VSCode 脚本中需要用async def main()包裹并用asyncio.run(main())运行上面脚本已包含这一处理。运行并判断结果直接运行脚本观察控制台输出。官方 notebook 中的示例输出文档示例文案内容会因模型而异是分四个段落、以 80 个-分隔的打印-------------------------------------------------------------------------------- ConceptExtractorAgent: **Key Features:** - Made from eco-friendly stainless steel - Can keep drinks cold for up to 24 hours ... -------------------------------------------------------------------------------- WriterAgent: Stay Hydrated, Stay Sustainable! ... -------------------------------------------------------------------------------- FormatProofAgent: ... -------------------------------------------------------------------------------- User received final copy: ...可以对照的判断点是四个主题类型依次出现ConceptExtractorAgent→WriterAgent→FormatProofAgent→User每段内容都是上一段的加工结果且以User received final copy:收尾。所有消息处理完后await runtime.stop_when_idle()返回程序退出说明整条链已跑完。链断掉或行为异常时检查什么结合 Topic and Subscription 和 Message and Communication 中的说明顺序工作流有几条文档明确给出的边界恰好也是最常见的断链原因某一级没有输出如果某个 topic 没有任何订阅发布到该 topic 的消息不会投递给任何智能体链会在这里静默终止。检查相邻两级代码里的 topic type 字符串是否拼写一致上一级publish_message的TopicId(type...)必须等于下一级type_subscription(topic_type...)的值。Topic Type 的命名有约束只允许字母、数字和下划线不能以数字开头、不能含空格Topic Source 允许 ASCII 32–126 的字符。用中文或带空格的 topic type 会不合法。广播是单向的发布订阅不能用于请求/响应即使某智能体的 handler 返回了值该返回值也会被丢弃。所以链的推进只能靠处理完再发布下一条不能靠返回值串联。自己发布、自己订阅不会收到如果某智能体发布的消息类型正是它自己订阅的它不会收到自己发布的消息这是运行时为防止死循环做的处理。下游 handler 抛异常不会传回发布者某智能体在处理发布消息时抛出的异常会被记录日志但不会传播回发布方。所以上游正常、下游不输出时去看运行时的日志而不是期待异常被抛出。如果希望所有智能体共享同一个发布/订阅范围单主题、单作用域文档给出了简化写法DefaultTopicId()与default_subscription但顺序工作流需要每一级用不同的 topic type 来区分发给谁所以本文主路径使用的是显式的TopicIdtype_subscription。下一步搭通这条链之后可以按需调整修改各智能体的系统提示词替换任务示例中的四个 prompt 分别对应分析、写作、校对、交付把UserAgent中的print换成存库、发邮件等动作——官方文档说明这在实际应用中是可以直接替换的。涉及多租户同一 agent type 需要多个实例并行处理不同会话时topic source 需要改成数据相关的唯一标识这部分用法在 Topic and Subscription 文档的 Multi-Tenant 一节有完整示例。【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
