顺序工作流#
顺序工作流是一种多代理设计模式，其中代理以确定性的顺序响应。工作流中的每个代理通过处理消息、生成响应，然后将其传递给下一个代理来执行特定任务。此模式对于创建确定性工作流非常有用，其中每个代理都为一个预先指定的子任务做出贡献。
在本例中，我们展示了一个顺序工作流程，其中多个代理协作将基本产品描述转化为精美的营销文案。
该管道由四个专门的代理组成：
概念提取代理：分析初始产品描述以提取关键特征、目标受众和独特卖点（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()
--------------------------------------------------------------------------------
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! 🌟🥤