Flujo de Trabajo Secuencial

El Flujo de Trabajo Secuencial es un patrón de diseño multiagente en el que los agentes responden en una secuencia determinista. Cada agente en el flujo realiza una tarea específica procesando un mensaje, generando una respuesta y luego pasándola al siguiente agente. Este patrón es útil para crear flujos de trabajo deterministas donde cada agente contribuye a una subtarea predefinida.

En este ejemplo, demostramos un flujo de trabajo secuencial donde múltiples agentes colaboran para transformar una descripción básica de un producto en una copia de marketing pulida.

El pipeline consiste en cuatro agentes especializados:

• Agente Extractor de Conceptos: Analiza la descripción inicial del producto para extraer características clave, público objetivo y propuestas únicas de venta (USPs). La salida es un análisis estructurado en un único bloque de texto.

• Agente Redactor: Redacta una copia de marketing atractiva basada en los conceptos extraídos. Este agente transforma los conocimientos analíticos en contenido promocional cautivador, entregando una narrativa coherente en un solo bloque de texto.

• Agente de Formato y Revisión: Pule la copia redactada refinando la gramática, mejorando la claridad y manteniendo un tono consistente. Este agente garantiza una calidad profesional y entrega una versión final bien formateada.

• Agente Usuario: Presenta la copia final y refinada de marketing al usuario, completando el flujo de trabajo.

El siguiente diagrama ilustra el flujo de trabajo secuencial en este ejemplo:

En este pipeline, los agentes se comunican entre sí publicando su trabajo completado como mensajes en el topic del siguiente agente en la secuencia. Por ejemplo, cuando el ConceptExtractor termina de analizar la descripción del producto, publica sus hallazgos en el topic de "WriterAgent", al cual está suscrito el WriterAgent. Este patrón continúa en cada paso del pipeline, donde cada agente publica en el topic al que está suscrito el siguiente agente en la línea.

from dataclasses import dataclass

from saptiva_agents.core import (
    MessageContext,
    RoutedAgent,
    SingleThreadedAgentRuntime,
    TopicId,
    TypeSubscription,
    message_handler,
    type_subscription,
)
from saptiva_agents import LLAMA_MODEL
from saptiva_agents.models import SystemMessage, UserMessage
from saptiva_agents.base import SaptivaAIChatCompletionClient

---

Protocolo de Mensaje

El protocolo de mensaje para este flujo de trabajo es un mensaje de texto simple que los agentes usarán para comunicar su trabajo.

@dataclass
class Message:
    content: str

---

Temas

Cada agente en el flujo de trabajo estará suscrito a un tipo de tema específico. Los tipos de temas están nombrados según los agentes en la secuencia.

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

---

Agents#

Cada clase de agente se define con un decorador type_subscription para especificar el tipo de tema al que está suscrito. Alternativamente, también puedes usar el método add_subscription() para suscribirte a un tema directamente desde el runtime.

El agente extractor de conceptos genera los puntos clave iniciales para la descripción del producto.

@type_subscription(topic_type=concept_extractor_topic_type)
class ConceptExtractorAgent(RoutedAgent):
    def __init__(self, model_client: SaptivaAIChatCompletionClient) -> 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))

El agente escritor redacta el contenido.

@type_subscription(topic_type=writer_topic_type)
class WriterAgent(RoutedAgent):
    def __init__(self, model_client: SaptivaAIChatCompletionClient) -> 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))

El agente de formato y corrección realiza el formateo y la revisión del contenido.

@type_subscription(topic_type=format_proof_topic_type)
class FormatProofAgent(RoutedAgent):
    def __init__(self, model_client: SaptivaAIChatCompletionClient) -> 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))

IEn este ejemplo, el agente usuario simplemente imprime el contenido final de marketing en consola. En una aplicación del mundo real, esto podría ser reemplazado por almacenar el resultado en una base de datos, enviar un correo electrónico, u otra acción.

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

---

Flujo de Trabajo

Ahora podemos registrar los agentes en el runtime. Como usamos el decorador type_subscription, el runtime suscribirá automáticamente los agentes a los temas correctos.

model_client = SaptivaAIChatCompletionClient(
    model=LLAMA_MODEL,
    api_key="TU_SAPTIVA_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())

---

Ejecutar el Flujo de Trabajo

Finalmente, podemos ejecutar el flujo publicando un mensaje al primer agente en la secuencia.

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