오케스트레이터-워커 패턴 완벽 가이드

AI 에이전트 시스템에서 복잡한 작업을 효율적으로 분배하고 처리하는 오케스트레이터-워커 패턴의 개념과 활용법을 초급 개발자도 이해할 수 있도록 쉽게 설명합니다.

1. 오케스트레이터-워커 패턴이란

어느 날 김개발 씨가 회사에서 AI 기반 문서 분석 시스템을 개발하게 되었습니다. 수백 개의 문서를 동시에 분석하고, 요약하고, 번역까지 해야 하는 복잡한 요구사항이었습니다.

"이걸 어떻게 효율적으로 처리하지?" 고민하던 김개발 씨에게 선배가 한마디 던졌습니다. "오케스트레이터-워커 패턴을 써봐."

오케스트레이터-워커 패턴은 한마디로 지휘자와 연주자의 관계와 같습니다. 오케스트라에서 지휘자가 악보를 해석하고 각 연주자에게 역할을 지시하듯이, 오케스트레이터는 복잡한 작업을 분석하여 여러 워커에게 적절히 분배합니다.

이 패턴을 사용하면 복잡한 AI 워크플로우를 체계적으로 관리하고, 각 작업을 병렬로 처리할 수 있습니다.

다음 코드를 살펴봅시다.

# 오케스트레이터-워커 패턴의 기본 구조
class Orchestrator:
    def __init__(self):
        self.workers = {}  # 워커들을 저장할 딕셔너리

    def register_worker(self, task_type, worker):
        # 작업 유형별로 워커를 등록합니다
        self.workers[task_type] = worker

    def process(self, task):
        # 작업을 분석하여 적절한 워커에게 분배합니다
        task_type = self.analyze_task(task)
        worker = self.workers.get(task_type)
        return worker.execute(task)

    def analyze_task(self, task):
        # 작업의 유형을 판단하는 핵심 로직
        return task.get("type", "default")

김개발 씨는 입사 1년 차 주니어 개발자입니다. 최근 회사에서 AI 기반 고객 서비스 자동화 프로젝트를 맡게 되었는데, 문제가 복잡했습니다.

고객 문의가 들어오면 내용을 분석하고, 적절한 답변을 생성하고, 필요하면 번역까지 해야 했습니다. 처음에 김개발 씨는 모든 로직을 하나의 거대한 함수에 넣었습니다.

분석도 하고, 답변도 생성하고, 번역도 하는 만능 함수였습니다. 하지만 코드는 점점 복잡해졌고, 새로운 기능을 추가할 때마다 기존 코드가 망가지는 일이 반복되었습니다.

그때 선배 개발자 박시니어 씨가 다가와 물었습니다. "혹시 오케스트레이터-워커 패턴 들어봤어요?" 박시니어 씨는 화이트보드에 그림을 그리기 시작했습니다.

"오케스트라 공연을 생각해봐요. 지휘자가 모든 악기를 직접 연주하진 않잖아요.

대신 악보를 해석해서 바이올린 주자에겐 바이올린 파트를, 첼로 주자에겐 첼로 파트를 지시하죠." 김개발 씨의 눈이 반짝였습니다. "아, 그러니까 제 코드에서도 작업을 지시하는 역할과 실제로 수행하는 역할을 분리하라는 거군요!" 맞습니다.

오케스트레이터는 들어온 요청을 분석하고, 어떤 작업이 필요한지 판단합니다. 그리고 각 작업에 적합한 워커에게 일을 맡깁니다.

워커는 자신이 맡은 일만 집중해서 처리합니다. 이 패턴의 핵심은 관심사의 분리입니다.

오케스트레이터는 전체 흐름을 관리하는 데 집중하고, 워커는 특정 작업을 잘 수행하는 데 집중합니다. 마치 회사에서 팀장이 업무를 분배하고 팀원들이 각자 맡은 일을 처리하는 것과 같습니다.

위의 코드를 살펴보면, Orchestrator 클래스가 중앙에서 모든 것을 조율합니다. register_worker 메서드로 작업 유형별로 워커를 등록해두고, process 메서드에서 들어온 작업을 분석하여 적절한 워커에게 전달합니다.

analyze_task 메서드가 특히 중요합니다. 이 부분에서 LLM을 활용하면 자연어로 된 요청을 분석하여 어떤 워커가 처리해야 할지 동적으로 판단할 수 있습니다.

단순한 규칙 기반이 아니라 맥락을 이해하는 지능적인 분배가 가능해지는 것입니다. 실무에서 이 패턴은 다양하게 활용됩니다.

챗봇 서비스에서 사용자 질문을 분석하여 FAQ 워커, 예약 워커, 불만 처리 워커 등에 분배할 수 있습니다. 데이터 파이프라인에서는 데이터 유형에 따라 전처리 워커, 분석 워커, 저장 워커를 선택적으로 호출합니다.

주의할 점도 있습니다. 오케스트레이터에 너무 많은 로직을 넣으면 안 됩니다.

오케스트레이터는 교통 경찰처럼 방향만 지시하고, 실제 운전은 각 차량이 해야 합니다. 또한 워커 간의 의존성이 생기지 않도록 설계해야 합니다.

다시 김개발 씨의 이야기로 돌아가면, 그는 이 패턴을 적용한 후 코드가 훨씬 깔끔해졌습니다. 새로운 기능을 추가할 때도 새 워커만 만들어서 등록하면 되었습니다.

"이제야 제대로 된 구조가 잡힌 것 같아요!"

실전 팁

💡 - 오케스트레이터는 가볍게 유지하고, 복잡한 비즈니스 로직은 워커에 넣으세요

워커는 독립적으로 테스트할 수 있도록 설계하면 유지보수가 쉬워집니다

2. 오케스트레이터의 역할

김개발 씨가 오케스트레이터-워커 패턴의 기본 개념을 이해한 후, 다음 질문이 떠올랐습니다. "그런데 오케스트레이터는 정확히 어떤 일을 하는 거죠?

그냥 작업을 분배만 하면 되나요?" 박시니어 씨가 웃으며 대답했습니다. "그게 다가 아니에요.

오케스트레이터의 역할은 생각보다 다양해요."

오케스트레이터는 시스템의 두뇌 역할을 합니다. 단순히 작업을 분배하는 것을 넘어, 작업을 분석하고, 실행 계획을 수립하며, 워커들의 결과를 수집하여 최종 응답을 조합합니다.

마치 영화 감독이 시나리오를 해석하고 배우들에게 연기 지시를 내린 후 편집까지 총괄하는 것과 같습니다.

다음 코드를 살펴봅시다.

import asyncio
from openai import OpenAI

class SmartOrchestrator:
    def __init__(self, llm_client):
        self.llm = llm_client
        self.workers = {}
        self.execution_history = []

    async def plan_and_execute(self, user_request):
        # 1단계: LLM으로 작업 계획 수립
        plan = await self.create_execution_plan(user_request)

        # 2단계: 계획에 따라 워커들 실행
        results = []
        for step in plan["steps"]:
            worker = self.workers[step["worker_type"]]
            result = await worker.execute(step["task"])
            results.append(result)

        # 3단계: 결과 종합하여 최종 응답 생성
        final_response = await self.synthesize_results(results)
        return final_response

김개발 씨는 오케스트레이터가 단순한 라우터가 아니라는 것을 깨달았습니다. 박시니어 씨는 칠판에 오케스트레이터의 네 가지 핵심 역할을 적었습니다.

첫 번째는 작업 분석입니다. 사용자의 요청이 들어오면 오케스트레이터는 그것이 무엇을 원하는 것인지 파악해야 합니다.

"이 PDF 문서를 요약해줘"라는 요청이 들어오면, 단순 요약인지 핵심 포인트 추출인지 판단해야 합니다. 여기서 LLM의 자연어 이해 능력이 빛을 발합니다.

두 번째는 실행 계획 수립입니다. 복잡한 요청은 여러 단계로 나눠야 합니다.

예를 들어 "이 영문 논문을 번역하고 요약해줘"라는 요청은 먼저 번역 워커가 작업하고, 그 결과를 요약 워커에게 전달해야 합니다. 오케스트레이터는 이런 순서와 의존 관계를 계획합니다.

세 번째는 워커 조율입니다. 계획이 수립되면 적절한 워커들을 호출합니다.

병렬로 실행할 수 있는 작업은 동시에 처리하고, 순차적으로 처리해야 하는 작업은 순서를 지킵니다. 위 코드의 plan_and_execute 메서드가 바로 이 역할을 합니다.

네 번째는 결과 종합입니다. 각 워커가 반환한 결과를 모아서 하나의 의미 있는 응답으로 만들어야 합니다.

단순히 결과를 이어붙이는 것이 아니라, 맥락에 맞게 정리하고 사용자가 이해하기 쉽게 가공합니다. 박시니어 씨가 덧붙였습니다.

"오케스트레이터는 또한 오류 처리도 담당해요. 워커가 실패하면 다른 워커로 대체하거나, 사용자에게 적절한 안내를 해야 하죠." 실제로 위 코드를 보면 create_execution_plan 메서드에서 LLM을 활용하여 실행 계획을 동적으로 생성합니다.

이것이 규칙 기반 시스템과의 큰 차이점입니다. LLM 덕분에 미리 정의하지 않은 새로운 유형의 요청도 유연하게 처리할 수 있습니다.

김개발 씨가 질문했습니다. "그러면 오케스트레이터가 너무 무거워지지 않나요?" 좋은 질문입니다.

오케스트레이터는 의사결정과 조율에만 집중해야 합니다. 실제 데이터 처리나 복잡한 로직은 워커에게 맡겨야 합니다.

마치 회사의 CEO가 모든 실무를 직접 처리하지 않는 것과 같습니다. CEO는 전략을 수립하고 팀을 조율하며 결과를 검토합니다.

실제 업무는 각 팀이 수행합니다. 오케스트레이터도 마찬가지입니다.

execution_history를 저장하는 것도 주목할 만합니다. 실행 이력을 기록해두면 디버깅이 쉬워지고, 어떤 워커가 얼마나 자주 호출되는지 분석할 수도 있습니다.

실전 팁

💡 - LLM을 활용한 동적 계획 수립으로 유연성을 확보하세요

오케스트레이터에 실행 이력을 남겨두면 문제 추적이 쉬워집니다

3. 워커의 역할과 종류

오케스트레이터의 역할을 배운 김개발 씨가 이번엔 워커에 관심을 가졌습니다. "그럼 워커는 어떻게 설계해야 하나요?

그냥 함수 하나면 되나요?" 박시니어 씨가 고개를 저었습니다. "워커도 제대로 설계해야 해요.

종류도 다양하고, 각자 특성이 달라요."

워커는 실제로 작업을 수행하는 전문가입니다. 각 워커는 한 가지 작업에 특화되어 있으며, 독립적으로 동작합니다.

마치 병원에서 내과 전문의, 외과 전문의, 소아과 전문의가 각자 전문 분야에서 환자를 치료하는 것과 같습니다. 워커는 크게 LLM 기반 워커, 도구 기반 워커, 하이브리드 워커로 나눌 수 있습니다.

다음 코드를 살펴봅시다.

from abc import ABC, abstractmethod

# 워커의 기본 인터페이스
class BaseWorker(ABC):
    @abstractmethod
    async def execute(self, task: dict) -> dict:
        pass

# LLM 기반 워커: 텍스트 생성, 요약 등
class SummarizerWorker(BaseWorker):
    def __init__(self, llm_client):
        self.llm = llm_client

    async def execute(self, task: dict) -> dict:
        prompt = f"다음 내용을 요약해주세요: {task['content']}"
        response = await self.llm.generate(prompt)
        return {"summary": response, "status": "success"}

# 도구 기반 워커: API 호출, 데이터베이스 조회 등
class DatabaseWorker(BaseWorker):
    def __init__(self, db_connection):
        self.db = db_connection

    async def execute(self, task: dict) -> dict:
        query = task.get("query")
        result = await self.db.execute(query)
        return {"data": result, "status": "success"}

박시니어 씨는 워커의 세 가지 유형을 설명하기 시작했습니다. 첫 번째는 LLM 기반 워커입니다.

이 워커는 언어 모델을 활용하여 텍스트 생성, 요약, 번역, 감정 분석 등을 수행합니다. 위 코드의 SummarizerWorker가 좋은 예시입니다.

들어온 텍스트를 LLM에 전달하고 결과를 반환합니다. 두 번째는 도구 기반 워커입니다.

이 워커는 외부 시스템과 상호작용합니다. 데이터베이스 조회, API 호출, 파일 시스템 접근 등이 여기에 해당합니다.

DatabaseWorker처럼 LLM 없이도 동작하며, 특정 시스템과의 연동에 집중합니다. 세 번째는 하이브리드 워커입니다.

LLM과 도구를 함께 활용합니다. 예를 들어 검색 워커는 먼저 도구로 검색 API를 호출하고, 그 결과를 LLM으로 정리하여 반환할 수 있습니다.

김개발 씨가 물었습니다. "각 워커를 어떻게 잘 설계할 수 있을까요?" 박시니어 씨는 워커 설계의 핵심 원칙을 알려주었습니다.

첫째, 단일 책임 원칙입니다. 각 워커는 한 가지 일만 해야 합니다.

요약 워커는 요약만, 번역 워커는 번역만 합니다. 여러 기능을 한 워커에 넣으면 재사용성이 떨어집니다.

둘째, 독립성입니다. 워커끼리 직접 통신하면 안 됩니다.

모든 조율은 오케스트레이터가 담당합니다. 워커 A가 워커 B를 직접 호출하면 의존성이 생기고 테스트가 어려워집니다.

셋째, 일관된 인터페이스입니다. 위 코드에서 BaseWorker 추상 클래스를 정의한 것처럼, 모든 워커는 동일한 인터페이스를 따라야 합니다.

그래야 오케스트레이터가 워커를 교체하거나 추가할 때 수정할 코드가 줄어듭니다. 넷째, 오류 처리입니다.

각 워커는 자신의 오류를 적절히 처리해야 합니다. 외부 API 호출이 실패하면 재시도할지, 대체 값을 반환할지 결정해야 합니다.

실무에서 자주 만나는 워커 종류를 소개합니다. 검색 워커는 벡터 데이터베이스나 검색 엔진에서 관련 정보를 찾아옵니다.

코드 실행 워커는 사용자가 요청한 코드를 안전한 샌드박스에서 실행합니다. 이미지 분석 워커는 멀티모달 LLM을 활용하여 이미지를 분석합니다.

김개발 씨가 정리했습니다. "결국 워커는 레고 블록 같은 거네요.

각 블록은 독립적이고, 조합하면 다양한 것을 만들 수 있는 거죠." 정확합니다. 잘 설계된 워커는 여러 프로젝트에서 재사용할 수 있습니다.

실전 팁

💡 - 워커의 execute 메서드는 항상 dict를 반환하도록 표준화하세요

워커 내부에서 다른 워커를 직접 호출하지 마세요

4. 패턴의 장점

김개발 씨가 오케스트레이터-워커 패턴을 프로젝트에 적용하기 시작했습니다. 몇 주 후, 팀 회고 시간에 김개발 씨가 발표했습니다.

"이 패턴을 적용하고 나서 정말 많은 것이 좋아졌어요." 팀원들이 귀를 기울였습니다. "어떤 점이 좋아졌나요?"

오케스트레이터-워커 패턴은 확장성, 유연성, 유지보수성이라는 세 가지 핵심 장점을 제공합니다. 마치 잘 조직된 회사처럼, 새로운 팀원을 쉽게 추가할 수 있고, 역할 변경이 간편하며, 문제가 생겼을 때 원인을 빠르게 찾을 수 있습니다.

다음 코드를 살펴봅시다.

class ScalableOrchestrator:
    def __init__(self):
        self.workers = {}
        self.worker_pool = {}  # 워커 풀로 확장성 확보

    def add_worker(self, task_type: str, worker: BaseWorker):
        # 새 워커 추가가 간단합니다 - 확장성
        self.workers[task_type] = worker

    def replace_worker(self, task_type: str, new_worker: BaseWorker):
        # 기존 워커 교체도 간단합니다 - 유연성
        old_worker = self.workers.get(task_type)
        self.workers[task_type] = new_worker
        return old_worker

    async def execute_with_fallback(self, task_type: str, task: dict):
        # 실패 시 대체 워커 사용 - 견고성
        try:
            primary_worker = self.workers[task_type]
            return await primary_worker.execute(task)
        except Exception as e:
            fallback_worker = self.workers.get(f"{task_type}_fallback")
            if fallback_worker:
                return await fallback_worker.execute(task)
            raise e

김개발 씨는 발표 자료를 펼쳤습니다. 첫 번째 장점은 확장성입니다.

"예전에는 새 기능을 추가하려면 기존 코드를 많이 수정해야 했어요. 하지만 이제는 새 워커만 만들어서 등록하면 됩니다." 위 코드의 add_worker 메서드를 보면, 새 워커를 추가하는 것이 얼마나 간단한지 알 수 있습니다.

팀원 이주니어 씨가 물었습니다. "정말요?

번역 기능을 추가할 때도 그랬나요?" 김개발 씨가 고개를 끄덕였습니다. "네, TranslatorWorker 클래스를 만들고 등록만 했어요.

기존 코드는 한 줄도 수정하지 않았습니다." 두 번째 장점은 유연성입니다. "워커를 교체하는 것도 쉬워요.

요약 워커가 GPT-3.5 기반이었는데, GPT-4로 바꾸고 싶었어요. replace_worker 메서드로 간단히 교체했습니다." 이것은 개방-폐쇄 원칙의 좋은 예시입니다.

확장에는 열려 있고 수정에는 닫혀 있는 구조입니다. 세 번째 장점은 유지보수성입니다.

"버그가 발생했을 때 어디를 봐야 하는지 명확해요. 요약이 이상하면 요약 워커를 보면 되고, 번역이 이상하면 번역 워커를 보면 됩니다." 각 워커가 독립적이기 때문에 문제 격리가 쉽습니다.

네 번째 장점은 테스트 용이성입니다. "각 워커를 독립적으로 테스트할 수 있어요.

오케스트레이터 테스트에서는 목 워커를 사용하면 됩니다." 단위 테스트와 통합 테스트를 깔끔하게 분리할 수 있습니다. 다섯 번째 장점은 견고성입니다.

위 코드의 execute_with_fallback 메서드처럼, 주 워커가 실패하면 대체 워커를 사용할 수 있습니다. 시스템의 안정성이 높아집니다.

여섯 번째 장점은 병렬 처리입니다. 서로 독립적인 워커들은 동시에 실행할 수 있습니다.

요약과 번역이 동시에 필요하다면 병렬로 처리하여 응답 시간을 단축할 수 있습니다. 박시니어 씨가 추가했습니다.

"팀 협업에도 좋아요. 각 개발자가 자기 담당 워커만 집중해서 개발할 수 있으니까요." 하지만 주의할 점도 있습니다.

작은 프로젝트에 이 패턴을 적용하면 오버엔지니어링이 될 수 있습니다. 워커가 3개 미만이라면 굳이 이 패턴을 적용하지 않아도 됩니다.

실전 팁

💡 - 워커 추가 시 기존 오케스트레이터 코드를 수정할 필요가 없다면 잘 설계된 것입니다

대체 워커(fallback)를 미리 등록해두면 시스템 안정성이 높아집니다

5. 동적 작업 분배 전략

김개발 씨의 시스템이 점점 복잡해졌습니다. 워커 종류가 10개가 넘어가자 새로운 고민이 생겼습니다.

"사용자 요청이 들어올 때마다 어떤 워커를 써야 할지 어떻게 판단하지?" 하드코딩된 규칙으로는 한계가 있었습니다. 박시니어 씨가 해결책을 제시했습니다.

"LLM을 활용한 동적 분배를 써봐요."

동적 작업 분배는 사용자 요청의 의도를 실시간으로 분석하여 적절한 워커를 선택하는 전략입니다. 규칙 기반 분배가 정해진 키워드에만 반응한다면, 동적 분배는 문맥을 이해하고 유연하게 판단합니다.

마치 숙련된 상담원이 고객의 말을 듣고 적절한 부서로 연결해주는 것과 같습니다.

다음 코드를 살펴봅시다.

class DynamicOrchestrator:
    def __init__(self, llm_client):
        self.llm = llm_client
        self.workers = {}
        self.worker_descriptions = {}  # 워커 설명 저장

    def register_worker(self, name: str, worker: BaseWorker, description: str):
        self.workers[name] = worker
        self.worker_descriptions[name] = description

    async def route_request(self, user_request: str) -> str:
        # LLM에게 워커 목록과 설명을 제공하고 선택하게 합니다
        prompt = f"""사용자 요청: {user_request}

사용 가능한 워커:
{self._format_worker_list()}

가장 적합한 워커 이름을 선택하세요."""

        selected_worker = await self.llm.generate(prompt)
        return selected_worker.strip()

    def _format_worker_list(self) -> str:
        return "\n".join([
            f"- {name}: {desc}"
            for name, desc in self.worker_descriptions.items()
        ])

박시니어 씨가 화이트보드에 두 가지 분배 방식을 그렸습니다. 첫 번째는 규칙 기반 분배입니다.

"요청에 '요약'이라는 단어가 있으면 요약 워커를 호출해"와 같은 방식입니다. 구현이 간단하지만 한계가 명확합니다.

"이 글의 핵심만 알려줘"라고 하면 요약 워커를 호출해야 하는데, '요약'이라는 단어가 없어서 실패합니다. 두 번째는 LLM 기반 동적 분배입니다.

LLM에게 사용자 요청과 워커 목록을 주고 판단하게 합니다. LLM은 "핵심만 알려줘"가 요약 요청이라는 것을 이해합니다.

위 코드의 route_request 메서드가 이 방식을 구현합니다. 김개발 씨가 질문했습니다.

"매번 LLM을 호출하면 비용이 너무 많이 들지 않나요?" 좋은 지적입니다. 몇 가지 최적화 전략이 있습니다.

캐싱을 활용하면 비슷한 요청에 대한 라우팅 결과를 재사용할 수 있습니다. 작은 모델 사용도 효과적입니다.

라우팅은 복잡한 추론이 필요 없으므로 GPT-3.5 같은 가벼운 모델로도 충분합니다. 또한 하이브리드 전략도 있습니다.

먼저 규칙 기반으로 시도하고, 규칙에 맞지 않으면 LLM을 호출합니다. 자주 들어오는 패턴은 규칙으로 빠르게 처리하고, 새로운 패턴만 LLM이 처리합니다.

위 코드에서 worker_descriptions가 중요합니다. 각 워커의 역할을 명확하게 설명해야 LLM이 정확한 판단을 내릴 수 있습니다.

"요약 워커"보다 "긴 텍스트를 핵심 포인트 3-5개로 요약하는 워커"라고 설명하는 것이 좋습니다. 멀티 워커 선택도 고려해야 합니다.

하나의 요청이 여러 워커를 필요로 할 수 있습니다. "이 문서를 번역하고 요약해줘"는 번역 워커와 요약 워커 둘 다 필요합니다.

이 경우 LLM이 실행 순서까지 결정하도록 프롬프트를 수정합니다. 오케스트레이터는 선택된 워커가 실제로 존재하는지 검증해야 합니다.

LLM이 등록되지 않은 워커 이름을 반환할 수도 있기 때문입니다. 항상 워커 존재 여부를 확인하고, 없으면 기본 워커를 사용하거나 사용자에게 알려야 합니다.

김개발 씨가 코드를 수정한 후 테스트했습니다. "와, 이제 '간략하게 정리해줘'라고 해도 요약 워커가 호출되네요!" 동적 분배의 힘을 체감한 순간이었습니다.

실전 팁

💡 - 워커 설명은 구체적으로 작성하세요. LLM의 판단 정확도가 올라갑니다

자주 사용되는 패턴은 캐싱하여 비용을 절감하세요

6. 주요 활용 사례 분석

이론을 충분히 배운 김개발 씨가 물었습니다. "실제 현업에서는 이 패턴을 어떻게 쓰나요?

구체적인 사례가 궁금해요." 박시니어 씨가 노트북을 열었습니다. "마침 내가 참여했던 프로젝트 몇 개를 보여줄게요.

다양한 산업에서 이 패턴을 활용하고 있어요."

오케스트레이터-워커 패턴은 고객 서비스 자동화, 문서 처리 파이프라인, 데이터 분석 시스템 등 다양한 분야에서 활용됩니다. 특히 복잡한 작업을 여러 단계로 나눠 처리해야 하는 시스템에서 빛을 발합니다.

실제 사례를 통해 패턴의 실무 적용 방법을 살펴보겠습니다.

다음 코드를 살펴봅시다.

# 고객 서비스 자동화 시스템 예시
class CustomerServiceOrchestrator:
    def __init__(self):
        self.workers = {
            "intent_classifier": IntentClassifierWorker(),
            "faq_responder": FAQResponderWorker(),
            "order_tracker": OrderTrackerWorker(),
            "complaint_handler": ComplaintHandlerWorker(),
            "human_handoff": HumanHandoffWorker(),
        }

    async def handle_inquiry(self, customer_message: str) -> dict:
        # 1단계: 의도 분류
        intent = await self.workers["intent_classifier"].execute({
            "message": customer_message
        })

        # 2단계: 의도에 맞는 워커 실행
        handler = self.workers.get(intent["handler"])
        if handler:
            response = await handler.execute({"message": customer_message})
        else:
            response = await self.workers["human_handoff"].execute({
                "message": customer_message,
                "reason": "unknown_intent"
            })

        return response

박시니어 씨가 첫 번째 사례를 소개했습니다. 첫 번째는 고객 서비스 자동화 시스템입니다.

위 코드가 바로 이 사례입니다. 고객 문의가 들어오면 먼저 IntentClassifierWorker가 의도를 파악합니다.

FAQ 질문인지, 주문 추적인지, 불만 사항인지 분류합니다. 분류 결과에 따라 적절한 워커가 호출됩니다.

FAQ 질문이면 FAQResponderWorker가 답변하고, 주문 추적이면 OrderTrackerWorker가 주문 상태를 조회합니다. 처리할 수 없는 복잡한 문의는 HumanHandoffWorker가 상담원에게 연결합니다.

김개발 씨가 물었습니다. "이렇게 하면 상담원이 필요 없어지나요?" 박시니어 씨가 고개를 저었습니다.

"아니요, 단순 문의를 자동화해서 상담원이 복잡한 문제에 집중할 수 있게 하는 거예요." 두 번째는 문서 처리 파이프라인입니다. 법률 사무소에서 계약서를 분석하는 시스템을 예로 들어보겠습니다.

PDF 파싱 워커가 문서를 텍스트로 변환합니다. 조항 추출 워커가 주요 조항을 식별합니다.

위험 분석 워커가 불리한 조항을 찾아냅니다. 요약 워커가 핵심 내용을 정리합니다.

각 워커가 순차적으로 실행되며, 이전 워커의 결과가 다음 워커의 입력이 됩니다. 오케스트레이터는 이 파이프라인을 관리하고, 중간에 오류가 발생하면 적절히 처리합니다.

세 번째는 RAG 기반 질의응답 시스템입니다. 검색 워커가 벡터 데이터베이스에서 관련 문서를 찾습니다.

리랭킹 워커가 검색 결과의 관련성을 재평가합니다. 답변 생성 워커가 검색된 맥락을 바탕으로 답변을 생성합니다.

인용 추가 워커가 출처를 명시합니다. 네 번째는 코드 리뷰 자동화 시스템입니다.

코드 파싱 워커가 PR의 변경 사항을 분석합니다. 스타일 검사 워커가 코딩 컨벤션 위반을 찾습니다.

보안 검사 워커가 취약점을 탐지합니다. 리뷰 생성 워커가 종합적인 리뷰 코멘트를 작성합니다.

김개발 씨가 감탄했습니다. "정말 다양한 곳에서 쓰이네요!" 핵심은 복잡한 작업을 의미 있는 단위로 분해하는 것입니다.

각 단위가 워커가 되고, 오케스트레이터가 이들을 조율합니다. 실무에서는 워커 간 데이터 전달 형식을 표준화하는 것이 중요합니다.

모든 워커가 동일한 구조의 딕셔너리를 주고받으면 조합이 쉬워집니다.

실전 팁

💡 - 의도 분류 워커는 다른 모든 워커의 관문이므로 정확도가 중요합니다

처리할 수 없는 경우를 위한 대체 경로(human handoff)를 항상 준비하세요

7. 복잡한 워크플로 설계하기

김개발 씨가 이제 자신감이 생겼습니다. 하지만 새로운 도전이 기다리고 있었습니다.

"이번 프로젝트는 워커가 순차적으로만 실행되면 안 돼요. 어떤 건 병렬로, 어떤 건 조건부로 실행해야 해요." 박시니어 씨가 고개를 끄덕였습니다.

"드디어 복잡한 워크플로를 배울 때가 됐네요."

복잡한 워크플로는 순차 실행, 병렬 실행, 조건부 분기, 반복을 조합하여 설계합니다. 마치 복잡한 요리 레시피처럼, 어떤 단계는 순서대로, 어떤 단계는 동시에, 어떤 단계는 상황에 따라 건너뛰어야 합니다.

이를 체계적으로 관리하면 복잡한 비즈니스 로직도 깔끔하게 구현할 수 있습니다.

다음 코드를 살펴봅시다.

import asyncio
from enum import Enum

class StepType(Enum):
    SEQUENTIAL = "sequential"
    PARALLEL = "parallel"
    CONDITIONAL = "conditional"

class WorkflowOrchestrator:
    def __init__(self):
        self.workers = {}

    async def execute_workflow(self, workflow: list, context: dict) -> dict:
        for step in workflow:
            if step["type"] == StepType.PARALLEL:
                # 병렬 실행: 여러 워커를 동시에 실행
                tasks = [
                    self.workers[w].execute(context)
                    for w in step["workers"]
                ]
                results = await asyncio.gather(*tasks)
                context["parallel_results"] = results

            elif step["type"] == StepType.CONDITIONAL:
                # 조건부 분기: 조건에 따라 다른 워커 실행
                condition_result = step["condition"](context)
                worker_name = step["if_true"] if condition_result else step["if_false"]
                result = await self.workers[worker_name].execute(context)
                context[step["output_key"]] = result

            else:  # SEQUENTIAL
                result = await self.workers[step["worker"]].execute(context)
                context[step["output_key"]] = result

        return context

박시니어 씨가 복잡한 워크플로의 세 가지 패턴을 설명했습니다. 첫 번째는 순차 실행입니다.

워커 A의 결과가 워커 B의 입력이 되는 경우입니다. 문서 번역 후 요약하는 작업이 좋은 예입니다.

번역이 완료되어야 요약할 수 있으므로 순서가 중요합니다. 두 번째는 병렬 실행입니다.

서로 독립적인 워커들을 동시에 실행합니다. 위 코드에서 asyncio.gather를 사용한 부분입니다.

문서의 감정 분석, 키워드 추출, 요약을 동시에 하면 시간을 크게 절약할 수 있습니다. 세 번째는 조건부 분기입니다.

이전 단계의 결과에 따라 다른 워커를 호출합니다. 예를 들어 고객 문의가 영어면 영어 응답 워커를, 한국어면 한국어 응답 워커를 호출합니다.

김개발 씨가 코드를 자세히 살펴보았습니다. context라는 딕셔너리가 모든 단계에서 공유됩니다.

각 워커의 결과가 context에 저장되고, 다음 워커가 이를 참조합니다. 이렇게 하면 워커 간 데이터 전달이 명확해집니다.

박시니어 씨가 실제 워크플로 정의 예시를 보여주었습니다. "이 워크플로를 보세요.

먼저 문서 파싱을 하고, 그 다음 감정 분석과 키워드 추출을 병렬로 합니다. 그리고 감정이 부정적이면 위기 대응 워커를, 그렇지 않으면 일반 응답 워커를 호출해요." 워크플로를 코드가 아닌 데이터로 정의하면 장점이 많습니다.

코드를 수정하지 않고 워크플로를 변경할 수 있습니다. JSON이나 YAML 파일로 워크플로를 관리하면 비개발자도 수정할 수 있습니다.

오류 처리 전략도 중요합니다. 병렬 실행 중 일부 워커가 실패하면 어떻게 할까요?

asyncio.gather의 return_exceptions=True 옵션을 사용하면 실패한 워커의 예외를 결과로 받을 수 있습니다. 이를 확인하여 적절히 처리합니다.

반복 패턴도 있습니다. 예를 들어 답변이 품질 기준을 충족할 때까지 재생성하는 경우입니다.

품질 평가 워커가 "불합격"을 반환하면 답변 생성 워커를 다시 호출합니다. 단, 무한 반복을 방지하기 위해 최대 반복 횟수를 설정해야 합니다.

김개발 씨가 정리했습니다. "결국 복잡한 워크플로도 기본 패턴의 조합이네요." 맞습니다.

순차, 병렬, 조건부라는 세 가지 기본 패턴을 이해하면 어떤 복잡한 워크플로도 설계할 수 있습니다. 마지막으로 박시니어 씨가 조언했습니다.

"처음부터 복잡하게 만들지 마세요. 단순한 순차 워크플로로 시작하고, 필요할 때 병렬과 조건부를 추가하세요."

실전 팁

💡 - 워크플로를 JSON/YAML로 외부화하면 유연성이 높아집니다

병렬 실행 시 return_exceptions=True로 부분 실패를 처리하세요

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

#Python#Orchestrator#Worker#AIAgent#DesignPattern#Python,AI,LLM

오케스트레이터-워커 패턴 완벽 가이드

# 오케스트레이터-워커 패턴 완벽 가이드 AI 에이전트 시스템에서 복잡한 작업을 효율적으로 분배하고 처리하는 오케스트레이터-워커 패턴의 개념과 활용법을 초급 개발자도 이해할 수 있도록 쉽게 설명합니다. --- ## 목차 1. [오케스트레이터-워커_패턴이란](#오케스트레이터-워커-패턴이란) 2. [오케스트레이터의_역할](#오케스트레이터의-역할) 3. [워커의_역할과_종류](#워커의-역할과-종류) 4. [패턴의_장점](#패턴의-장점) 5. [동적_작업_분배_전략](#동적-작업-분배-전략) 6. [주요_활용_사례_분석](#주요-활용-사례-분석) 7. [복잡한_워크플로_설계하기](#복잡한-워크플로-설계하기) --- ## 1. 오케스트레이터-워커_패턴이란 어느 날 김개발 씨가 회사에서 AI 기반 문서 분석 시스템을 개발하게 되었습니다. 수백 개의 문서를 동시에 분석하고, 요약하고, 번역까지 해야 하는 복잡한 요구사항이었습니다. "이걸 어떻게 효율적으로 처리하지?" 고민하던 김개발 씨에게 선배가 한마디 던졌습니다. "오케스트레이터-워커 패턴을 써봐." **오케스트레이터-워커 패턴**은 한마디로 지휘자와 연주자의 관계와 같습니다. 오케스트라에서 지휘자가 악보를 해석하고 각 연주자에게 역할을 지시하듯이, 오케스트레이터는 복잡한 작업을 분석하여 여러 워커에게 적절히 분배합니다. 이 패턴을 사용하면 복잡한 AI 워크플로우를 체계적으로 관리하고, 각 작업을 병렬로 처리할 수 있습니다. 다음 코드를 살펴봅시다. ```python # 오케스트레이터-워커 패턴의 기본 구조 class Orchestrator: def __init__(self): self.workers = {} # 워커들을 저장할 딕셔너리 def register_worker(self, task_type, worker): # 작업 유형별로 워커를 등록합니다 self.workers[task_type] = worker def process(self, task): # 작업을 분석하여 적절한 워커에게 분배합니다 task_type = self.analyze_task(task) worker = self.workers.get(task_type) return worker.execute(task) def analyze_task(self, task): # 작업의 유형을 판단하는 핵심 로직 return task.get("type", "default") ``` 김개발 씨는 입사 1년 차 주니어 개발자입니다. 최근 회사에서 AI 기반 고객 서비스 자동화 프로젝트를 맡게 되었는데, 문제가 복잡했습니다. 고객 문의가 들어오면 내용을 분석하고, 적절한 답변을 생성하고, 필요하면 번역까지 해야 했습니다. 처음에 김개발 씨는 모든 로직을 하나의 거대한 함수에 넣었습니다. 분석도 하고, 답변도 생성하고, 번역도 하는 만능 함수였습니다. 하지만 코드는 점점 복잡해졌고, 새로운 기능을 추가할 때마다 기존 코드가 망가지는 일이 반복되었습니다. 그때 선배 개발자 박시니어 씨가 다가와 물었습니다. "혹시 오케스트레이터-워커 패턴 들어봤어요?" 박시니어 씨는 화이트보드에 그림을 그리기 시작했습니다. "오케스트라 공연을 생각해봐요. 지휘자가 모든 악기를 직접 연주하진 않잖아요. 대신 악보를 해석해서 바이올린 주자에겐 바이올린 파트를, 첼로 주자에겐 첼로 파트를 지시하죠." 김개발 씨의 눈이 반짝였습니다. "아, 그러니까 제 코드에서도 작업을 지시하는 역할과 실제로 수행하는 역할을 분리하라는 거군요!" 맞습니다. **오케스트레이터**는 들어온 요청을 분석하고, 어떤 작업이 필요한지 판단합니다. 그리고 각 작업에 적합한 **워커**에게 일을 맡깁니다. 워커는 자신이 맡은 일만 집중해서 처리합니다. 이 패턴의 핵심은 **관심사의 분리**입니다. 오케스트레이터는 전체 흐름을 관리하는 데 집중하고, 워커는 특정 작업을 잘 수행하는 데 집중합니다. 마치 회사에서 팀장이 업무를 분배하고 팀원들이 각자 맡은 일을 처리하는 것과 같습니다. 위의 코드를 살펴보면, Orchestrator 클래스가 중앙에서 모든 것을 조율합니다. register_worker 메서드로 작업 유형별로 워커를 등록해두고, process 메서드에서 들어온 작업을 분석하여 적절한 워커에게 전달합니다. analyze_task 메서드가 특히 중요합니다. 이 부분에서 LLM을 활용하면 자연어로 된 요청을 분석하여 어떤 워커가 처리해야 할지 동적으로 판단할 수 있습니다. 단순한 규칙 기반이 아니라 맥락을 이해하는 지능적인 분배가 가능해지는 것입니다. 실무에서 이 패턴은 다양하게 활용됩니다. 챗봇 서비스에서 사용자 질문을 분석하여 FAQ 워커, 예약 워커, 불만 처리 워커 등에 분배할 수 있습니다. 데이터 파이프라인에서는 데이터 유형에 따라 전처리 워커, 분석 워커, 저장 워커를 선택적으로 호출합니다. 주의할 점도 있습니다. 오케스트레이터에 너무 많은 로직을 넣으면 안 됩니다. 오케스트레이터는 교통 경찰처럼 방향만 지시하고, 실제 운전은 각 차량이 해야 합니다. 또한 워커 간의 의존성이 생기지 않도록 설계해야 합니다. 다시 김개발 씨의 이야기로 돌아가면, 그는 이 패턴을 적용한 후 코드가 훨씬 깔끔해졌습니다. 새로운 기능을 추가할 때도 새 워커만 만들어서 등록하면 되었습니다. "이제야 제대로 된 구조가 잡힌 것 같아요!" **실전 팁** 💡 - 오케스트레이터는 가볍게 유지하고, 복잡한 비즈니스 로직은 워커에 넣으세요 - 워커는 독립적으로 테스트할 수 있도록 설계하면 유지보수가 쉬워집니다 --- ## 2. 오케스트레이터의_역할 김개발 씨가 오케스트레이터-워커 패턴의 기본 개념을 이해한 후, 다음 질문이 떠올랐습니다. "그런데 오케스트레이터는 정확히 어떤 일을 하는 거죠? 그냥 작업을 분배만 하면 되나요?" 박시니어 씨가 웃으며 대답했습니다. "그게 다가 아니에요. 오케스트레이터의 역할은 생각보다 다양해요." **오케스트레이터**는 시스템의 두뇌 역할을 합니다. 단순히 작업을 분배하는 것을 넘어, 작업을 분석하고, 실행 계획을 수립하며, 워커들의 결과를 수집하여 최종 응답을 조합합니다. 마치 영화 감독이 시나리오를 해석하고 배우들에게 연기 지시를 내린 후 편집까지 총괄하는 것과 같습니다. 다음 코드를 살펴봅시다. ```python import asyncio from openai import OpenAI class SmartOrchestrator: def __init__(self, llm_client): self.llm = llm_client self.workers = {} self.execution_history = [] async def plan_and_execute(self, user_request): # 1단계: LLM으로 작업 계획 수립 plan = await self.create_execution_plan(user_request) # 2단계: 계획에 따라 워커들 실행 results = [] for step in plan["steps"]: worker = self.workers[step["worker_type"]] result = await worker.execute(step["task"]) results.append(result) # 3단계: 결과 종합하여 최종 응답 생성 final_response = await self.synthesize_results(results) return final_response ``` 김개발 씨는 오케스트레이터가 단순한 라우터가 아니라는 것을 깨달았습니다. 박시니어 씨는 칠판에 오케스트레이터의 네 가지 핵심 역할을 적었습니다. 첫 번째는 **작업 분석**입니다. 사용자의 요청이 들어오면 오케스트레이터는 그것이 무엇을 원하는 것인지 파악해야 합니다. "이 PDF 문서를 요약해줘"라는 요청이 들어오면, 단순 요약인지 핵심 포인트 추출인지 판단해야 합니다. 여기서 LLM의 자연어 이해 능력이 빛을 발합니다. 두 번째는 **실행 계획 수립**입니다. 복잡한 요청은 여러 단계로 나눠야 합니다. 예를 들어 "이 영문 논문을 번역하고 요약해줘"라는 요청은 먼저 번역 워커가 작업하고, 그 결과를 요약 워커에게 전달해야 합니다. 오케스트레이터는 이런 순서와 의존 관계를 계획합니다. 세 번째는 **워커 조율**입니다. 계획이 수립되면 적절한 워커들을 호출합니다. 병렬로 실행할 수 있는 작업은 동시에 처리하고, 순차적으로 처리해야 하는 작업은 순서를 지킵니다. 위 코드의 plan_and_execute 메서드가 바로 이 역할을 합니다. 네 번째는 **결과 종합**입니다. 각 워커가 반환한 결과를 모아서 하나의 의미 있는 응답으로 만들어야 합니다. 단순히 결과를 이어붙이는 것이 아니라, 맥락에 맞게 정리하고 사용자가 이해하기 쉽게 가공합니다. 박시니어 씨가 덧붙였습니다. "오케스트레이터는 또한 **오류 처리**도 담당해요. 워커가 실패하면 다른 워커로 대체하거나, 사용자에게 적절한 안내를 해야 하죠." 실제로 위 코드를 보면 create_execution_plan 메서드에서 LLM을 활용하여 실행 계획을 동적으로 생성합니다. 이것이 규칙 기반 시스템과의 큰 차이점입니다. LLM 덕분에 미리 정의하지 않은 새로운 유형의 요청도 유연하게 처리할 수 있습니다. 김개발 씨가 질문했습니다. "그러면 오케스트레이터가 너무 무거워지지 않나요?" 좋은 질문입니다. 오케스트레이터는 의사결정과 조율에만 집중해야 합니다. 실제 데이터 처리나 복잡한 로직은 워커에게 맡겨야 합니다. 마치 회사의 CEO가 모든 실무를 직접 처리하지 않는 것과 같습니다. CEO는 전략을 수립하고 팀을 조율하며 결과를 검토합니다. 실제 업무는 각 팀이 수행합니다. 오케스트레이터도 마찬가지입니다. execution_history를 저장하는 것도 주목할 만합니다. 실행 이력을 기록해두면 디버깅이 쉬워지고, 어떤 워커가 얼마나 자주 호출되는지 분석할 수도 있습니다. **실전 팁** 💡 - LLM을 활용한 동적 계획 수립으로 유연성을 확보하세요 - 오케스트레이터에 실행 이력을 남겨두면 문제 추적이 쉬워집니다 --- ## 3. 워커의_역할과_종류 오케스트레이터의 역할을 배운 김개발 씨가 이번엔 워커에 관심을 가졌습니다. "그럼 워커는 어떻게 설계해야 하나요? 그냥 함수 하나면 되나요?" 박시니어 씨가 고개를 저었습니다. "워커도 제대로 설계해야 해요. 종류도 다양하고, 각자 특성이 달라요." **워커**는 실제로 작업을 수행하는 전문가입니다. 각 워커는 한 가지 작업에 특화되어 있으며, 독립적으로 동작합니다. 마치 병원에서 내과 전문의, 외과 전문의, 소아과 전문의가 각자 전문 분야에서 환자를 치료하는 것과 같습니다. 워커는 크게 LLM 기반 워커, 도구 기반 워커, 하이브리드 워커로 나눌 수 있습니다. 다음 코드를 살펴봅시다. ```python from abc import ABC, abstractmethod # 워커의 기본 인터페이스 class BaseWorker(ABC): @abstractmethod async def execute(self, task: dict) -> dict: pass # LLM 기반 워커: 텍스트 생성, 요약 등 class SummarizerWorker(BaseWorker): def __init__(self, llm_client): self.llm = llm_client async def execute(self, task: dict) -> dict: prompt = f"다음 내용을 요약해주세요: {task['content']}" response = await self.llm.generate(prompt) return {"summary": response, "status": "success"} # 도구 기반 워커: API 호출, 데이터베이스 조회 등 class DatabaseWorker(BaseWorker): def __init__(self, db_connection): self.db = db_connection async def execute(self, task: dict) -> dict: query = task.get("query") result = await self.db.execute(query) return {"data": result, "status": "success"} ``` 박시니어 씨는 워커의 세 가지 유형을 설명하기 시작했습니다. 첫 번째는 **LLM 기반 워커**입니다. 이 워커는 언어 모델을 활용하여 텍스트 생성, 요약, 번역, 감정 분석 등을 수행합니다. 위 코드의 SummarizerWorker가 좋은 예시입니다. 들어온 텍스트를 LLM에 전달하고 결과를 반환합니다. 두 번째는 **도구 기반 워커**입니다. 이 워커는 외부 시스템과 상호작용합니다. 데이터베이스 조회, API 호출, 파일 시스템 접근 등이 여기에 해당합니다. DatabaseWorker처럼 LLM 없이도 동작하며, 특정 시스템과의 연동에 집중합니다. 세 번째는 **하이브리드 워커**입니다. LLM과 도구를 함께 활용합니다. 예를 들어 검색 워커는 먼저 도구로 검색 API를 호출하고, 그 결과를 LLM으로 정리하여 반환할 수 있습니다. 김개발 씨가 물었습니다. "각 워커를 어떻게 잘 설계할 수 있을까요?" 박시니어 씨는 워커 설계의 핵심 원칙을 알려주었습니다. 첫째, **단일 책임 원칙**입니다. 각 워커는 한 가지 일만 해야 합니다. 요약 워커는 요약만, 번역 워커는 번역만 합니다. 여러 기능을 한 워커에 넣으면 재사용성이 떨어집니다. 둘째, **독립성**입니다. 워커끼리 직접 통신하면 안 됩니다. 모든 조율은 오케스트레이터가 담당합니다. 워커 A가 워커 B를 직접 호출하면 의존성이 생기고 테스트가 어려워집니다. 셋째, **일관된 인터페이스**입니다. 위 코드에서 BaseWorker 추상 클래스를 정의한 것처럼, 모든 워커는 동일한 인터페이스를 따라야 합니다. 그래야 오케스트레이터가 워커를 교체하거나 추가할 때 수정할 코드가 줄어듭니다. 넷째, **오류 처리**입니다. 각 워커는 자신의 오류를 적절히 처리해야 합니다. 외부 API 호출이 실패하면 재시도할지, 대체 값을 반환할지 결정해야 합니다. 실무에서 자주 만나는 워커 종류를 소개합니다. **검색 워커**는 벡터 데이터베이스나 검색 엔진에서 관련 정보를 찾아옵니다. **코드 실행 워커**는 사용자가 요청한 코드를 안전한 샌드박스에서 실행합니다. **이미지 분석 워커**는 멀티모달 LLM을 활용하여 이미지를 분석합니다. 김개발 씨가 정리했습니다. "결국 워커는 레고 블록 같은 거네요. 각 블록은 독립적이고, 조합하면 다양한 것을 만들 수 있는 거죠." 정확합니다. 잘 설계된 워커는 여러 프로젝트에서 재사용할 수 있습니다. **실전 팁** 💡 - 워커의 execute 메서드는 항상 dict를 반환하도록 표준화하세요 - 워커 내부에서 다른 워커를 직접 호출하지 마세요 --- ## 4. 패턴의_장점 김개발 씨가 오케스트레이터-워커 패턴을 프로젝트에 적용하기 시작했습니다. 몇 주 후, 팀 회고 시간에 김개발 씨가 발표했습니다. "이 패턴을 적용하고 나서 정말 많은 것이 좋아졌어요." 팀원들이 귀를 기울였습니다. "어떤 점이 좋아졌나요?" 오케스트레이터-워커 패턴은 **확장성**, **유연성**, **유지보수성**이라는 세 가지 핵심 장점을 제공합니다. 마치 잘 조직된 회사처럼, 새로운 팀원을 쉽게 추가할 수 있고, 역할 변경이 간편하며, 문제가 생겼을 때 원인을 빠르게 찾을 수 있습니다. 다음 코드를 살펴봅시다. ```python class ScalableOrchestrator: def __init__(self): self.workers = {} self.worker_pool = {} # 워커 풀로 확장성 확보 def add_worker(self, task_type: str, worker: BaseWorker): # 새 워커 추가가 간단합니다 - 확장성 self.workers[task_type] = worker def replace_worker(self, task_type: str, new_worker: BaseWorker): # 기존 워커 교체도 간단합니다 - 유연성 old_worker = self.workers.get(task_type) self.workers[task_type] = new_worker return old_worker async def execute_with_fallback(self, task_type: str, task: dict): # 실패 시 대체 워커 사용 - 견고성 try: primary_worker = self.workers[task_type] return await primary_worker.execute(task) except Exception as e: fallback_worker = self.workers.get(f"{task_type}_fallback") if fallback_worker: return await fallback_worker.execute(task) raise e ``` 김개발 씨는 발표 자료를 펼쳤습니다. 첫 번째 장점은 **확장성**입니다. "예전에는 새 기능을 추가하려면 기존 코드를 많이 수정해야 했어요. 하지만 이제는 새 워커만 만들어서 등록하면 됩니다." 위 코드의 add_worker 메서드를 보면, 새 워커를 추가하는 것이 얼마나 간단한지 알 수 있습니다. 팀원 이주니어 씨가 물었습니다. "정말요? 번역 기능을 추가할 때도 그랬나요?" 김개발 씨가 고개를 끄덕였습니다. "네, TranslatorWorker 클래스를 만들고 등록만 했어요. 기존 코드는 한 줄도 수정하지 않았습니다." 두 번째 장점은 **유연성**입니다. "워커를 교체하는 것도 쉬워요. 요약 워커가 GPT-3.5 기반이었는데, GPT-4로 바꾸고 싶었어요. replace_worker 메서드로 간단히 교체했습니다." 이것은 **개방-폐쇄 원칙**의 좋은 예시입니다. 확장에는 열려 있고 수정에는 닫혀 있는 구조입니다. 세 번째 장점은 **유지보수성**입니다. "버그가 발생했을 때 어디를 봐야 하는지 명확해요. 요약이 이상하면 요약 워커를 보면 되고, 번역이 이상하면 번역 워커를 보면 됩니다." 각 워커가 독립적이기 때문에 문제 격리가 쉽습니다. 네 번째 장점은 **테스트 용이성**입니다. "각 워커를 독립적으로 테스트할 수 있어요. 오케스트레이터 테스트에서는 목 워커를 사용하면 됩니다." 단위 테스트와 통합 테스트를 깔끔하게 분리할 수 있습니다. 다섯 번째 장점은 **견고성**입니다. 위 코드의 execute_with_fallback 메서드처럼, 주 워커가 실패하면 대체 워커를 사용할 수 있습니다. 시스템의 안정성이 높아집니다. 여섯 번째 장점은 **병렬 처리**입니다. 서로 독립적인 워커들은 동시에 실행할 수 있습니다. 요약과 번역이 동시에 필요하다면 병렬로 처리하여 응답 시간을 단축할 수 있습니다. 박시니어 씨가 추가했습니다. "팀 협업에도 좋아요. 각 개발자가 자기 담당 워커만 집중해서 개발할 수 있으니까요." 하지만 주의할 점도 있습니다. 작은 프로젝트에 이 패턴을 적용하면 오버엔지니어링이 될 수 있습니다. 워커가 3개 미만이라면 굳이 이 패턴을 적용하지 않아도 됩니다. **실전 팁** 💡 - 워커 추가 시 기존 오케스트레이터 코드를 수정할 필요가 없다면 잘 설계된 것입니다 - 대체 워커(fallback)를 미리 등록해두면 시스템 안정성이 높아집니다 --- ## 5. 동적_작업_분배_전략 김개발 씨의 시스템이 점점 복잡해졌습니다. 워커 종류가 10개가 넘어가자 새로운 고민이 생겼습니다. "사용자 요청이 들어올 때마다 어떤 워커를 써야 할지 어떻게 판단하지?" 하드코딩된 규칙으로는 한계가 있었습니다. 박시니어 씨가 해결책을 제시했습니다. "LLM을 활용한 동적 분배를 써봐요." **동적 작업 분배**는 사용자 요청의 의도를 실시간으로 분석하여 적절한 워커를 선택하는 전략입니다. 규칙 기반 분배가 정해진 키워드에만 반응한다면, 동적 분배는 문맥을 이해하고 유연하게 판단합니다. 마치 숙련된 상담원이 고객의 말을 듣고 적절한 부서로 연결해주는 것과 같습니다. 다음 코드를 살펴봅시다. ```python class DynamicOrchestrator: def __init__(self, llm_client): self.llm = llm_client self.workers = {} self.worker_descriptions = {} # 워커 설명 저장 def register_worker(self, name: str, worker: BaseWorker, description: str): self.workers[name] = worker self.worker_descriptions[name] = description async def route_request(self, user_request: str) -> str: # LLM에게 워커 목록과 설명을 제공하고 선택하게 합니다 prompt = f"""사용자 요청: {user_request} 사용 가능한 워커: {self._format_worker_list()} 가장 적합한 워커 이름을 선택하세요.""" selected_worker = await self.llm.generate(prompt) return selected_worker.strip() def _format_worker_list(self) -> str: return "\n".join([ f"- {name}: {desc}" for name, desc in self.worker_descriptions.items() ]) ``` 박시니어 씨가 화이트보드에 두 가지 분배 방식을 그렸습니다. 첫 번째는 **규칙 기반 분배**입니다. "요청에 '요약'이라는 단어가 있으면 요약 워커를 호출해"와 같은 방식입니다. 구현이 간단하지만 한계가 명확합니다. "이 글의 핵심만 알려줘"라고 하면 요약 워커를 호출해야 하는데, '요약'이라는 단어가 없어서 실패합니다. 두 번째는 **LLM 기반 동적 분배**입니다. LLM에게 사용자 요청과 워커 목록을 주고 판단하게 합니다. LLM은 "핵심만 알려줘"가 요약 요청이라는 것을 이해합니다. 위 코드의 route_request 메서드가 이 방식을 구현합니다. 김개발 씨가 질문했습니다. "매번 LLM을 호출하면 비용이 너무 많이 들지 않나요?" 좋은 지적입니다. 몇 가지 최적화 전략이 있습니다. **캐싱**을 활용하면 비슷한 요청에 대한 라우팅 결과를 재사용할 수 있습니다. **작은 모델 사용**도 효과적입니다. 라우팅은 복잡한 추론이 필요 없으므로 GPT-3.5 같은 가벼운 모델로도 충분합니다. 또한 **하이브리드 전략**도 있습니다. 먼저 규칙 기반으로 시도하고, 규칙에 맞지 않으면 LLM을 호출합니다. 자주 들어오는 패턴은 규칙으로 빠르게 처리하고, 새로운 패턴만 LLM이 처리합니다. 위 코드에서 worker_descriptions가 중요합니다. 각 워커의 역할을 명확하게 설명해야 LLM이 정확한 판단을 내릴 수 있습니다. "요약 워커"보다 "긴 텍스트를 핵심 포인트 3-5개로 요약하는 워커"라고 설명하는 것이 좋습니다. **멀티 워커 선택**도 고려해야 합니다. 하나의 요청이 여러 워커를 필요로 할 수 있습니다. "이 문서를 번역하고 요약해줘"는 번역 워커와 요약 워커 둘 다 필요합니다. 이 경우 LLM이 실행 순서까지 결정하도록 프롬프트를 수정합니다. 오케스트레이터는 선택된 워커가 실제로 존재하는지 검증해야 합니다. LLM이 등록되지 않은 워커 이름을 반환할 수도 있기 때문입니다. 항상 워커 존재 여부를 확인하고, 없으면 기본 워커를 사용하거나 사용자에게 알려야 합니다. 김개발 씨가 코드를 수정한 후 테스트했습니다. "와, 이제 '간략하게 정리해줘'라고 해도 요약 워커가 호출되네요!" 동적 분배의 힘을 체감한 순간이었습니다. **실전 팁** 💡 - 워커 설명은 구체적으로 작성하세요. LLM의 판단 정확도가 올라갑니다 - 자주 사용되는 패턴은 캐싱하여 비용을 절감하세요 --- ## 6. 주요_활용_사례_분석 이론을 충분히 배운 김개발 씨가 물었습니다. "실제 현업에서는 이 패턴을 어떻게 쓰나요? 구체적인 사례가 궁금해요." 박시니어 씨가 노트북을 열었습니다. "마침 내가 참여했던 프로젝트 몇 개를 보여줄게요. 다양한 산업에서 이 패턴을 활용하고 있어요." 오케스트레이터-워커 패턴은 **고객 서비스 자동화**, **문서 처리 파이프라인**, **데이터 분석 시스템** 등 다양한 분야에서 활용됩니다. 특히 복잡한 작업을 여러 단계로 나눠 처리해야 하는 시스템에서 빛을 발합니다. 실제 사례를 통해 패턴의 실무 적용 방법을 살펴보겠습니다. 다음 코드를 살펴봅시다. ```python # 고객 서비스 자동화 시스템 예시 class CustomerServiceOrchestrator: def __init__(self): self.workers = { "intent_classifier": IntentClassifierWorker(), "faq_responder": FAQResponderWorker(), "order_tracker": OrderTrackerWorker(), "complaint_handler": ComplaintHandlerWorker(), "human_handoff": HumanHandoffWorker(), } async def handle_inquiry(self, customer_message: str) -> dict: # 1단계: 의도 분류 intent = await self.workers["intent_classifier"].execute({ "message": customer_message }) # 2단계: 의도에 맞는 워커 실행 handler = self.workers.get(intent["handler"]) if handler: response = await handler.execute({"message": customer_message}) else: response = await self.workers["human_handoff"].execute({ "message": customer_message, "reason": "unknown_intent" }) return response ``` 박시니어 씨가 첫 번째 사례를 소개했습니다. 첫 번째는 **고객 서비스 자동화 시스템**입니다. 위 코드가 바로 이 사례입니다. 고객 문의가 들어오면 먼저 IntentClassifierWorker가 의도를 파악합니다. FAQ 질문인지, 주문 추적인지, 불만 사항인지 분류합니다. 분류 결과에 따라 적절한 워커가 호출됩니다. FAQ 질문이면 FAQResponderWorker가 답변하고, 주문 추적이면 OrderTrackerWorker가 주문 상태를 조회합니다. 처리할 수 없는 복잡한 문의는 HumanHandoffWorker가 상담원에게 연결합니다. 김개발 씨가 물었습니다. "이렇게 하면 상담원이 필요 없어지나요?" 박시니어 씨가 고개를 저었습니다. "아니요, 단순 문의를 자동화해서 상담원이 복잡한 문제에 집중할 수 있게 하는 거예요." 두 번째는 **문서 처리 파이프라인**입니다. 법률 사무소에서 계약서를 분석하는 시스템을 예로 들어보겠습니다. PDF 파싱 워커가 문서를 텍스트로 변환합니다. 조항 추출 워커가 주요 조항을 식별합니다. 위험 분석 워커가 불리한 조항을 찾아냅니다. 요약 워커가 핵심 내용을 정리합니다. 각 워커가 순차적으로 실행되며, 이전 워커의 결과가 다음 워커의 입력이 됩니다. 오케스트레이터는 이 파이프라인을 관리하고, 중간에 오류가 발생하면 적절히 처리합니다. 세 번째는 **RAG 기반 질의응답 시스템**입니다. 검색 워커가 벡터 데이터베이스에서 관련 문서를 찾습니다. 리랭킹 워커가 검색 결과의 관련성을 재평가합니다. 답변 생성 워커가 검색된 맥락을 바탕으로 답변을 생성합니다. 인용 추가 워커가 출처를 명시합니다. 네 번째는 **코드 리뷰 자동화 시스템**입니다. 코드 파싱 워커가 PR의 변경 사항을 분석합니다. 스타일 검사 워커가 코딩 컨벤션 위반을 찾습니다. 보안 검사 워커가 취약점을 탐지합니다. 리뷰 생성 워커가 종합적인 리뷰 코멘트를 작성합니다. 김개발 씨가 감탄했습니다. "정말 다양한 곳에서 쓰이네요!" 핵심은 복잡한 작업을 의미 있는 단위로 분해하는 것입니다. 각 단위가 워커가 되고, 오케스트레이터가 이들을 조율합니다. 실무에서는 워커 간 데이터 전달 형식을 표준화하는 것이 중요합니다. 모든 워커가 동일한 구조의 딕셔너리를 주고받으면 조합이 쉬워집니다. **실전 팁** 💡 - 의도 분류 워커는 다른 모든 워커의 관문이므로 정확도가 중요합니다 - 처리할 수 없는 경우를 위한 대체 경로(human handoff)를 항상 준비하세요 --- ## 7. 복잡한_워크플로_설계하기 김개발 씨가 이제 자신감이 생겼습니다. 하지만 새로운 도전이 기다리고 있었습니다. "이번 프로젝트는 워커가 순차적으로만 실행되면 안 돼요. 어떤 건 병렬로, 어떤 건 조건부로 실행해야 해요." 박시니어 씨가 고개를 끄덕였습니다. "드디어 복잡한 워크플로를 배울 때가 됐네요." 복잡한 워크플로는 **순차 실행**, **병렬 실행**, **조건부 분기**, **반복**을 조합하여 설계합니다. 마치 복잡한 요리 레시피처럼, 어떤 단계는 순서대로, 어떤 단계는 동시에, 어떤 단계는 상황에 따라 건너뛰어야 합니다. 이를 체계적으로 관리하면 복잡한 비즈니스 로직도 깔끔하게 구현할 수 있습니다. 다음 코드를 살펴봅시다. ```python import asyncio from enum import Enum class StepType(Enum): SEQUENTIAL = "sequential" PARALLEL = "parallel" CONDITIONAL = "conditional" class WorkflowOrchestrator: def __init__(self): self.workers = {} async def execute_workflow(self, workflow: list, context: dict) -> dict: for step in workflow: if step["type"] == StepType.PARALLEL: # 병렬 실행: 여러 워커를 동시에 실행 tasks = [ self.workers[w].execute(context) for w in step["workers"] ] results = await asyncio.gather(*tasks) context["parallel_results"] = results elif step["type"] == StepType.CONDITIONAL: # 조건부 분기: 조건에 따라 다른 워커 실행 condition_result = step["condition"](context) worker_name = step["if_true"] if condition_result else step["if_false"] result = await self.workers[worker_name].execute(context) context[step["output_key"]] = result else: # SEQUENTIAL result = await self.workers[step["worker"]].execute(context) context[step["output_key"]] = result return context ``` 박시니어 씨가 복잡한 워크플로의 세 가지 패턴을 설명했습니다. 첫 번째는 **순차 실행**입니다. 워커 A의 결과가 워커 B의 입력이 되는 경우입니다. 문서 번역 후 요약하는 작업이 좋은 예입니다. 번역이 완료되어야 요약할 수 있으므로 순서가 중요합니다. 두 번째는 **병렬 실행**입니다. 서로 독립적인 워커들을 동시에 실행합니다. 위 코드에서 asyncio.gather를 사용한 부분입니다. 문서의 감정 분석, 키워드 추출, 요약을 동시에 하면 시간을 크게 절약할 수 있습니다. 세 번째는 **조건부 분기**입니다. 이전 단계의 결과에 따라 다른 워커를 호출합니다. 예를 들어 고객 문의가 영어면 영어 응답 워커를, 한국어면 한국어 응답 워커를 호출합니다. 김개발 씨가 코드를 자세히 살펴보았습니다. context라는 딕셔너리가 모든 단계에서 공유됩니다. 각 워커의 결과가 context에 저장되고, 다음 워커가 이를 참조합니다. 이렇게 하면 워커 간 데이터 전달이 명확해집니다. 박시니어 씨가 실제 워크플로 정의 예시를 보여주었습니다. "이 워크플로를 보세요. 먼저 문서 파싱을 하고, 그 다음 감정 분석과 키워드 추출을 병렬로 합니다. 그리고 감정이 부정적이면 위기 대응 워커를, 그렇지 않으면 일반 응답 워커를 호출해요." 워크플로를 코드가 아닌 데이터로 정의하면 장점이 많습니다. 코드를 수정하지 않고 워크플로를 변경할 수 있습니다. JSON이나 YAML 파일로 워크플로를 관리하면 비개발자도 수정할 수 있습니다. **오류 처리 전략**도 중요합니다. 병렬 실행 중 일부 워커가 실패하면 어떻게 할까요? asyncio.gather의 return_exceptions=True 옵션을 사용하면 실패한 워커의 예외를 결과로 받을 수 있습니다. 이를 확인하여 적절히 처리합니다. **반복 패턴**도 있습니다. 예를 들어 답변이 품질 기준을 충족할 때까지 재생성하는 경우입니다. 품질 평가 워커가 "불합격"을 반환하면 답변 생성 워커를 다시 호출합니다. 단, 무한 반복을 방지하기 위해 최대 반복 횟수를 설정해야 합니다. 김개발 씨가 정리했습니다. "결국 복잡한 워크플로도 기본 패턴의 조합이네요." 맞습니다. 순차, 병렬, 조건부라는 세 가지 기본 패턴을 이해하면 어떤 복잡한 워크플로도 설계할 수 있습니다. 마지막으로 박시니어 씨가 조언했습니다. "처음부터 복잡하게 만들지 마세요. 단순한 순차 워크플로로 시작하고, 필요할 때 병렬과 조건부를 추가하세요." **실전 팁** 💡 - 워크플로를 JSON/YAML로 외부화하면 유연성이 높아집니다 - 병렬 실행 시 return_exceptions=True로 부분 실패를 처리하세요 --- 이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

카테고리: Python,AI,LLM

언어: Python

태그: Python, Orchestrator, Worker, AIAgent, DesignPattern, Python,AI,LLM

작성자: AI Generated

프리미엄 콘텐츠 - 3개월 무료 체험 가능

CodeDeck

오케스트레이터-워커 패턴 완벽 가이드

목차

1. 오케스트레이터-워커 패턴이란

2. 오케스트레이터의 역할

3. 워커의 역할과 종류

4. 패턴의 장점

5. 동적 작업 분배 전략

6. 주요 활용 사례 분석

7. 복잡한 워크플로 설계하기

이 카드뉴스가 포함된 코스

파이썬으로 만드는 AI 에이전트

댓글 (0)

함께 보면 좋은 카드 뉴스

Helm 마이크로서비스 패키징 완벽 가이드

보안 아키텍처 구성 완벽 가이드

AWS Organizations 완벽 가이드

AWS KMS 암호화 완벽 가이드

AWS Secrets Manager 완벽 가이드