AI 에이전트 오케스트레이션 패턴 3가지 — Pipeline, Supervisor, Swarm 실전 비교

여러 에이전트를 만들고 나면 이런 고민이 생겨요.

"에이전트들을 어떻게 연결하지? 누가 누구한테 일을 시키고, 결과는 어떻게 모으지?"

이게 오케스트레이션 패턴 선택의 문제예요. 패턴을 잘못 고르면 에이전트들이 무한루프에 빠지거나, 병목이 생기거나, 디버깅이 불가능한 시스템이 돼요. 이번 글에서는 세 가지 핵심 패턴을 원리부터 실전 적용까지 정리해 드릴게요.

---

패턴 1: Pipeline (순차 파이프라인)

개념

에이전트들이 고정된 순서대로 실행돼요. A의 출력이 B의 입력이 되고, B의 출력이 C의 입력이 되는 조립 라인 구조예요.

사용자 입력
    │
    ▼
[에이전트 A: 정보 수집]
    │
    ▼
[에이전트 B: 분석]
    │
    ▼
[에이전트 C: 글쓰기]
    │
    ▼
[에이전트 D: 검수]
    │
    ▼
최종 출력

동작 방식

각 에이전트는 이전 에이전트의 결과만 받아서 처리하고 다음 에이전트로 넘겨요. 에이전트들은 서로를 모르고, 자기 앞 단계의 결과만 알아요. 분기 없이 순서만 있어요.

구현

from langgraph.graph import StateGraph, END
from typing import TypedDict

class PipelineState(TypedDict):
    query: str
    research_result: str
    analysis_result: str
    draft: str
    final_output: str

def research_agent(state: PipelineState):
    result = llm.invoke(f"다음 주제를 조사해줘: {state['query']}")
    return {"research_result": result.content}

def analysis_agent(state: PipelineState):
    result = llm.invoke(f"분석해줘: {state['research_result']}")
    return {"analysis_result": result.content}

def writing_agent(state: PipelineState):
    result = llm.invoke(f"글 써줘: {state['analysis_result']}")
    return {"draft": result.content}

def review_agent(state: PipelineState):
    result = llm.invoke(f"검수해줘: {state['draft']}")
    return {"final_output": result.content}

workflow = StateGraph(PipelineState)
workflow.add_node("research", research_agent)
workflow.add_node("analysis", analysis_agent)
workflow.add_node("writing", writing_agent)
workflow.add_node("review", review_agent)

workflow.set_entry_point("research")
workflow.add_edge("research", "analysis")
workflow.add_edge("analysis", "writing")
workflow.add_edge("writing", "review")
workflow.add_edge("review", END)

app = workflow.compile()

장점

• 구조가 단순해서 디버깅이 쉬워요. 어느 단계에서 문제가 생겼는지 바로 파악해요
• 각 에이전트가 이전 결과만 보면 되니까 컨텍스트 관리가 깔끔해요
• 흐름이 예측 가능해서 테스트하기 쉬워요

단점

• 유연성이 없어요. 순서가 고정돼 있어서 중간에 분기가 필요하면 구조를 바꿔야 해요
• 한 단계가 느리면 전체가 느려지는 병목이 생겨요
• 이전 단계 결과가 나빠도 그냥 다음 단계로 넘어가요

실제 예시

콘텐츠 생성 파이프라인

리서치 에이전트 → 아웃라인 에이전트 → 작성 에이전트 → 교정 에이전트 → 발행

문서 처리 파이프라인

추출 에이전트 → 변환 에이전트 → 검증 에이전트 → 저장 에이전트

코드 리뷰 파이프라인

정적 분석 에이전트 → 보안 검사 에이전트 → 코드 품질 에이전트 → 리포트 에이전트

언제 써야 하나

단계 순서가 명확하고 고정돼 있을 때. 각 단계가 독립적으로 테스트 가능할 때. 빠른 구현이 필요한 프로토타입 단계.

---

패턴 2: Supervisor (중앙 감독자)

개념

중앙 Supervisor가 하위 에이전트들을 동적으로 지휘해요. Supervisor는 전체 목표를 보고 "이 상황에선 A가 필요하다, 이제 B가 필요하다"를 실시간으로 판단해요.

사용자 입력
    │
    ▼
[Supervisor]
  │      │      │
  ▼      ▼      ▼
[A]    [B]    [C]
  │      │      │
  └──────┴──────┘
         │
    결과 취합
         │
       출력

하위 에이전트 A, B, C는 서로를 모르고 Supervisor만 바라봐요.

동작 방식

1. Supervisor가 사용자 요청을 수신
2. 어떤 에이전트가 필요한지 LLM으로 판단
3. 해당 에이전트 호출 + 필요한 input 전달
4. 결과 수신 후 다음 에이전트 결정 (루프)
5. "충분하다"고 판단하면 결과 취합 후 최종 응답

구현

from langgraph.graph import StateGraph, END
from typing import TypedDict
import json

class SupervisorState(TypedDict):
    messages: list
    next_agent: str
    results: dict

def search_agent(state: SupervisorState):
    query = state["messages"][-1]["content"]
    result = web_search(query)
    return {"results": {**state["results"], "search": result}}

def analysis_agent(state: SupervisorState):
    data = state["results"].get("search", "")
    result = llm.invoke(f"분석해줘: {data}")
    return {"results": {**state["results"], "analysis": result.content}}

def report_agent(state: SupervisorState):
    analysis = state["results"].get("analysis", "")
    result = llm.invoke(f"보고서 써줘: {analysis}")
    return {"results": {**state["results"], "report": result.content}}

def supervisor(state: SupervisorState):
    prompt = f"""
    사용자 요청: {state['messages'][0]['content']}
    완료된 작업: {list(state['results'].keys())}

    다음에 실행할 에이전트를 선택하세요.
    선택지: search_agent, analysis_agent, report_agent, FINISH
    JSON으로만 답하세요: {{"next": "에이전트명"}}
    """
    result = llm.invoke(prompt)
    next_agent = json.loads(result.content)["next"]
    return {"next_agent": next_agent}

def route(state: SupervisorState) -> str:
    return state["next_agent"]

workflow = StateGraph(SupervisorState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("search_agent", search_agent)
workflow.add_node("analysis_agent", analysis_agent)
workflow.add_node("report_agent", report_agent)

workflow.set_entry_point("supervisor")
workflow.add_conditional_edges(
    "supervisor",
    route,
    {
        "search_agent": "search_agent",
        "analysis_agent": "analysis_agent",
        "report_agent": "report_agent",
        "FINISH": END
    }
)

for agent in ["search_agent", "analysis_agent", "report_agent"]:
    workflow.add_edge(agent, "supervisor")

app = workflow.compile()

장점

• 유연성이 높아요. 상황에 따라 에이전트 호출 순서를 동적으로 바꿀 수 있어요
• 디버깅이 쉬워요. Supervisor의 의사결정을 추적하면 전체 흐름이 보여요
• 에이전트 추가가 쉬워요. Supervisor한테 새 에이전트를 알려주기만 하면 돼요
• 하위 에이전트가 단순해져요. 자기 역할만 하면 되거든요

단점

• Supervisor가 단일 장애점(SPOF)이에요. Supervisor가 틀린 판단을 내리면 모든 하위 에이전트가 잘못된 방향으로 가요
• Supervisor 자체가 LLM 호출이라 매 스텝마다 지연이 생겨요
• 규모가 커지면 Supervisor가 병목이 될 수 있어요

실제 예시

고객 서비스 시스템

Supervisor
  ├─ 주문 조회 에이전트 (주문 관련 질문)
  ├─ 결제 에이전트 (결제 관련 문제)
  ├─ 기술 지원 에이전트 (기술 문제)
  └─ 환불 에이전트 (환불 요청)

리서치 시스템

Supervisor
  ├─ 웹 검색 에이전트
  ├─ 논문 검색 에이전트
  ├─ 데이터 분석 에이전트
  └─ 요약 에이전트

언제 써야 하나

복잡하고 동적인 워크플로우. 에이전트 호출 순서가 상황에 따라 달라져야 할 때. 프로덕션 시스템의 약 70%가 이 패턴을 써요.

---

패턴 3: Swarm (분산 군집)

개념

중앙 관리자가 없어요. 에이전트들이 자율적으로 작업을 주고받아요. 작업을 완료하거나 자기 전문 영역을 벗어나면 다음에 적합한 에이전트한테 직접 넘겨요. 이걸 Handoff라고 해요.

사용자 입력
    │
    ▼
[에이전트 A]
    │ "이건 B가 더 잘해" (Handoff)
    ▼
[에이전트 B]
    │ "이 부분은 C가 처리해" (Handoff)
    ▼
[에이전트 C]
    │
최종 출력

에이전트들은 서로를 알고 있어요. 누가 어떤 전문성을 가졌는지 알기 때문에 직접 넘길 수 있어요.

동작 방식

1. 첫 에이전트가 요청 수신
2. 자기가 처리 가능하면 직접 처리
3. 처리 범위를 벗어나면 적합한 에이전트로 Handoff
4. 받은 에이전트가 동일하게 처리 또는 재Handoff
5. 최종 처리 에이전트가 결과 반환

구현

from openai import OpenAI

client = OpenAI()

# Handoff 함수 정의
def transfer_to_tech_agent():
    """기술 문제 전문가에게 전달"""
    return tech_agent

def transfer_to_billing_agent():
    """결제 문제 전문가에게 전달"""
    return billing_agent

def transfer_to_triage_agent():
    """처음으로 돌아가기"""
    return triage_agent

# 에이전트 정의
triage_agent = {
    "name": "triage_agent",
    "instructions": """
    사용자 요청을 분석해서 적절한 전문가에게 넘겨줘.
    - 기술 문제 → transfer_to_tech_agent 호출
    - 결제 문제 → transfer_to_billing_agent 호출
    - 일반 문의 → 직접 답변
    """,
    "functions": [transfer_to_tech_agent, transfer_to_billing_agent]
}

tech_agent = {
    "name": "tech_agent",
    "instructions": """
    기술 문제 전문가야. 오직 기술 관련 문제만 처리해.
    결제 관련 문제가 나오면 → transfer_to_billing_agent 호출
    해결이 안 되면 → transfer_to_triage_agent 호출
    """,
    "functions": [transfer_to_billing_agent, transfer_to_triage_agent]
}

billing_agent = {
    "name": "billing_agent",
    "instructions": """
    결제 문제 전문가야. 오직 결제 관련 문제만 처리해.
    기술 관련 문제가 나오면 → transfer_to_tech_agent 호출
    해결이 안 되면 → transfer_to_triage_agent 호출
    """,
    "functions": [transfer_to_tech_agent, transfer_to_triage_agent]
}

# Swarm 실행 루프
def run_swarm(user_message: str):
    current_agent = triage_agent
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=[
                {"type": "function", "function": {
                    "name": fn.__name__,
                    "description": fn.__doc__
                }}
                for fn in current_agent["functions"]
            ]
        )

        message = response.choices[0].message

        # Handoff 발생
        if message.tool_calls:
            tool_name = message.tool_calls[0].function.name
            fn_map = {fn.__name__: fn for fn in current_agent["functions"]}
            current_agent = fn_map[tool_name]()
            messages.append({"role": "assistant", "content": f"Handoff to {current_agent['name']}"})

        # 최종 답변
        elif message.content:
            return message.content

        messages.append(message)

장점

• 단일 장애점이 없어요. 어떤 에이전트가 죽어도 다른 에이전트들이 계속 동작해요
• 확장이 쉬워요. 에이전트 하나 추가하면 다른 에이전트들이 그쪽으로 Handoff할 수 있어요
• 각 에이전트가 자기 전문 영역만 담당해서 단순하고 명확해요
• 자연스러운 부하 분산이 돼요

단점

• 디버깅이 어려워요. A → B → C → A 같은 순환 루프가 생기면 잡기 어려워요
• 전체 흐름을 추적하기 힘들어요. 누가 어디에 있는지 파악하기 어렵거든요
• 종료 조건을 명확하게 설계하지 않으면 무한루프에 빠져요
• 에이전트 간 상태 공유가 복잡해요

실제 예시

고객 서비스 라우팅

Triage 에이전트
  ↔ 주문 에이전트 (주문 조회, 배송 추적)
  ↔ 결제 에이전트 (결제, 환불)
  ↔ 기술 에이전트 (앱 오류, 계정 문제)
  ↔ VIP 에이전트 (프리미엄 고객)

리서치 팀 시뮬레이션

리서처 에이전트
  ↔ 팩트체커 에이전트
  ↔ 분석가 에이전트
  ↔ 작성자 에이전트

언제 써야 하나

역할이 명확하게 분리된 전문가 팀 구조. 워크플로우가 미리 정해지지 않고 상황에 따라 달라지는 경우. 고객 서비스처럼 요청 유형이 다양하고 전문화된 처리가 필요한 경우.

---

세 패턴 전체 비교

구분 Pipeline Supervisor Swarm

제어 방식	순서 고정	중앙 집중	분산 자율
유연성	낮음	중간	높음
디버깅 난이도	쉬움	중간	어려움
단일 장애점	없음	Supervisor	없음
구현 복잡도	낮음	중간	높음
병렬 처리	불가	가능	가능
에이전트 간 통신	없음	Supervisor 경유	직접
적합한 규모	소규모	중간	대규모
프레임워크	LangGraph	LangGraph, CrewAI	OpenAI Agents SDK, AutoGen

---

언제 어떤 패턴을 써야 하나

Pipeline을 선택해야 할 때

• 단계 순서가 고정되어 있고 바뀌지 않을 때
• 빠르게 프로토타입을 만들어야 할 때
• 각 단계를 독립적으로 테스트하고 싶을 때

Supervisor를 선택해야 할 때

• 동적이고 복잡한 워크플로우가 필요할 때
• 에이전트 호출 순서가 상황마다 달라져야 할 때
• 중앙 제어와 감시가 필요할 때 (금융, 의료 등 규제 산업)

Swarm을 선택해야 할 때

• 역할이 명확하게 분리된 전문가 팀 구조일 때
• 고가용성이 중요해서 단일 장애점을 없애야 할 때
• 대규모로 에이전트를 확장해야 할 때

---

실전에서는 패턴을 조합한다

프로덕션 시스템에서는 세 패턴을 섞어서 써요.

[Supervisor]                    ← 전체 조율
    │
    ├─ [Pipeline]               ← 문서 처리 단계는 순차
    │    추출 → 변환 → 검증
    │
    └─ [Swarm]                  ← 고객 응대는 분산
         Triage ↔ Tech ↔ Billing

Microsoft 권장사항처럼 항상 제일 단순한 패턴부터 시작하세요. Pipeline으로 충분하면 Pipeline, 동적 분기가 필요하면 Supervisor, 그 다음에야 Swarm을 고려하는 순서로 가는 게 맞아요. 😄

---