顺序工作流#
顺序工作流是一种多智能体设计模式，其中智能体按照确定性的顺序响应。工作流中的每个智能体通过处理消息、生成响应，然后将其传递给下一个智能体来执行特定任务。该模式适用于创建确定性工作流，其中每个智能体负责完成预先指定的子任务。
在本示例中，我们演示了一个顺序工作流，多个智能体协作将基础产品描述转化为精美的营销文案。
该流水线包含四个专业智能体：
概念提取智能体：分析初始产品描述以提取关键特征、目标受众和独特卖点（USPs）。输出为结构化分析结果，以单一文本块呈现。
文案撰写智能体：根据提取的概念创作引人入胜的营销文案。该智能体将分析洞察转化为吸引人的促销内容，以单一文本块提供连贯的叙述。
格式校对智能体：通过优化语法、提升清晰度和保持一致的语调来润色文案草稿。该智能体确保专业质量并输出格式规范的最终版本。
用户交互智能体：向用户展示最终润色后的营销文案，完成整个工作流。
下图展示了本示例中的顺序工作流：
我们将使用发布-订阅消息机制实现此工作流。 请阅读主题与订阅了解核心概念， 以及广播消息了解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! 🌟🥤