Sequenzieller Workflow#

Sequenzieller Workflow ist ein Multi-Agenten-Designmuster, bei dem Agenten in einer deterministischen Reihenfolge reagieren. Jeder Agent im Workflow führt eine bestimmte Aufgabe aus, indem er eine Nachricht verarbeitet, eine Antwort generiert und diese dann an den nächsten Agenten weiterleitet. Dieses Muster eignet sich zur Erstellung deterministischer Workflows, bei denen jeder Agent zu einer vordefinierten Teilaufgabe beiträgt.

In diesem Beispiel demonstrieren wir einen sequenziellen Workflow, bei dem mehrere Agenten zusammenarbeiten, um eine grundlegende Produktbeschreibung in ansprechende Marketingtexte zu verwandeln.

Die Pipeline besteht aus vier spezialisierten Agenten

  • Konzeptextraktor-Agent: Analysiert die anfängliche Produktbeschreibung, um wichtige Funktionen, die Zielgruppe und einzigartige Verkaufsargumente (USPs) zu extrahieren. Das Ergebnis ist eine strukturierte Analyse in einem einzigen Textblock.

  • Schreiber-Agent: Erstellt auf der Grundlage der extrahierten Konzepte überzeugende Marketingtexte. Dieser Agent wandelt die analytischen Erkenntnisse in ansprechende Werbeinhalte um und liefert eine kohärente Erzählung in einem einzigen Textblock.

  • Format- & Korrektur-Agent: Poliert den Entwurf, indem er Grammatik verfeinert, die Klarheit verbessert und einen konsistenten Ton beibehält. Dieser Agent stellt professionelle Qualität sicher und liefert eine gut formatierte Endversion.

  • Benutzer-Agent: Präsentiert dem Benutzer den endgültigen, verfeinerten Marketingtext und schließt damit den Workflow ab.

Das folgende Diagramm veranschaulicht den sequenziellen Workflow in diesem Beispiel

Sequential Workflow

Wir werden diesen Workflow mithilfe von Publish-Subscribe-Nachrichten implementieren. Bitte lesen Sie über Thema und Abonnement für die Kernkonzepte und Broadcast-Nachrichten für die API-Nutzung.

In dieser Pipeline kommunizieren Agenten miteinander, indem sie ihre abgeschlossene Arbeit als Nachrichten an das Thema des nächsten Agenten in der Sequenz veröffentlichen. Wenn beispielsweise der KonzeptExtraktor mit der Analyse der Produktbeschreibung fertig ist, veröffentlicht er seine Ergebnisse an das Thema "SchreiberAgent", das der SchreiberAgent abonniert hat. Dieses Muster setzt sich in jedem Schritt der Pipeline fort, wobei jeder Agent an das Thema veröffentlicht, das der nächste Agent in der Reihe abonniert hat.

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

Nachrichtenprotokoll#

Das Nachrichtenprotokoll für diesen Beispiel-Workflow ist eine einfache Textnachricht, die die Agenten verwenden werden, um ihre Arbeit weiterzugeben.

@dataclass
class Message:
    content: str

Themen#

Jeder Agent im Workflow abonniert einen bestimmten Thema-Typ. Die Thema-Typen sind nach den Agenten in der Sequenz benannt. Dies ermöglicht es jedem Agenten, seine Arbeit an den nächsten Agenten in der Sequenz zu veröffentlichen.

concept_extractor_topic_type = "ConceptExtractorAgent"
writer_topic_type = "WriterAgent"
format_proof_topic_type = "FormatProofAgent"
user_topic_type = "User"

Agenten#

Jede Agentenklasse wird mit einem type_subscription-Decorator definiert, um den von ihm abonnierten Thema-Typ anzugeben. Alternativ zum Decorator können Sie auch die Methode add_subscription() verwenden, um zur Laufzeit direkt ein Thema zu abonnieren.

Der Konzeptextraktor-Agent erstellt die anfänglichen Stichpunkte für die Produktbeschreibung.

@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))

Der Schreiber-Agent führt das Schreiben durch.

@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))

Der Format-Korrektur-Agent führt die Formatierung durch.

@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))

In diesem Beispiel gibt der Benutzer-Agent einfach den endgültigen Marketingtext auf der Konsole aus. In einer realen Anwendung könnte dies durch das Speichern des Ergebnisses in einer Datenbank, das Senden einer E-Mail oder eine andere gewünschte Aktion ersetzt werden.

@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}")

Arbeitsablauf#

Nun können wir die Agenten in der Runtime registrieren. Da wir den type_subscription-Decorator verwendet haben, abonniert die Runtime die Agenten automatisch für die richtigen Themen.

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())

Führen Sie den Workflow aus#

Schließlich können wir den Workflow ausführen, indem wir eine Nachricht an den ersten Agenten in der Sequenz senden.

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! 🌟🥤