Sequential Agent 오케스트레이션

Sequential 오케스트레이션

[!IMPORTANT] Agent Framework의 오케스트레이션 기능은 실험 단계(experimental)예요. 활발히 개발 중이며, preview/RC 단계로 넘어가기 전에 크게 바뀔 수 있어요.

순차(sequential) 오케스트레이션 에서는 에이전트가 파이프라인으로 조직돼요. 각 에이전트가 차례대로 작업을 처리하고, 그 출력을 다음 에이전트에게 넘겨주죠. 문서 검토, 데이터 처리 파이프라인, 다단계 추론처럼 각 단계가 이전 단계 위에 쌓이는 워크플로에 딱 맞아요. 이 패턴을 언제 쓰고 피할지는 순차 오케스트레이션 문서를 참고하세요.

흔한 사용 사례

문서 하나가 요약 에이전트 → 번역 에이전트 → 품질 보증 에이전트를 차례로 통과하며, 각 에이전트가 이전 출력 위에서 일을 이어가는 흐름이 전형적이에요.

배울 내용

  • 각자 특화된 역할을 가진 에이전트 시퀀스를 정의하는 법
  • 각 에이전트가 이전 에이전트의 출력을 처리하도록 조율하는 법
  • 중간 출력을 관찰하고 최종 결과를 모으는 법

에이전트 정의하기

시퀀스의 각 에이전트는 하나의 책임을 가져요. 아래 예시는 개념 추출 → 카피 작성 → 교정·다듬기 순서의 세 에이전트예요. 여기서 ChatCompletionAgent 를 쓰지만 어떤 에이전트 타입이든 사용 가능해요.

from semantic_kernel.agents import Agent, ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion

def get_agents() -> list[Agent]:
    concept_extractor_agent = ChatCompletionAgent(
        name="ConceptExtractorAgent",
        instructions=(
            "You are a marketing analyst. Given a product description, identify:\n"
            "- Key features\n"
            "- Target audience\n"
            "- Unique selling points\n\n"
        ),
        service=AzureChatCompletion(),
    )
    writer_agent = ChatCompletionAgent(
        name="WriterAgent",
        instructions=(
            "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."
        ),
        service=AzureChatCompletion(),
    )
    format_proof_agent = ChatCompletionAgent(
        name="FormatProofAgent",
        instructions=(
            "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."
        ),
        service=AzureChatCompletion(),
    )
    return [concept_extractor_agent, writer_agent, format_proof_agent]

에이전트 응답 관찰하기(선택)

시퀀스가 진행되면서 각 에이전트의 출력을 살펴볼 수 있는 콜백을 정의해요.

from semantic_kernel.contents import ChatMessageContent

def agent_response_callback(message: ChatMessageContent) -> None:
    print(f"# {message.name}\n{message.content}")

순차 오케스트레이션 구성

SequentialOrchestration 객체에 에이전트 목록과 선택적 응답 콜백을 넘겨요.

from semantic_kernel.agents import SequentialOrchestration

agents = get_agents()
sequential_orchestration = SequentialOrchestration(
    members=agents,
    agent_response_callback=agent_response_callback,
)

런타임 시작

에이전트 실행을 관리할 런타임을 시작합니다.

from semantic_kernel.agents.runtime import InProcessRuntime

runtime = InProcessRuntime()
runtime.start()

오케스트레이션 호출

초기 작업(예: 제품 설명)으로 오케스트레이션을 호출하면, 출력이 시퀀스의 각 에이전트를 차례로 통과해요.

orchestration_result = await sequential_orchestration.invoke(
    task="An eco-friendly stainless steel water bottle that keeps drinks cold for 24 hours",
    runtime=runtime,
)

결과 수집

오케스트레이션이 끝날 때까지 기다렸다가 최종 출력을 받아요.

value = await orchestration_result.get(timeout=20)
print(f"***** Final Result *****\n{value}")

런타임 정지(선택)

처리가 끝나면 런타임을 정지해 리소스를 정리합니다.

await runtime.stop_when_idle()

샘플 출력

  • ConceptExtractorAgent: 제품 설명에서 핵심 기능·타겟 고객·차별점을 추출해요.
  • WriterAgent: 추출된 정보로 마케팅 카피를 작성해요.
  • FormatProofAgent: 초안을 교정하고 다듬어 최종 문구를 완성해요.

마지막 에이전트의 출력이 곧 최종 결과가 돼요. "Keep your beverages refreshingly chilled all day long with our eco-friendly stainless steel bottles…" 같은 문구로 이어지는 식이죠. 각 단계가 이전 단계의 결과를 받아 더 나은 결과로 가공한다는 흐름을 확인할 수 있어요.