顺序工作流#
顺序工作流是一种多智能体设计模式，其中智能体以确定性序列响应。工作流中的每个智能体通过处理消息、生成响应，然后将其传递给下一个智能体来执行特定任务。此模式适用于创建确定性工作流，其中每个智能体都为预先指定的子任务做出贡献。
在此示例中，我们演示了一个顺序工作流，其中多个智能体协作将基本产品描述转换为精美的营销文案。
该管道由四个专门的智能体组成
概念提取器智能体：分析初始产品描述以提取关键特征、目标受众和独特卖点（USP）。输出是一个结构化的分析，在一个文本块中。
文案智能体：根据提取的概念撰写引人注目的营销文案。此智能体将分析洞察转化为引人入胜的宣传内容，在一个文本块中提供连贯的叙述。
格式与校对智能体：通过改进语法、增强清晰度并保持一致的语气来润色草稿文案。此智能体确保专业质量并提供格式良好的最终版本。
用户智能体：向用户呈现最终的、精炼的营销文案，完成工作流。
下图说明了此示例中的顺序工作流
我们将使用发布-订阅消息机制来实现此工作流。请阅读主题和订阅以了解核心概念，并阅读广播消息以了解 API 用法。
在此管道中，智能体通过将其完成的工作作为消息发布到序列中下一个智能体的主题来相互通信。例如，当ConceptExtractor完成产品描述分析时，它将其发现发布到"WriterAgent"主题，而WriterAgent已订阅此主题。这种模式贯穿管道的每个步骤，每个智能体都发布到序列中下一个智能体已订阅的主题。
from dataclasses import dataclass
from autogen_core import (
MessageContext,
RoutedAgent,
SingleThreadedAgentRuntime,
TopicId,
TypeSubscription,
message_handler,
type_subscription,
)
from autogen_core.models import ChatCompletionClient, SystemMessage, UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient
消息协议#
此示例工作流的消息协议是智能体将用于转发其工作的简单文本消息。
@dataclass
class Message:
content: str
主题#
工作流中的每个智能体都将订阅特定的主题类型。主题类型以序列中的智能体命名，这允许每个智能体将其工作发布到序列中的下一个智能体。
concept_extractor_topic_type = "ConceptExtractorAgent"
writer_topic_type = "WriterAgent"
format_proof_topic_type = "FormatProofAgent"
user_topic_type = "User"
代理#
每个智能体类都使用type_subscription装饰器定义，以指定其订阅的主题类型。除了装饰器，您还可以使用add_subscription()方法直接通过运行时订阅主题。
概念提取器智能体为产品描述提供初始要点。
@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}")
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}")
工作流程#
现在我们可以将智能体注册到运行时。由于我们使用了type_subscription装饰器，运行时将自动将智能体订阅到正确的主题。
model_client = OpenAIChatCompletionClient(
model="gpt-4o-mini",
# 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()
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()
--------------------------------------------------------------------------------
ConceptExtractorAgent:
**Key Features:**
- Made from eco-friendly stainless steel
- Can keep drinks cold for up to 24 hours
- Durable and reusable design
- Lightweight and portable
- BPA-free and non-toxic materials
- Sleek, modern aesthetic available in various colors
**Target Audience:**
- Environmentally conscious consumers
- Health and fitness enthusiasts
- Outdoor adventurers (hikers, campers, etc.)
- Urban dwellers looking for sustainable alternatives
- Individuals seeking stylish and functional drinkware
**Unique Selling Points:**
- Eco-friendly design minimizes plastic waste and supports sustainability
- Superior insulation technology that maintains cold temperatures for a full day
- Durable construction ensures long-lasting use, offering a great return on investment
- Attractive design that caters to fashion-forward individuals
- Versatile use for both everyday hydration and outdoor activities
--------------------------------------------------------------------------------
WriterAgent:
🌍🌿 Stay Hydrated, Stay Sustainable! 🌿🌍
Introducing our eco-friendly stainless steel drinkware, the perfect companion for the environmentally conscious and style-savvy individuals. With superior insulation technology, our bottles keep your beverages cold for an impressive 24 hours—ideal for hiking, camping, or just tackling a busy day in the city. Made from lightweight, BPA-free materials, this durable and reusable design not only helps reduce plastic waste but also ensures you’re making a responsible choice for our planet.
Available in a sleek, modern aesthetic with various colors to match your personality, this drinkware isn't just functional—it’s fashionable! Whether you’re hitting the trails or navigating urban life, equip yourself with a stylish hydration solution that supports your active and sustainable lifestyle. Join the movement today and make a positive impact without compromising on style! 🌟🥤
--------------------------------------------------------------------------------
FormatProofAgent:
🌍🌿 Stay Hydrated, Stay Sustainable! 🌿🌍
Introducing our eco-friendly stainless steel drinkware—the perfect companion for environmentally conscious and style-savvy individuals. With superior insulation technology, our bottles keep your beverages cold for an impressive 24 hours, making them ideal for hiking, camping, or simply tackling a busy day in the city. Crafted from lightweight, BPA-free materials, this durable and reusable design not only helps reduce plastic waste but also ensures that you’re making a responsible choice for our planet.
Our drinkware features a sleek, modern aesthetic available in a variety of colors to suit your personality. It’s not just functional; it’s also fashionable! Whether you’re exploring the trails or navigating urban life, equip yourself with a stylish hydration solution that supports your active and sustainable lifestyle. Join the movement today and make a positive impact without compromising on style! 🌟🥤
--------------------------------------------------------------------------------
User received final copy:
🌍🌿 Stay Hydrated, Stay Sustainable! 🌿🌍
Introducing our eco-friendly stainless steel drinkware—the perfect companion for environmentally conscious and style-savvy individuals. With superior insulation technology, our bottles keep your beverages cold for an impressive 24 hours, making them ideal for hiking, camping, or simply tackling a busy day in the city. Crafted from lightweight, BPA-free materials, this durable and reusable design not only helps reduce plastic waste but also ensures that you’re making a responsible choice for our planet.
Our drinkware features a sleek, modern aesthetic available in a variety of colors to suit your personality. It’s not just functional; it’s also fashionable! Whether you’re exploring the trails or navigating urban life, equip yourself with a stylish hydration solution that supports your active and sustainable lifestyle. Join the movement today and make a positive impact without compromising on style! 🌟🥤