顺序工作流#
顺序工作流是一种多 Agent 设计模式，其中 Agent 以确定性的顺序响应。 工作流中的每个 Agent 通过处理消息、生成响应，然后将其传递给下一个 Agent 来执行特定任务。 此模式对于创建确定性工作流非常有用，其中每个 Agent 都有助于预先指定的子任务。
在此示例中，我们将演示一个顺序工作流，其中多个 Agent 协同工作，将基本产品描述转换为精美的营销文案。
该管道由四个专业的 Agent 组成
概念提取 Agent：分析初始产品描述以提取关键功能、目标受众和独特的卖点 (USP)。 输出是单个文本块中的结构化分析。
编写 Agent：根据提取的概念制作引人注目的营销文案。 此 Agent 将分析见解转化为引人入胜的促销内容，并在单个文本块中提供连贯的叙述。
格式和校对 Agent：通过改进语法、增强清晰度和保持一致的语气来润色草稿副本。 此 Agent 确保专业质量并提供格式良好的最终版本。
用户 Agent：向用户展示最终的、精美的营销文案，完成工作流程。
下图说明了此示例中的顺序工作流程
我们将使用发布-订阅消息传递来实现此工作流程。 有关核心概念，请阅读主题和订阅，有关 API 用法，请阅读 广播消息传递。
在此管道中，Agent 通过将其完成的工作作为消息发布到序列中下一个 Agent 的主题来相互通信。 例如，当 ConceptExtractor 完成产品描述的分析时，它会将其发现发布到 "WriterAgent" 主题，WriterAgent 订阅了该主题。 这种模式在管道的每个步骤中持续存在，每个 Agent 都发布到序列中下一个 Agent 订阅的主题。
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
消息协议#
此示例工作流程的消息协议是一个简单的文本消息，Agent 将使用该消息来传递他们的工作。
@dataclass
class Message:
content: str
主题#
工作流程中的每个 Agent 都将订阅特定的主题类型。 主题类型以序列中 Agent 的名称命名， 这允许每个 Agent 将其工作发布到序列中的下一个 Agent。
concept_extractor_topic_type = "ConceptExtractorAgent"
writer_topic_type = "WriterAgent"
format_proof_topic_type = "FormatProofAgent"
user_topic_type = "User"
Agent#
每个 Agent 类都使用 type_subscription 装饰器来指定它订阅的主题类型。 除了装饰器，您还可以使用 add_subscription() 方法直接通过运行时订阅主题。
概念提取 Agent 提出了产品描述的初始要点。
@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))
编写 Agent 执行编写。
@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))
格式校对 Agent 执行格式化。
@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))
在此示例中，用户 Agent 只是将最终的营销文案打印到控制台。 在实际应用中，这可以替换为将结果存储到数据库、发送电子邮件或任何其他所需的操作。
@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}")
工作流程#
现在我们可以将 Agent 注册到运行时。 因为我们使用了 type_subscription 装饰器，所以运行时会自动将 Agent 订阅到正确的主题。
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())
运行工作流程#
最后，我们可以通过将消息发布到序列中的第一个 Agent 来运行工作流程。
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! 🌟🥤