오케스트레이터-워커 에이전트 구현 완벽 가이드

AI 에이전트 패턴 중 가장 실용적인 오케스트레이터-워커 패턴을 단계별로 구현합니다. 복잡한 작업을 여러 워커에게 분배하고 결과를 수집하는 방법을 배웁니다.

---

목차

1. 오케스트레이터_클래스_설계
2. 워커_클래스_구현
3. 작업_분배_로직_작성
4. 워커_실행과_결과_수집
5. 오케스트레이터_종합_로직
6. 스트림릿_UI_구현
7. 에이전트_테스트_및_개선

---

1. 오케스트레이터 클래스 설계

김개발 씨는 요즘 AI 에이전트 개발에 푹 빠져 있습니다. 그런데 하나의 에이전트가 모든 일을 처리하다 보니 코드가 점점 복잡해지고 있었습니다.

"이렇게 하면 안 될 것 같은데..."라며 고민하던 중, 선배가 다가와 말했습니다. "오케스트레이터-워커 패턴을 써보는 게 어때?"

오케스트레이터는 한마디로 지휘자입니다. 오케스트라의 지휘자가 각 악기 파트에 연주를 지시하듯, 오케스트레이터는 복잡한 작업을 분석하고 적절한 워커들에게 일을 분배합니다.

이 패턴을 이해하면 확장 가능하고 유지보수하기 쉬운 AI 시스템을 설계할 수 있습니다.

다음 코드를 살펴봅시다.

from anthropic import Anthropic
from dataclasses import dataclass, field
from typing import List, Dict, Any

@dataclass
class Orchestrator:
    """작업을 분석하고 워커들에게 분배하는 지휘자 클래스"""
    client: Anthropic = field(default_factory=Anthropic)
    model: str = "claude-sonnet-4-20250514"
    workers: List['Worker'] = field(default_factory=list)

    def analyze_task(self, task: str) -> List[Dict[str, Any]]:
        """복잡한 작업을 분석하여 하위 작업으로 분해합니다"""
        # 하위 작업 분해 로직은 다음 섹션에서 구현
        pass

    def distribute_tasks(self, subtasks: List[Dict]) -> Dict[str, Any]:
        """분해된 작업을 적절한 워커에게 분배합니다"""
        pass

김개발 씨는 입사 6개월 차 주니어 개발자입니다. 최근 회사에서 AI 챗봇 프로젝트를 맡게 되었는데, 챗봇이 해야 할 일이 점점 많아지면서 코드가 스파게티처럼 엉켜가고 있었습니다.

고객 문의 응답, 데이터 분석, 보고서 작성까지 하나의 에이전트가 모든 것을 처리하려니 도저히 감당이 되지 않았습니다. 그때 옆자리 선배 박시니어 씨가 김개발 씨의 화면을 슬쩍 보더니 말했습니다.

"개발 씨, 혹시 오케스트레이터-워커 패턴 들어봤어요?" 그렇다면 오케스트레이터란 정확히 무엇일까요? 쉽게 비유하자면, 오케스트레이터는 마치 오케스트라의 지휘자와 같습니다.

지휘자는 직접 악기를 연주하지 않습니다. 대신 악보를 분석하고, 바이올린 파트에는 이렇게, 첼로 파트에는 저렇게 연주하라고 지시합니다.

각 파트가 제 역할을 수행하면 지휘자가 이를 조율하여 아름다운 교향곡이 완성됩니다. AI 에이전트 세계에서도 마찬가지입니다.

오케스트레이터는 복잡한 요청을 받으면 이를 분석합니다. "이 부분은 데이터 분석 워커가 처리하고, 저 부분은 텍스트 생성 워커가 담당하면 되겠군." 이렇게 작업을 분배하고 최종 결과를 취합하는 것이 오케스트레이터의 역할입니다.

오케스트레이터가 없던 시절에는 어땠을까요? 개발자들은 하나의 거대한 에이전트에 모든 기능을 넣어야 했습니다.

코드는 수천 줄로 늘어나고, 작은 수정 하나에도 전체 시스템을 테스트해야 했습니다. 더 큰 문제는 확장성이었습니다.

새로운 기능을 추가할 때마다 기존 코드와의 충돌을 걱정해야 했고, 프로젝트가 커질수록 유지보수는 악몽이 되었습니다. 바로 이런 문제를 해결하기 위해 오케스트레이터-워커 패턴이 등장했습니다.

이 패턴을 사용하면 관심사의 분리가 가능해집니다. 각 워커는 자신의 전문 분야만 담당하므로 코드가 깔끔해집니다.

또한 확장성도 얻을 수 있습니다. 새로운 기능이 필요하면 새 워커만 추가하면 됩니다.

무엇보다 테스트 용이성이라는 큰 이점이 있습니다. 각 워커를 독립적으로 테스트할 수 있으니까요.

위의 코드를 한 줄씩 살펴보겠습니다. 먼저 dataclass 데코레이터를 사용한 것을 볼 수 있습니다.

Python의 dataclass는 클래스 정의를 간결하게 만들어줍니다. 생성자, repr, 비교 메서드 등을 자동으로 생성해주기 때문입니다.

client 필드는 Anthropic API 클라이언트입니다. field(default_factory=Anthropic)을 사용하면 인스턴스가 생성될 때마다 새로운 클라이언트가 만들어집니다.

이렇게 하는 이유는 mutable default argument 문제를 피하기 위해서입니다. workers 필드는 이 오케스트레이터가 관리하는 워커들의 목록입니다.

처음에는 빈 리스트로 시작하고, 나중에 워커를 등록하게 됩니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 고객 서비스 챗봇을 개발한다고 가정해봅시다. 고객이 "지난달 주문 내역을 분석해서 추천 상품 목록을 보여줘"라고 요청하면, 오케스트레이터는 이를 세 가지 하위 작업으로 분해합니다.

주문 내역 조회, 패턴 분석, 추천 생성. 각각의 전문 워커가 자기 일을 처리하고, 오케스트레이터가 결과를 조합하여 최종 응답을 만듭니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 오케스트레이터에 너무 많은 로직을 넣는 것입니다.

오케스트레이터는 말 그대로 조율자여야 합니다. 실제 작업 처리 로직은 워커에게 맡기세요.

오케스트레이터가 비대해지면 이 패턴을 쓰는 의미가 없어집니다. 박시니어 씨의 설명을 들은 김개발 씨는 고개를 끄덕였습니다.

"아, 그래서 설계가 중요하군요!"

실전 팁

💡 - dataclass의 field(default_factory=...)를 사용하면 mutable default 문제를 피할 수 있습니다

• 오케스트레이터는 최대한 가볍게 유지하고, 실제 로직은 워커에게 위임하세요
• 타입 힌트를 적극 활용하면 코드 가독성과 IDE 지원이 좋아집니다

---

2. 워커 클래스 구현

오케스트레이터 설계를 마친 김개발 씨는 이제 실제로 일을 수행할 워커 클래스를 만들어야 합니다. "지휘자는 만들었으니, 이제 연주자들을 만들 차례군요." 박시니어 씨가 옆에서 힌트를 줍니다.

"워커는 자기 전문 분야에만 집중하게 만들어야 해요."

워커는 오케스트레이터로부터 할당받은 특정 작업을 실제로 수행하는 클래스입니다. 마치 공장의 전문 기술자처럼, 각 워커는 자신만의 전문 영역을 가지고 해당 작업을 처리합니다.

워커를 잘 설계하면 재사용성이 높아지고 테스트도 쉬워집니다.

다음 코드를 살펴봅시다.

@dataclass
class Worker:
    """특정 작업을 전문적으로 수행하는 워커 클래스"""
    name: str
    specialty: str  # 워커의 전문 분야
    client: Anthropic = field(default_factory=Anthropic)
    model: str = "claude-sonnet-4-20250514"
    system_prompt: str = ""

    def __post_init__(self):
        """워커 초기화 후 시스템 프롬프트 설정"""
        if not self.system_prompt:
            self.system_prompt = f"당신은 {self.specialty} 전문가입니다."

    def execute(self, task: str) -> str:
        """할당받은 작업을 실행하고 결과를 반환합니다"""
        response = self.client.messages.create(
            model=self.model,
            max_tokens=2048,
            system=self.system_prompt,
            messages=[{"role": "user", "content": task}]
        )
        return response.content[0].text

오케스트레이터가 지휘자라면, 워커는 실제로 악기를 연주하는 연주자입니다. 김개발 씨는 이제 이 연주자들을 어떻게 만들어야 할지 고민에 빠졌습니다.

박시니어 씨가 화이트보드에 그림을 그리며 설명했습니다. "워커를 설계할 때 가장 중요한 건 단일 책임 원칙이에요.

하나의 워커는 하나의 일만 잘하면 됩니다." 그렇다면 좋은 워커란 어떤 것일까요? 쉽게 비유하자면, 워커는 마치 전문 요리사와 같습니다.

이탈리안 셰프는 파스타와 피자를 만들고, 일식 셰프는 스시와 라멘을 만듭니다. 각자 자기 분야에서 최고의 실력을 발휘합니다.

만약 한 요리사가 모든 요리를 다 하려 한다면 어떻게 될까요? 아마 어떤 요리도 제대로 만들지 못할 것입니다.

워커도 마찬가지입니다. 데이터 분석 워커는 데이터 분석만, 텍스트 생성 워커는 텍스트 생성만 담당합니다.

이렇게 역할을 명확히 나누면 각 워커의 성능을 최적화하기 쉽습니다. 코드를 살펴보면, name과 specialty 필드가 눈에 띕니다.

name은 워커를 식별하는 데 사용되고, specialty는 이 워커가 무엇을 잘하는지 정의합니다. 이 정보는 나중에 오케스트레이터가 작업을 분배할 때 중요한 기준이 됩니다.

post_init 메서드는 dataclass의 특별한 기능입니다. 객체가 생성된 직후에 자동으로 호출됩니다.

여기서는 system_prompt가 비어있으면 기본값을 설정하는 용도로 사용했습니다. 이런 초기화 로직을 넣기에 딱 좋은 곳입니다.

execute 메서드가 워커의 핵심입니다. Anthropic API를 호출하여 실제 작업을 수행합니다.

system_prompt를 통해 워커의 역할과 전문성을 LLM에게 알려주고, 할당받은 task를 처리합니다. 실제 프로젝트에서는 다양한 워커를 만들 수 있습니다.

예를 들어 코드 리뷰 시스템을 만든다면 이런 워커들이 필요할 것입니다. 보안 취약점을 찾는 SecurityWorker, 성능 문제를 분석하는 PerformanceWorker, 코드 스타일을 검사하는 StyleWorker.

각각의 워커가 자기 분야에서 최선을 다하고, 오케스트레이터가 이를 종합하면 완성도 높은 코드 리뷰가 탄생합니다. 하지만 주의할 점이 있습니다.

워커의 system_prompt를 설계할 때 너무 광범위하게 만들면 안 됩니다. "당신은 모든 것을 잘하는 전문가입니다"라고 하면 아무것도 잘하지 못하는 워커가 됩니다.

구체적이고 명확한 역할 정의가 중요합니다. 김개발 씨가 물었습니다.

"그러면 워커마다 다른 모델을 쓸 수도 있나요?" 박시니어 씨가 고개를 끄덕였습니다. "물론이죠.

간단한 작업에는 가벼운 모델을, 복잡한 작업에는 강력한 모델을 쓰면 비용도 아끼고 속도도 빨라져요."

실전 팁

💡 - 워커의 specialty는 구체적으로 정의하세요. "분석 전문가"보다 "고객 행동 패턴 분석 전문가"가 좋습니다

• system_prompt에 예시를 포함하면 워커의 출력 품질이 향상됩니다
• 작업 특성에 따라 워커별로 다른 모델이나 max_tokens를 설정할 수 있습니다

---

3. 작업 분배 로직 작성

워커 클래스까지 만든 김개발 씨는 이제 핵심 중의 핵심인 작업 분배 로직을 구현해야 합니다. "지휘자가 악보를 보고 각 파트에 지시를 내리듯이, 오케스트레이터도 작업을 분석해서 워커들에게 나눠줘야 해요." 박시니어 씨의 말에 김개발 씨는 펜을 들었습니다.

작업 분배 로직은 오케스트레이터의 두뇌라고 할 수 있습니다. 복잡한 요청을 받으면 이를 분석하여 어떤 하위 작업으로 나눌지, 각 작업은 어떤 워커가 처리해야 할지 결정합니다.

LLM의 추론 능력을 활용하면 이 과정을 자동화할 수 있습니다.

다음 코드를 살펴봅시다.

import json

def analyze_task(self, task: str) -> List[Dict[str, Any]]:
    """복잡한 작업을 분석하여 하위 작업으로 분해합니다"""
    worker_info = [{"name": w.name, "specialty": w.specialty}
                   for w in self.workers]

    analysis_prompt = f"""다음 작업을 분석하여 하위 작업으로 분해해주세요.

가용 워커 목록: {json.dumps(worker_info, ensure_ascii=False)}

작업: {task}

JSON 형식으로 응답해주세요:
[{{"worker_name": "워커이름", "subtask": "하위작업 설명"}}]"""

    response = self.client.messages.create(
        model=self.model, max_tokens=1024,
        messages=[{"role": "user", "content": analysis_prompt}]
    )
    # JSON 파싱하여 하위 작업 목록 반환
    return json.loads(response.content[0].text)

오케스트레이터-워커 패턴에서 가장 까다로운 부분은 바로 작업 분배입니다. 김개발 씨도 이 부분에서 한참을 고민했습니다.

"어떻게 하면 작업을 똑똑하게 나눌 수 있을까?" 박시니어 씨가 커피를 한 모금 마시며 말했습니다. "여기서 LLM의 진가가 발휘되는 거예요.

사람이 규칙을 일일이 정하는 대신, LLM에게 작업 분석을 맡기는 거죠." 이것이 바로 LLM 기반 작업 분해의 핵심입니다. 쉽게 비유하자면, 마치 경험 많은 프로젝트 매니저와 같습니다.

신입사원이 "이 기능 개발해주세요"라는 막연한 요청을 받으면 당황합니다. 하지만 경험 많은 PM은 이를 "DB 스키마 설계", "API 개발", "프론트엔드 구현", "테스트 작성" 같은 구체적인 작업으로 나눌 수 있습니다.

LLM이 바로 이 PM 역할을 합니다. 코드를 자세히 살펴보겠습니다.

먼저 worker_info를 만드는 부분이 있습니다. 현재 등록된 모든 워커의 이름과 전문 분야를 수집합니다.

이 정보가 있어야 LLM이 "아, 데이터 분석은 이 워커한테, 텍스트 생성은 저 워커한테 맡기면 되겠구나"라고 판단할 수 있습니다. analysis_prompt가 핵심입니다.

여기서 LLM에게 세 가지 정보를 제공합니다. 첫째, 어떤 워커들이 있는지.

둘째, 처리해야 할 작업이 무엇인지. 셋째, 어떤 형식으로 응답해야 하는지.

특히 JSON 형식을 명시한 것이 중요합니다. 구조화된 응답을 받아야 프로그램에서 처리하기 쉽기 때문입니다.

실무에서는 이 분석 로직을 더 정교하게 만들 수 있습니다. 예를 들어 작업 간의 의존성을 고려할 수 있습니다.

"데이터 수집이 끝나야 분석을 시작할 수 있다"는 관계를 LLM이 파악하게 하는 것이죠. 또한 작업의 우선순위를 정하거나, 예상 소요 시간을 추정하게 할 수도 있습니다.

하지만 주의할 점이 있습니다. LLM의 응답이 항상 완벽한 JSON인 것은 아닙니다.

가끔 추가 설명을 붙이거나 형식이 살짝 어긋날 수 있습니다. 따라서 실제 프로덕션 코드에서는 JSON 파싱에 대한 에러 처리와 검증 로직이 필요합니다.

또한 워커가 없는 작업이 분배되는 경우도 대비해야 합니다. 김개발 씨가 코드를 작성하다가 물었습니다.

"만약 하나의 작업을 여러 워커가 협력해서 처리해야 하면 어떻게 하죠?" 박시니어 씨가 미소를 지었습니다. "좋은 질문이에요.

그건 다음 단계인 결과 수집에서 다루게 될 거예요. 워커들의 결과를 조합하는 것도 오케스트레이터의 중요한 역할이거든요."

실전 팁

💡 - JSON 응답을 요청할 때 예시를 함께 제공하면 파싱 오류가 줄어듭니다

• 복잡한 작업일수록 분해 단계를 여러 번 거치는 것이 좋습니다
• 분해된 작업을 로깅해두면 디버깅과 성능 분석에 도움이 됩니다

---

4. 워커 실행과 결과 수집

작업 분배 로직까지 완성한 김개발 씨는 이제 실제로 워커들을 실행하고 결과를 모아야 합니다. 박시니어 씨가 말했습니다.

"여러 워커가 동시에 일하면 더 빠르겠죠? 병렬 처리를 고려해보세요." 김개발 씨의 눈이 반짝였습니다.

워커 실행과 결과 수집은 분배된 작업을 실제로 처리하는 단계입니다. 여러 워커가 독립적으로 작업을 수행하고, 오케스트레이터는 모든 결과를 수집하여 최종 응답을 구성합니다.

병렬 처리를 활용하면 전체 처리 시간을 크게 단축할 수 있습니다.

다음 코드를 살펴봅시다.

import asyncio
from concurrent.futures import ThreadPoolExecutor

def execute_workers(self, subtasks: List[Dict]) -> List[Dict[str, Any]]:
    """워커들에게 작업을 실행시키고 결과를 수집합니다"""
    results = []

    def run_worker(subtask_info):
        worker_name = subtask_info["worker_name"]
        worker = next((w for w in self.workers if w.name == worker_name), None)
        if worker:
            result = worker.execute(subtask_info["subtask"])
            return {"worker": worker_name, "result": result}
        return {"worker": worker_name, "result": "워커를 찾을 수 없습니다"}

    # 병렬로 워커 실행
    with ThreadPoolExecutor(max_workers=len(subtasks)) as executor:
        results = list(executor.map(run_worker, subtasks))

    return results

작업을 분배했다면 이제 실제로 워커들이 일을 할 차례입니다. 김개발 씨는 처음에 간단하게 for 루프로 워커를 하나씩 실행하려 했습니다.

박시니어 씨가 그 코드를 보더니 고개를 저었습니다. "잠깐, 그렇게 하면 세 개의 워커가 각각 5초씩 걸린다면 총 15초가 걸려요.

하지만 동시에 실행하면 5초면 끝나죠." 병렬 처리의 힘입니다. 쉽게 비유하자면, 식당에서 주문을 받는 것과 같습니다.

손님 세 명이 각각 스테이크, 파스타, 샐러드를 주문했습니다. 요리사가 한 명이라면 세 요리를 순서대로 만들어야 합니다.

하지만 요리사가 세 명이면? 동시에 세 요리를 만들어 훨씬 빠르게 서빙할 수 있습니다.

워커도 마찬가지입니다. 각 워커는 독립적으로 작업을 수행하므로 굳이 순서대로 기다릴 필요가 없습니다.

Python의 ThreadPoolExecutor를 사용하면 이를 쉽게 구현할 수 있습니다. 코드를 살펴보겠습니다.

run_worker 함수는 하나의 하위 작업을 처리합니다. 먼저 worker_name으로 적절한 워커를 찾습니다.

next()와 제너레이터 표현식을 사용한 패턴은 리스트에서 조건에 맞는 첫 번째 요소를 찾을 때 유용합니다. 워커를 찾으면 execute 메서드를 호출하고, 결과를 딕셔너리로 감싸 반환합니다.

ThreadPoolExecutor의 map 메서드가 마법을 부립니다. 이 메서드는 주어진 함수를 iterable의 각 요소에 병렬로 적용합니다.

max_workers를 작업 수만큼 설정하면 모든 워커가 동시에 실행됩니다. 실무에서 고려해야 할 점이 있습니다.

API 호출에는 rate limit이 있습니다. 너무 많은 워커를 동시에 실행하면 API 제한에 걸릴 수 있습니다.

따라서 max_workers를 적절히 조절하거나, 재시도 로직을 추가해야 합니다. 또한 에러 처리도 중요합니다.

한 워커가 실패했다고 전체가 실패하면 안 됩니다. 실패한 작업만 기록하고 나머지 결과는 정상적으로 반환하는 것이 좋습니다.

김개발 씨가 테스트를 돌려보니 처리 시간이 확연히 줄었습니다. "와, 정말 빨라졌어요!" 박시니어 씨가 덧붙였습니다.

"비동기 처리를 쓰면 더 효율적일 수 있어요. 하지만 지금은 ThreadPool로도 충분합니다.

나중에 필요하면 asyncio로 전환해도 돼요."

실전 팁

💡 - ThreadPoolExecutor는 I/O 바운드 작업에 적합합니다. CPU 바운드 작업은 ProcessPoolExecutor를 고려하세요

• API rate limit을 고려하여 max_workers를 적절히 설정하세요
• 각 워커의 실행 시간을 로깅하면 병목 지점을 파악할 수 있습니다

---

5. 오케스트레이터 종합 로직

개별 컴포넌트들이 모두 준비되었습니다. 이제 김개발 씨는 이것들을 하나로 엮어 완전한 오케스트레이터를 완성해야 합니다.

"퍼즐 조각을 맞추듯이, 분석, 분배, 실행, 종합을 하나의 흐름으로 만들어봐요." 박시니어 씨의 안내에 김개발 씨가 키보드를 두드리기 시작했습니다.

오케스트레이터 종합 로직은 지금까지 만든 모든 컴포넌트를 연결하는 메인 메서드입니다. 사용자의 요청을 받아 작업을 분석하고, 워커에게 분배하고, 결과를 수집하여 최종 응답을 생성합니다.

이것이 오케스트레이터-워커 패턴의 완성입니다.

다음 코드를 살펴봅시다.

def process(self, user_request: str) -> str:
    """사용자 요청을 처리하는 메인 메서드"""
    # 1단계: 작업 분석 및 분해
    subtasks = self.analyze_task(user_request)
    print(f"분해된 작업: {len(subtasks)}개")

    # 2단계: 워커 실행 및 결과 수집
    results = self.execute_workers(subtasks)

    # 3단계: 결과 종합하여 최종 응답 생성
    synthesis_prompt = f"""다음 결과들을 종합하여 최종 응답을 작성해주세요.

원래 요청: {user_request}

각 워커의 결과:
{json.dumps(results, ensure_ascii=False, indent=2)}

자연스럽고 통합된 응답을 작성해주세요."""

    final_response = self.client.messages.create(
        model=self.model, max_tokens=2048,
        messages=[{"role": "user", "content": synthesis_prompt}]
    )
    return final_response.content[0].text

드디어 모든 퍼즐 조각이 맞춰질 시간입니다. 김개발 씨는 지금까지 만든 analyze_task, execute_workers 메서드를 하나의 흐름으로 연결해야 합니다.

박시니어 씨가 화이트보드에 플로우차트를 그렸습니다. "전체 흐름을 보면 이래요.

요청 → 분석 → 분배 → 실행 → 수집 → 종합 → 응답. 단순하죠?" 실제로 단순합니다.

복잡한 시스템도 잘 설계하면 각 단계가 명확해집니다. process 메서드는 세 단계로 나뉩니다.

첫 번째 단계는 작업 분석입니다. 사용자의 요청을 받아 analyze_task 메서드에 전달합니다.

LLM이 요청을 분석하고 하위 작업 목록을 반환합니다. print문으로 분해된 작업 수를 출력하는데, 이는 디버깅과 모니터링에 유용합니다.

두 번째 단계는 워커 실행입니다. 분해된 작업을 execute_workers에 전달하면 각 워커가 병렬로 작업을 수행합니다.

결과는 워커 이름과 처리 결과가 담긴 딕셔너리 리스트로 반환됩니다. 세 번째 단계가 중요합니다.

결과 종합입니다. 여러 워커의 결과를 그냥 이어붙이면 어색한 응답이 됩니다.

따라서 LLM에게 결과를 종합하여 자연스러운 응답을 만들도록 요청합니다. synthesis_prompt를 보면, 원래 요청과 각 워커의 결과를 함께 제공합니다.

이렇게 하면 LLM이 맥락을 이해하고 일관성 있는 최종 응답을 생성할 수 있습니다. 쉽게 비유하자면, 마치 뉴스 앵커와 같습니다.

여러 기자가 취재해온 내용을 앵커가 종합하여 시청자에게 전달합니다. "방금 A 기자가 전한 경제 소식과 B 기자의 국제 뉴스를 종합해보면..."처럼요.

실무에서 이 종합 단계는 매우 중요합니다. 단순히 결과를 나열하면 사용자 경험이 나빠집니다.

"데이터 분석 결과: ..., 추천 결과: ..."보다는 "고객님의 지난달 구매 패턴을 분석한 결과, 가장 관심 있으신 카테고리는 전자제품입니다. 이를 바탕으로 추천드리는 상품은..."처럼 자연스러운 응답이 좋습니다.

김개발 씨가 전체 코드를 실행해보았습니다. "오, 정말 돌아가네요!" 박시니어 씨가 고개를 끄덕였습니다.

"기본 구조는 완성됐어요. 이제 UI를 붙이면 실제로 사용할 수 있는 애플리케이션이 되겠죠."

실전 팁

💡 - 각 단계 사이에 로깅을 추가하면 문제 발생 시 디버깅이 쉬워집니다

• 종합 단계의 프롬프트를 잘 설계하면 응답 품질이 크게 향상됩니다
• 전체 처리 시간을 측정하여 성능 병목을 파악하세요

---

6. 스트림릿 UI 구현

백엔드 로직이 완성되자 김개발 씨는 뿌듯함을 느꼈습니다. 하지만 박시니어 씨가 말했습니다.

"콘솔에서만 돌아가면 아무도 안 써요. 예쁜 UI를 붙여봐요." 김개발 씨는 Streamlit을 열었습니다.

Python만으로 웹 UI를 만들 수 있다니, 정말 좋은 시대입니다.

Streamlit은 Python으로 웹 애플리케이션을 빠르게 만들 수 있는 프레임워크입니다. 복잡한 HTML, CSS, JavaScript 없이도 데이터 과학자와 개발자가 인터랙티브한 UI를 구현할 수 있습니다.

오케스트레이터-워커 시스템에 Streamlit을 붙이면 사용하기 편한 에이전트 인터페이스가 완성됩니다.

다음 코드를 살펴봅시다.

import streamlit as st

st.title("오케스트레이터-워커 에이전트")

# 워커 설정 UI
with st.sidebar:
    st.header("워커 설정")
    num_workers = st.slider("워커 수", 1, 5, 3)
    specialties = ["데이터 분석", "텍스트 생성", "코드 작성", "요약", "번역"]
    selected = st.multiselect("워커 전문 분야", specialties, default=specialties[:num_workers])

# 메인 입력 영역
user_input = st.text_area("요청을 입력하세요", height=100)

if st.button("실행", type="primary"):
    with st.spinner("처리 중..."):
        # 오케스트레이터 초기화 및 실행
        orchestrator = Orchestrator()
        for i, spec in enumerate(selected):
            orchestrator.workers.append(Worker(name=f"worker_{i}", specialty=spec))
        result = orchestrator.process(user_input)
    st.success("완료!")
    st.markdown(result)

아무리 뛰어난 엔진이라도 운전대와 계기판이 없으면 차를 몰 수 없습니다. 마찬가지로 아무리 좋은 에이전트 시스템이라도 UI가 없으면 사용하기 어렵습니다.

김개발 씨는 처음에 Flask로 웹 서버를 만들어야 하나 고민했습니다. HTML 템플릿도 만들고, CSS도 작성하고, JavaScript로 API 호출도 구현하고...

생각만 해도 막막했습니다. 박시니어 씨가 구원의 손길을 내밀었습니다.

"Streamlit 써봤어요? Python 코드 몇 줄이면 웹 앱이 완성돼요." Streamlit은 데이터 과학자들 사이에서 인기 있는 프레임워크입니다.

그 이유는 간단합니다. Python만 알면 되니까요.

코드를 살펴보겠습니다. st.title로 페이지 제목을 설정합니다.

st.sidebar는 왼쪽에 접을 수 있는 사이드바를 만듭니다. 설정 옵션을 넣기 좋은 곳입니다.

st.slider는 숫자를 선택하는 슬라이더 UI를 만듭니다. 워커 수를 1에서 5 사이에서 선택할 수 있게 했습니다.

st.multiselect는 여러 항목을 선택할 수 있는 드롭다운입니다. 어떤 전문 분야의 워커를 사용할지 선택합니다.

st.text_area는 여러 줄 텍스트를 입력받는 영역입니다. 사용자가 여기에 요청을 작성합니다.

st.button이 클릭되면 if 블록 안의 코드가 실행됩니다. st.spinner는 처리 중임을 보여주는 스피너를 표시합니다.

사용자가 "아, 지금 처리 중이구나"라고 알 수 있어 좋은 UX입니다. 처리가 완료되면 st.success로 성공 메시지를 보여주고, st.markdown으로 결과를 렌더링합니다.

마크다운 형식이라 굵은 글씨, 목록 등이 예쁘게 표시됩니다. 실무에서 Streamlit의 장점은 빠른 프로토타이핑입니다.

아이디어가 떠오르면 몇 시간 안에 데모 앱을 만들 수 있습니다. 이해관계자에게 보여주고 피드백을 받기 좋습니다.

물론 프로덕션 환경에서는 더 견고한 프레임워크가 필요할 수 있지만, MVP나 내부 도구로는 Streamlit이 충분합니다. 김개발 씨가 앱을 실행하자 브라우저에 깔끔한 UI가 나타났습니다.

"이게 Python으로 만든 거라고요?" 박시니어 씨가 웃었습니다. "네, 프론트엔드 코드 한 줄도 안 썼어요.

이제 테스트해봐요."

실전 팁

💡 - st.cache_data 데코레이터를 사용하면 비용이 큰 연산 결과를 캐싱할 수 있습니다

• st.session_state를 활용하면 페이지 새로고침 사이에도 상태를 유지할 수 있습니다
• streamlit run app.py로 앱을 실행하면 자동으로 브라우저가 열립니다

---

7. 에이전트 테스트 및 개선

UI까지 완성한 김개발 씨는 자신감이 충만했습니다. 하지만 박시니어 씨가 진지한 얼굴로 말했습니다.

"개발은 절반밖에 안 끝났어요. 테스트와 개선이 남았죠." 실제로 첫 테스트에서 예상치 못한 문제들이 속출했습니다.

테스트와 개선은 개발의 필수 단계입니다. 아무리 잘 설계한 시스템도 실제로 돌려보면 예상치 못한 문제가 발생합니다.

에러 처리, 로깅, 성능 최적화를 통해 시스템을 견고하게 만들어야 합니다. 특히 LLM 기반 시스템은 비결정적 특성 때문에 더 많은 테스트가 필요합니다.

다음 코드를 살펴봅시다.

import logging
from typing import Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def process_with_retry(self, request: str, max_retries: int = 3) -> Optional[str]:
    """재시도 로직이 포함된 안전한 처리 메서드"""
    for attempt in range(max_retries):
        try:
            logger.info(f"처리 시도 {attempt + 1}/{max_retries}")
            result = self.process(request)
            logger.info("처리 성공")
            return result
        except json.JSONDecodeError as e:
            logger.warning(f"JSON 파싱 실패: {e}")
        except Exception as e:
            logger.error(f"예상치 못한 오류: {e}")

        if attempt < max_retries - 1:
            logger.info("재시도합니다...")

    logger.error("최대 재시도 횟수 초과")
    return None

"되는 것 같은데요?"라는 말은 개발자의 착각 중 하나입니다. 김개발 씨도 처음에는 그렇게 생각했습니다.

하지만 실제 테스트를 시작하자 문제가 터져나왔습니다. 첫 번째 문제는 JSON 파싱 오류였습니다.

LLM이 JSON을 생성할 때 가끔 앞뒤에 설명을 붙이거나, 따옴표를 잘못 사용했습니다. json.loads()가 실패하고 전체 프로세스가 멈췄습니다.

두 번째 문제는 타임아웃이었습니다. 네트워크 상태가 좋지 않을 때 API 호출이 무한정 기다리는 경우가 있었습니다.

세 번째 문제는 디버깅의 어려움이었습니다. 에러가 발생해도 어느 단계에서 문제가 생겼는지 알기 어려웠습니다.

박시니어 씨가 말했습니다. "이래서 로깅과 에러 처리가 중요한 거예요." 코드를 살펴보면, logging 모듈을 사용합니다.

logger.info는 일반적인 정보를, logger.warning은 경고를, logger.error는 오류를 기록합니다. 이렇게 해두면 문제 발생 시 로그를 보고 원인을 파악할 수 있습니다.

재시도 로직도 중요합니다. 네트워크 오류나 일시적인 API 문제는 다시 시도하면 해결되는 경우가 많습니다.

max_retries만큼 시도하고, 각 시도 사이에 상태를 로깅합니다. try-except 구문에서 구체적인 예외를 먼저 처리하는 것이 좋습니다.

JSONDecodeError는 자주 발생하는 예상 가능한 오류입니다. 이를 별도로 처리하면 적절한 대응이 가능합니다.

예상치 못한 오류는 마지막에 Exception으로 잡습니다. 실무에서 추가로 고려할 점이 있습니다.

성능 모니터링이 중요합니다. 각 단계별 소요 시간을 측정하면 병목 지점을 찾을 수 있습니다.

API 호출이 느린지, 결과 종합이 느린지 파악해야 개선할 수 있습니다. 테스트 케이스를 다양하게 준비하세요.

정상적인 요청뿐만 아니라 빈 입력, 매우 긴 입력, 특수문자가 포함된 입력 등 엣지 케이스도 테스트해야 합니다. 비용 관리도 잊지 마세요.

LLM API 호출에는 비용이 발생합니다. 불필요한 재시도를 줄이고, 캐싱을 활용하면 비용을 절약할 수 있습니다.

김개발 씨가 개선된 코드로 다시 테스트했습니다. JSON 파싱 오류가 발생해도 자동으로 재시도하여 성공했습니다.

로그를 보니 어떤 과정을 거쳤는지 한눈에 들어왔습니다. "이제야 진짜 완성된 것 같아요." 김개발 씨가 뿌듯하게 말했습니다.

박시니어 씨가 어깨를 두드렸습니다. "좋아요.

이제 실제 서비스에 적용할 준비가 됐네요. 오케스트레이터-워커 패턴은 다양한 곳에 활용할 수 있어요.

오늘 배운 걸 바탕으로 더 발전시켜 보세요."

실전 팁

💡 - 프로덕션 환경에서는 구조화된 로깅(JSON 형식)을 사용하면 로그 분석이 쉬워집니다

• API 호출에 timeout 파라미터를 설정하여 무한 대기를 방지하세요
• 단위 테스트와 통합 테스트를 모두 작성하면 리팩토링 시 안심할 수 있습니다

---

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