첫 멀티에이전트 예제 - 순차 워크플로

첫 멀티에이전트 예제 - 순차 워크플로

순차 워크플로(Sequential Workflow)는 에이전트들이 결정적인 순서로 응답하는 멀티에이전트 설계 패턴이에요. 워크플로의 각 에이전트는 메시지를 처리해서 응답을 생성한 뒤 다음 에이전트에게 넘기는 특정 작업을 수행해요. 이 패턴은 각 에이전트가 미리 정해진 하위 작업에 기여하는, 결정적인 워크플로를 만드는 데 유용해요.

이 예시에서는 여러 에이전트가 협력해서 기본 제품 설명을 다듬어진 마케팅 카피로 바꾸는 순차 워크플로를 보여드릴게요. 파이프라인은 네 개의 전문화된 에이전트로 이뤄져 있어요:

  • Concept Extractor Agent — 초기 제품 설명을 분석해서 핵심 기능, 타겟 고객, 고유 판매 포인트(USP)를 추출해요. 출력은 단일 텍스트 블록의 구조화된 분석이에요.
  • Writer Agent — 추출된 개념을 바탕으로 설득력 있는 마케팅 카피를 만듭니다. 분석적 통찰을 매력적인 홍보 콘텐츠로 바꿔서, 단일 텍스트 블록에 응집력 있는 서사를 담아요.
  • Format & Proof Agent — 문법을 다듬고, 명확성을 높이고, 일관된 톤을 유지하며 초안 카피를 다듬어요. 전문적인 품질을 보장하고 잘 정리된 최종 버전을 내보내요.
  • User Agent — 완성된 최종 마케팅 카피를 사용자에게 보여주며 워크플로를 마무리해요.

이 워크플로는 발행·구독 메시징으로 구현할 거예요. 핵심 개념은 주제와 구독에서, API 사용법은 브로드캐스트 메시징에서 읽어 보세요.

이 파이프라인에서 에이전트들은 완성된 작업을 시퀀스의 다음 에이전트 토픽에 메시지로 발행하며 서로 통신해요. 예를 들어 ConceptExtractor가 제품 설명 분석을 끝내면 "WriterAgent" 토픽에 분석 결과를 발행하고, WriterAgent가 그 토픽을 구독하고 있어요. 이 패턴은 파이프라인의 각 단계를 거치며, 각 에이전트가 시퀀스의 다음 에이전트가 구독한 토픽으로 발행하면서 계속돼요.

출처: 공식 문서 - Sequential Workflow

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"

에이전트

각 에이전트 클래스는 구독할 토픽 타입을 지정하는 autogen_core.type_subscription 데코레이터로 정의돼요. 데코레이터 대신 autogen_core.AgentRuntime.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}")

워크플로

이제 에이전트를 런타임에 등록할 수 있어요. autogen_core.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()

더 알아보기 (Learn more)