企业官网建设流程全解析

如何用 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_key="YOUR_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_type=concept_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 = f"Product description: {message.content}" llm_result = await self._model_client.create( messages=[self._system_message, UserMessage(content=prompt, source=self.id.key)], cancellation_token=ctx.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_id=TopicId(writer_topic_type, source=self.id.key)) @type_subscription(topic_type=writer_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 = f"Below is the info about the product:\n\n{message.content}" llm_result = await self._model_client.create( messages=[self._system_message, UserMessage(content=prompt, source=self.id.key)], cancellation_token=ctx.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_id=TopicId(format_proof_topic_type, source=self.id.key)) @type_subscription(topic_type=format_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 = f"Draft copy:\n{message.content}." llm_result = await self._model_client.create( messages=[self._system_message, UserMessage(content=prompt, source=self.id.key)], cancellation_token=ctx.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_id=TopicId(user_topic_type, source=self.id.key)) @type_subscription(topic_type=user_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( model="gpt-4o-mini", # 如环境未设置 OPENAI_API_KEY，可改为 api_key="YOUR_API_KEY" ) runtime = SingleThreadedAgentRuntime() await ConceptExtractorAgent.register( runtime, type=concept_extractor_topic_type, factory=lambda: ConceptExtractorAgent(model_client=model_client) ) await WriterAgent.register(runtime, type=writer_topic_type, factory=lambda: WriterAgent(model_client=model_client)) await FormatProofAgent.register( runtime, type=format_proof_topic_type, factory=lambda: FormatProofAgent(model_client=model_client) ) await UserAgent.register(runtime, type=user_topic_type, factory=lambda: UserAgent()) runtime.start() # 向链条头（第一个智能体）订阅的主题发布首条消息，source 用 "default" await runtime.publish_message( Message(content="An eco-friendly stainless steel water bottle that keeps drinks cold for 24 hours"), topic_id=TopicId(concept_extractor_topic_type, source="default"), ) 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="..."))。
• 发布时source=self.id.key是维持"同一组实例"的关键：消息投到主题("WriterAgent", <key>)时，实际收件人是 agent key 等于该 source 的那个WriterAgent实例，运行时会在实例不存在时自动创建它。
• 首条消息从外部发布，source="default"，因此整条链路的实例 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 来区分"发给谁"，所以本文主路径使用的是显式的TopicId+@type_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