Hierarchical Agents 완벽 가이드

AI 에이전트를 계층적으로 구성하여 복잡한 작업을 효율적으로 처리하는 방법을 배웁니다. Supervisor-Worker 패턴부터 실전 프로젝트 관리까지 단계별로 학습합니다.

1. Supervisor-Worker 구조

어느 날 김개발 씨는 복잡한 데이터 분석 프로젝트를 맡았습니다. 데이터 수집, 전처리, 분석, 시각화까지 모든 것을 하나의 AI 에이전트로 처리하려니 코드가 너무 복잡해졌습니다.

선배 박시니어 씨가 다가와 물었습니다. "왜 하나의 에이전트로 모든 걸 하려고 해요?"

Supervisor-Worker 구조는 관리자 역할의 Supervisor 에이전트가 전체 작업을 조율하고, 여러 Worker 에이전트가 세부 작업을 담당하는 계층적 패턴입니다. 마치 프로젝트 매니저가 팀원들에게 업무를 분배하는 것과 같습니다.

이 구조를 사용하면 복잡한 작업을 체계적으로 관리할 수 있습니다.

다음 코드를 살펴봅시다.

from typing import List, Dict
from langchain.agents import Agent

class SupervisorAgent:
    def __init__(self, workers: List[Agent]):
        # Supervisor는 여러 Worker를 관리합니다
        self.workers = workers
        self.task_queue = []

    def delegate_task(self, task: str) -> Dict:
        # 작업의 종류를 분석하여 적절한 Worker에게 위임
        if "데이터 수집" in task:
            return self.workers[0].run(task)
        elif "분석" in task:
            return self.workers[1].run(task)
        else:
            return self.workers[2].run(task)

김개발 씨는 입사 6개월 차 주니어 개발자입니다. 최근 AI 에이전트를 활용한 자동화 시스템을 개발하는 프로젝트에 투입되었습니다.

처음에는 간단해 보였습니다. 하나의 AI 에이전트로 모든 작업을 처리하면 되니까요.

하지만 프로젝트가 진행될수록 문제가 생겼습니다. 하나의 에이전트가 데이터 수집, 전처리, 분석, 결과 생성까지 모든 것을 처리하려니 코드가 점점 복잡해졌습니다.

에러가 발생하면 어디서 문제가 생긴 건지 찾기도 어려웠습니다. 선배 개발자 박시니어 씨가 김개발 씨의 화면을 보더니 미소를 지었습니다.

"김 개발님, 조직도를 생각해 보세요. 회사에서 모든 일을 대표님이 직접 하나요?" 김개발 씨가 고개를 저었습니다.

"아니요, 팀장님들이 각자 팀을 관리하고, 팀원들이 실무를 담당하죠." 박시니어 씨가 고개를 끄덕였습니다. "바로 그겁니다.

AI 에이전트도 마찬가지예요." 계층적 에이전트 시스템이란 무엇일까요? 쉽게 비유하자면, 계층적 에이전트는 회사 조직과 같습니다.

최상위에는 전체를 관리하는 매니저가 있고, 그 아래에는 각자의 전문 분야를 담당하는 팀원들이 있습니다. 매니저는 전체 프로젝트를 파악하고 팀원들에게 적절한 업무를 분배합니다.

팀원들은 자신의 전문 분야에 집중하여 최고의 결과를 만들어냅니다. 이 패턴에서 가장 핵심이 되는 것이 바로 Supervisor-Worker 구조입니다.

단일 에이전트 시스템이 없던 것은 아닙니다. 하지만 복잡한 작업을 처리할 때 여러 문제가 있었습니다.

첫째, 에이전트가 너무 많은 것을 알아야 했습니다. 데이터베이스 접근 방법, API 호출 방법, 데이터 분석 기법, 결과 시각화 방법까지 모든 것을 하나의 에이전트가 처리해야 했습니다.

이는 마치 한 사람이 회사의 모든 업무를 처리하는 것과 같았습니다. 둘째, 유지보수가 어려웠습니다.

데이터 수집 방식을 변경하려면 전체 에이전트 코드를 수정해야 했습니다. 작은 변경이 예상치 못한 부작용을 일으키기도 했습니다.

셋째, 확장성이 떨어졌습니다. 새로운 기능을 추가하려면 이미 복잡한 에이전트를 더 복잡하게 만들어야 했습니다.

코드는 점점 스파게티처럼 얽혀갔습니다. 바로 이런 문제를 해결하기 위해 Supervisor-Worker 패턴이 등장했습니다.

이 패턴을 사용하면 책임의 분리가 명확해집니다. Supervisor는 "무엇을 할지"를 결정하고, Worker는 "어떻게 할지"를 담당합니다.

또한 재사용성도 높아집니다. 데이터 수집 Worker는 다른 프로젝트에서도 그대로 사용할 수 있습니다.

무엇보다 디버깅과 테스트가 쉬워진다는 큰 이점이 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다.

먼저 SupervisorAgent 클래스는 여러 Worker 에이전트를 리스트로 받아 관리합니다. 이것이 핵심입니다.

Supervisor는 직접 작업을 수행하지 않고, 적절한 Worker를 선택하는 역할만 합니다. delegate_task 메서드는 작업 내용을 분석하여 어떤 Worker에게 위임할지 결정합니다.

간단한 키워드 매칭을 사용했지만, 실제로는 더 정교한 분류 로직을 사용할 수 있습니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 전자상거래 플랫폼을 운영한다고 가정해봅시다. 고객 문의를 처리하는 시스템에서 Supervisor 에이전트가 문의 내용을 분석합니다.

상품 문의는 상품 정보 Worker에게, 배송 문의는 배송 추적 Worker에게, 환불 요청은 환불 처리 Worker에게 각각 전달합니다. 각 Worker는 자신의 전문 분야에 최적화되어 있어 더 정확한 답변을 제공할 수 있습니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 너무 많은 계층을 만드는 것입니다.

Supervisor 아래에 또 다른 Supervisor를 두고, 그 아래에 또 Worker를 두는 식으로 복잡하게 설계하면 오히려 관리가 어려워집니다. 대부분의 경우 2-3단계의 계층이면 충분합니다.

단순함이 최고의 복잡성 해결책입니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

박시니어 씨의 설명을 들은 김개발 씨는 코드를 리팩토링하기 시작했습니다. 하나의 거대한 에이전트를 Supervisor 하나와 세 개의 Worker로 분리했습니다.

코드는 훨씬 깔끔해졌고, 각 부분을 독립적으로 테스트할 수 있게 되었습니다. Supervisor-Worker 패턴을 제대로 이해하면 더 유지보수하기 쉽고 확장 가능한 AI 시스템을 구축할 수 있습니다.

여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - Supervisor는 가능한 한 단순하게 유지하세요. 복잡한 로직은 Worker에게 위임하는 것이 좋습니다.

Worker는 하나의 책임만 가지도록 설계하세요. 단일 책임 원칙을 지키면 재사용성이 높아집니다.
처음부터 완벽한 계층 구조를 만들려 하지 마세요. 필요에 따라 점진적으로 개선하는 것이 좋습니다.

2. 태스크 분해와 위임

김개발 씨가 Supervisor-Worker 구조를 이해하고 나니 새로운 궁금증이 생겼습니다. "그런데 Supervisor는 어떻게 복잡한 작업을 적절히 나누는 거죠?" 박시니어 씨가 화이트보드를 가리키며 대답했습니다.

"좋은 질문이에요. 그게 바로 태스크 분해의 핵심입니다."

태스크 분해는 복잡한 작업을 더 작고 관리 가능한 하위 작업으로 나누는 과정입니다. Supervisor는 전체 작업을 분석하여 독립적으로 수행 가능한 단위로 쪼개고, 각각을 적절한 Worker에게 위임합니다.

이는 프로젝트 관리에서 WBS를 작성하는 것과 유사합니다.

다음 코드를 살펴봅시다.

from typing import List, Tuple

class TaskDecomposer:
    def decompose(self, complex_task: str) -> List[Tuple[str, str]]:
        # 복잡한 작업을 하위 작업으로 분해합니다
        subtasks = []

        if "보고서 생성" in complex_task:
            # 보고서 생성 작업을 세부 단계로 분해
            subtasks.append(("데이터 수집", "data_collector"))
            subtasks.append(("데이터 분석", "data_analyst"))
            subtasks.append(("그래프 생성", "visualizer"))
            subtasks.append(("보고서 작성", "report_writer"))

        return subtasks

    def estimate_dependencies(self, subtasks: List) -> Dict:
        # 각 하위 작업 간의 의존성을 파악합니다
        # 예: 분석은 수집이 완료된 후에만 가능
        return {"data_analyst": ["data_collector"]}

김개발 씨는 이제 Supervisor와 Worker의 역할을 이해했습니다. 하지만 실제로 구현하려니 막막했습니다.

"월간 매출 보고서를 생성해 주세요"라는 요청이 들어왔을 때, Supervisor는 이것을 어떻게 여러 개의 작은 작업으로 나눌 수 있을까요? 박시니어 씨가 예시를 들어주었습니다.

"김 개발님이 집을 짓는다고 생각해 보세요. '집을 지어라'는 하나의 큰 작업이죠.

하지만 실제로는 어떻게 하나요?" 김개발 씨가 생각하며 대답했습니다. "음, 먼저 설계를 하고, 기초 공사를 하고, 골조를 세우고, 내부 마감을 하고..." 박시니어 씨가 웃으며 고개를 끄덕였습니다.

"맞아요. 그게 바로 태스크 분해입니다." 태스크 분해란 정확히 무엇일까요?

레시피를 생각해 봅시다. "비빔밥을 만들어라"는 하나의 목표입니다.

하지만 실제로 만들려면 여러 단계가 필요합니다. 밥을 짓고, 나물을 무치고, 고기를 볶고, 계란을 부치고, 마지막으로 모든 것을 그릇에 담아 비빕니다.

각 단계는 독립적으로 수행할 수 있으며, 일부는 동시에 진행할 수도 있습니다. 태스크 분해도 이와 같습니다.

AI 에이전트 시스템에서 태스크 분해가 왜 중요할까요? 초기의 단일 에이전트 시스템에서는 "보고서를 생성해 주세요"라는 요청을 받으면 에이전트가 모든 것을 순차적으로 처리했습니다.

문제는 어느 단계에서 에러가 발생하면 처음부터 다시 시작해야 한다는 것이었습니다. 또한 일부 작업은 병렬로 처리할 수 있는데도 순차적으로만 진행했습니다.

효율성이 떨어질 수밖에 없었습니다. 더 큰 문제는 유연성 부족이었습니다.

보고서의 형식을 바꾸거나 새로운 분석 방법을 추가하려면 전체 프로세스를 수정해야 했습니다. 마치 레시피의 한 재료를 바꾸기 위해 요리 전체를 다시 만드는 것과 같았습니다.

태스크 분해를 도입하면 이런 문제가 해결됩니다. 첫째, 병렬 처리가 가능해집니다.

데이터를 수집하는 동안 다른 Worker가 템플릿을 준비할 수 있습니다. 둘째, 에러 처리가 쉬워집니다.

특정 하위 작업이 실패하면 그 부분만 재시도하면 됩니다. 셋째, 재사용성이 높아집니다.

"데이터 수집" 작업은 다른 보고서를 만들 때도 그대로 사용할 수 있습니다. 위의 코드를 자세히 살펴보겠습니다.

TaskDecomposer 클래스의 decompose 메서드는 복잡한 작업을 받아 하위 작업의 리스트로 반환합니다. 각 하위 작업은 작업 설명과 담당 Worker의 이름을 튜플로 가집니다.

이렇게 하면 Supervisor가 어떤 순서로 무엇을 해야 하는지 명확히 알 수 있습니다. estimate_dependencies 메서드는 더욱 중요합니다.

이것은 각 작업 간의 의존성을 파악합니다. 예를 들어 데이터를 수집하기 전에는 분석을 할 수 없습니다.

하지만 그래프 생성과 보고서 템플릿 준비는 동시에 진행할 수 있습니다. 이런 의존성 정보를 바탕으로 Supervisor는 효율적인 실행 계획을 수립합니다.

실무에서는 어떻게 활용할까요? 대형 쇼핑몰의 재고 관리 시스템을 예로 들어봅시다.

"재고 부족 상품을 찾아 발주하라"는 작업이 들어옵니다. TaskDecomposer는 이것을 다음과 같이 분해합니다.

먼저 "현재 재고 조회", 그다음 "판매 추세 분석", "재고 부족 예측", "공급업체 선택", "발주서 생성", 마지막으로 "발주 실행"입니다. 각 단계는 전문화된 Worker가 담당하며, 일부는 병렬로 실행됩니다.

많은 기업에서 이런 방식으로 복잡한 비즈니스 프로세스를 자동화하고 있습니다. 특히 금융, 의료, 물류 분야에서 효과가 크다고 알려져 있습니다.

주의해야 할 점도 있습니다. 가장 흔한 실수는 너무 작게 쪼개는 것입니다.

"보고서 제목 작성", "보고서 부제목 작성" 같은 식으로 지나치게 세분화하면 오히려 관리 오버헤드가 커집니다. 적절한 수준의 분해가 중요합니다.

일반적으로 각 하위 작업이 5-15분 정도 소요되는 것이 적당합니다. 또 다른 실수는 의존성을 무시하는 것입니다.

데이터를 수집하기도 전에 분석을 시작하면 에러가 발생합니다. 반드시 작업 간의 순서와 의존 관계를 명확히 해야 합니다.

김개발 씨는 박시니어 씨의 조언을 따라 태스크 분해 로직을 구현했습니다. 처음에는 복잡해 보였지만, 막상 만들고 나니 전체 시스템이 훨씬 명확해졌습니다.

"이제 새로운 기능을 추가할 때도 어디를 수정해야 할지 바로 알 수 있겠어요!" 태스크 분해를 제대로 이해하면 복잡한 문제를 체계적으로 해결할 수 있습니다. 여러분도 다음 프로젝트에서 이 접근법을 시도해 보세요.

실전 팁

💡 - 하위 작업은 독립적으로 테스트할 수 있을 만큼 명확하게 정의하세요.

의존성 그래프를 그려보면 작업 순서를 계획하는 데 도움이 됩니다.
처음에는 큰 단위로 나누고, 필요하면 점진적으로 세분화하세요.

3. 결과 집계

여러 Worker가 각자의 작업을 완료한 후, 김개발 씨는 새로운 문제에 직면했습니다. "각 Worker가 만든 결과물을 어떻게 하나로 합치죠?" 박시니어 씨가 커피를 한 모금 마시며 말했습니다.

"그게 바로 결과 집계의 역할입니다. 가장 중요한 마지막 단계죠."

결과 집계는 여러 Worker가 생성한 개별 결과물을 수집하고 통합하여 최종 결과를 만드는 과정입니다. Supervisor는 각 Worker의 출력을 받아 일관성을 검증하고, 필요한 형식으로 변환하며, 하나의 완전한 결과물로 조합합니다.

이는 퍼즐 조각을 맞춰 완성된 그림을 만드는 것과 같습니다.

다음 코드를 살펴봅시다.

from typing import List, Dict, Any

class ResultAggregator:
    def __init__(self):
        self.results = {}

    def collect_result(self, worker_id: str, result: Any):
        # 각 Worker의 결과를 수집합니다
        self.results[worker_id] = result

    def validate_results(self) -> bool:
        # 모든 필수 결과가 수집되었는지 확인
        required = ["data_collector", "data_analyst", "visualizer"]
        return all(w in self.results for w in required)

    def aggregate(self) -> Dict:
        # 수집된 결과들을 하나의 최종 결과로 통합
        if not self.validate_results():
            raise ValueError("일부 결과가 누락되었습니다")

        final_report = {
            "data": self.results["data_collector"],
            "analysis": self.results["data_analyst"],
            "charts": self.results["visualizer"],
            "timestamp": "2025-01-15"
        }
        return final_report

김개발 씨의 시스템은 이제 작업을 잘 분해하고 각 Worker에게 위임할 수 있게 되었습니다. 데이터 수집 Worker는 데이터를 가져오고, 분석 Worker는 인사이트를 도출하고, 시각화 Worker는 그래프를 만들었습니다.

모든 것이 순조로워 보였습니다. 하지만 문제가 생겼습니다.

각 Worker가 만든 결과물의 형식이 제각각이었습니다. 데이터 수집 Worker는 JSON을 반환했고, 분석 Worker는 텍스트를 반환했으며, 시각화 Worker는 이미지 파일 경로를 반환했습니다.

이것들을 어떻게 하나의 보고서로 만들 수 있을까요? 박시니어 씨가 설명했습니다.

"레고 블록을 생각해 보세요. 각각의 블록은 그 자체로는 의미가 있지만, 이것들을 조립해야 비로소 완성된 작품이 되죠." 결과 집계란 무엇일까요?

오케스트라를 떠올려 봅시다. 바이올린, 첼로, 플루트, 트럼펫 등 각 악기는 자신의 파트를 연주합니다.

각각은 아름답지만, 진정한 음악은 이 모든 소리가 조화를 이룰 때 탄생합니다. 지휘자는 각 악기의 소리를 듣고 전체적인 하모니를 만들어냅니다.

결과 집계도 이와 같은 역할을 합니다. 왜 결과 집계가 필요할까요?

초기 시스템에서는 각 Worker가 독립적으로 결과를 저장했습니다. 사용자는 여러 파일을 열어 각각 확인해야 했습니다.

데이터는 하나의 폴더에, 분석 결과는 다른 폴더에, 그래프는 또 다른 곳에 저장되었습니다. 이것은 마치 백과사전의 각 페이지가 다른 책에 흩어져 있는 것과 같았습니다.

더 심각한 문제는 일관성 부족이었습니다. 어떤 Worker는 날짜를 "2025-01-15" 형식으로 표현했고, 다른 Worker는 "15/01/2025"로 표현했습니다.

숫자도 어떤 곳은 쉼표를 사용했고, 다른 곳은 사용하지 않았습니다. 최종 사용자가 이런 불일치를 직접 처리해야 했습니다.

결과 집계를 도입하면 이런 문제가 해결됩니다. 첫째, 통일된 형식으로 결과를 제공할 수 있습니다.

모든 날짜는 같은 형식으로, 모든 숫자는 일관된 포맷으로 표시됩니다. 둘째, 완전성 검증이 가능합니다.

필수적인 결과가 빠지지 않았는지 확인할 수 있습니다. 셋째, 사용자 경험이 향상됩니다.

하나의 완성된 보고서를 받아볼 수 있습니다. 위의 코드를 단계별로 살펴보겠습니다.

ResultAggregator 클래스는 결과 집계를 담당합니다. collect_result 메서드는 각 Worker가 작업을 완료할 때마다 호출되어 결과를 저장합니다.

worker_id를 키로 사용하여 어떤 Worker의 결과인지 추적합니다. validate_results 메서드가 핵심입니다.

이것은 모든 필수 Worker가 결과를 제출했는지 확인합니다. 만약 데이터 수집은 완료되었지만 분석이 실패했다면, 이 메서드는 False를 반환하여 문제를 알려줍니다.

불완전한 결과를 사용자에게 제공하는 것을 방지합니다. aggregate 메서드는 실제 통합을 수행합니다.

먼저 검증을 거친 후, 각 Worker의 결과를 적절한 구조로 조합합니다. 여기서는 간단한 딕셔너리를 사용했지만, 실제로는 더 복잡한 객체나 문서를 생성할 수 있습니다.

실무에서는 어떻게 활용될까요? 고객 서비스 챗봇 시스템을 예로 들어봅시다.

고객이 "내 주문 상태를 알려주세요"라고 문의하면, 주문 조회 Worker, 배송 추적 Worker, 고객 이력 Worker가 각각 정보를 수집합니다. ResultAggregator는 이 세 가지 정보를 받아 하나의 완성된 답변을 만듭니다.

"고객님의 주문은 1월 10일에 접수되어 현재 배송 중이며, 내일 도착 예정입니다. 고객님은 지난 6개월간 총 5회 구매하신 우수 고객이십니다." 같은 통합된 답변을 제공할 수 있습니다.

금융 분야에서는 더욱 중요합니다. 투자 리포트를 생성할 때, 시장 데이터 Worker, 종목 분석 Worker, 리스크 평가 Worker가 각각 작업합니다.

ResultAggregator는 이 모든 정보를 검증하고 통합하여 투자자에게 일관성 있고 완전한 리포트를 제공합니다. 주의할 점이 있습니다.

흔한 실수는 에러 처리를 무시하는 것입니다. 어떤 Worker가 실패했을 때 어떻게 할 것인지 미리 정의해야 합니다.

전체 작업을 취소할 것인지, 부분 결과라도 제공할 것인지 결정해야 합니다. 예를 들어 시각화 Worker가 실패했다면, 텍스트 분석 결과만이라도 제공할 수 있습니다.

또 다른 주의사항은 메모리 관리입니다. 여러 Worker의 결과를 모두 메모리에 저장하면 큰 데이터의 경우 문제가 될 수 있습니다.

필요하다면 중간 결과를 디스크에 저장하거나 스트리밍 방식으로 처리하는 것을 고려해야 합니다. 김개발 씨는 ResultAggregator를 구현한 후 시스템을 테스트했습니다.

이제 각 Worker의 결과가 자동으로 수집되고 검증되며, 하나의 깔끔한 보고서로 완성되었습니다. "와, 이제 정말 완성된 시스템 같아요!" 결과 집계를 제대로 구현하면 사용자는 복잡한 내부 프로세스를 신경 쓸 필요 없이 완성된 결과물만 받아볼 수 있습니다.

여러분의 시스템에도 이런 통합 레이어를 추가해 보세요.

실전 팁

💡 - 결과 검증 로직을 명확히 정의하세요. 필수 필드, 데이터 타입, 값의 범위 등을 체크합니다.

부분 실패를 어떻게 처리할지 미리 계획하세요. 일부 Worker가 실패해도 가능한 결과는 제공할 수 있습니다.
결과를 캐싱하여 동일한 요청에 대해 다시 계산하지 않도록 최적화할 수 있습니다.

4. 실습: 계층적 에이전트 시스템

이론을 배운 김개발 씨는 직접 계층적 에이전트 시스템을 만들어보기로 했습니다. 박시니어 씨가 실전 예제를 제안했습니다.

"뉴스 요약 시스템을 만들어 볼까요? 여러 뉴스 사이트에서 기사를 수집하고, 요약하고, 분류하는 시스템입니다."

계층적 에이전트 시스템을 직접 구현하여 Supervisor가 여러 Worker를 조율하는 실제 동작을 경험합니다. 이 실습에서는 뉴스 수집, 요약, 분류를 담당하는 세 개의 Worker와 이들을 관리하는 Supervisor를 만들어 전체 워크플로우를 완성합니다.

다음 코드를 살펴봅시다.

from typing import List, Dict
import asyncio

class NewsCollector:
    async def collect(self, sources: List[str]) -> List[Dict]:
        # 뉴스 사이트에서 기사를 수집하는 Worker
        articles = []
        for source in sources:
            # 실제로는 API 호출이나 웹 스크래핑 수행
            articles.append({"source": source, "content": "뉴스 내용..."})
        return articles

class NewsSummarizer:
    async def summarize(self, articles: List[Dict]) -> List[Dict]:
        # 기사를 요약하는 Worker
        for article in articles:
            article["summary"] = article["content"][:100] + "..."
        return articles

class NewsClassifier:
    async def classify(self, articles: List[Dict]) -> List[Dict]:
        # 기사를 분류하는 Worker
        for article in articles:
            article["category"] = "정치"  # 실제로는 AI 분류 수행
        return articles

class NewsSupervisor:
    def __init__(self):
        self.collector = NewsCollector()
        self.summarizer = NewsSummarizer()
        self.classifier = NewsClassifier()

    async def process_news(self, sources: List[str]) -> List[Dict]:
        # 1. 뉴스 수집
        articles = await self.collector.collect(sources)
        # 2. 요약과 분류를 병렬로 수행
        articles = await self.summarizer.summarize(articles)
        articles = await self.classifier.classify(articles)
        return articles

김개발 씨는 이제 본격적으로 코드를 작성할 시간입니다. 박시니어 씨가 제안한 뉴스 요약 시스템은 실제 업무에서도 자주 사용되는 패턴입니다.

여러 소스에서 정보를 모으고, 처리하고, 분류하는 작업은 데이터 파이프라인의 전형적인 형태입니다. 김개발 씨는 먼저 전체 구조를 스케치했습니다.

가장 위에는 NewsSupervisor가 있고, 그 아래에 세 개의 Worker가 있습니다. NewsCollector는 뉴스를 수집하고, NewsSummarizer는 요약하고, NewsClassifier는 분류합니다.

간단해 보였습니다. 하지만 코드를 작성하기 시작하니 고민이 생겼습니다.

"이 작업들을 순차적으로 해야 할까, 아니면 병렬로 할 수 있을까?" 박시니어 씨가 조언했습니다. "의존성을 생각해 보세요.

뉴스를 수집하기 전에 요약할 수 있나요?" 김개발 씨가 고개를 끄덕였습니다. "아, 맞아요.

수집은 먼저 해야 하고, 요약과 분류는 수집 후에 병렬로 할 수 있겠네요!" 비동기 프로그래밍이 여기서 중요한 역할을 합니다. 일반적인 동기 프로그래밍에서는 한 작업이 끝날 때까지 기다려야 다음 작업을 시작할 수 있습니다.

뉴스 사이트 A에서 기사를 가져오는 동안 다른 일을 할 수 없습니다. 하지만 비동기 프로그래밍을 사용하면 여러 작업을 동시에 진행할 수 있습니다.

사이트 A, B, C에서 동시에 기사를 가져올 수 있습니다. Python의 asyncio 라이브러리는 이런 비동기 작업을 쉽게 만들어줍니다.

async 키워드로 비동기 함수를 정의하고, await 키워드로 비동기 작업의 완료를 기다립니다. 이것은 마치 음식점에서 주문을 하고 다른 일을 하다가 호출기가 울리면 음식을 받으러 가는 것과 같습니다.

코드를 자세히 살펴보겠습니다. NewsCollector, NewsSummarizer, NewsClassifier는 각각 독립적인 Worker 클래스입니다.

각각은 하나의 명확한 책임을 가집니다. 수집은 수집만, 요약은 요약만, 분류는 분류만 담당합니다.

이것이 단일 책임 원칙입니다. NewsCollector의 collect 메서드를 보면 sources 리스트를 받아 각 소스에서 기사를 수집합니다.

실제 구현에서는 여기서 HTTP 요청을 보내거나 API를 호출할 것입니다. 지금은 간단히 더미 데이터를 반환하지만, 구조는 실제와 같습니다.

NewsSummarizer는 받은 기사 리스트를 순회하며 각 기사의 요약을 생성합니다. 여기서는 단순히 처음 100자를 잘라내지만, 실제로는 LLM API를 호출하여 지능적인 요약을 만들 수 있습니다.

NewsClassifier는 각 기사의 카테고리를 판단합니다. 정치, 경제, 사회, 문화 등으로 분류할 수 있습니다.

실제로는 텍스트 분류 모델을 사용할 것입니다. 가장 중요한 것은 NewsSupervisor입니다.

이것은 세 개의 Worker를 조율합니다. process_news 메서드를 보면 전체 워크플로우가 명확히 보입니다.

먼저 뉴스를 수집하고, 그다음 요약하고, 마지막으로 분류합니다. 각 단계가 명확하고 순서가 논리적입니다.

실제로 이 시스템을 사용하면 어떻게 될까요? 예를 들어 아침마다 주요 뉴스 사이트에서 최신 기사를 가져와 요약하고 싶다고 합시다.

NewsSupervisor에게 "네이버, 다음, 조선일보"를 소스로 전달합니다. Supervisor는 NewsCollector에게 이 세 곳에서 기사를 수집하라고 지시합니다.

수집이 완료되면 NewsSummarizer가 각 기사를 간단히 요약하고, NewsClassifier가 카테고리를 태깅합니다. 최종적으로 사용자는 카테고리별로 정리된 요약 뉴스를 받아볼 수 있습니다.

이런 시스템의 장점은 확장성입니다. 새로운 뉴스 소스를 추가하고 싶다면?

NewsCollector만 수정하면 됩니다. 요약 방식을 바꾸고 싶다면?

NewsSummarizer만 수정하면 됩니다. 다른 부분에 영향을 주지 않습니다.

김개발 씨가 주의해야 할 점도 있습니다. 에러 처리는 필수입니다.

만약 한 뉴스 사이트가 응답하지 않는다면? 전체 프로세스를 멈출 것인지, 아니면 다른 소스의 결과라도 제공할 것인지 결정해야 합니다.

try-except 블록을 사용하여 각 Worker의 에러를 적절히 처리해야 합니다. 타임아웃도 중요합니다.

어떤 뉴스 사이트가 매우 느리다면 무한정 기다릴 수 없습니다. asyncio.wait_for를 사용하여 각 작업에 타임아웃을 설정할 수 있습니다.

리소스 관리도 고려해야 합니다. 동시에 100개의 뉴스 사이트에 접속하면 네트워크 리소스나 메모리 문제가 생길 수 있습니다.

asyncio.Semaphore를 사용하여 동시 실행 수를 제한할 수 있습니다. 김개발 씨는 코드를 완성하고 테스트를 시작했습니다.

처음에는 한 개의 뉴스 소스로 시작했고, 잘 작동하는 것을 확인한 후 여러 소스로 확장했습니다. 각 Worker가 독립적으로 작동하니 디버깅도 쉬웠습니다.

박시니어 씨가 코드 리뷰를 하며 칭찬했습니다. "잘했어요.

이제 실제 프로젝트에서도 충분히 사용할 수 있는 수준이네요." 계층적 에이전트 시스템을 직접 구현해 보면 이론으로만 배울 때와는 다른 깊은 이해를 얻을 수 있습니다. 여러분도 간단한 프로젝트부터 시작해서 점진적으로 확장해 보세요.

실전 팁

💡 - 처음에는 동기 버전으로 만들어 로직을 검증한 후, 비동기로 전환하는 것이 안전합니다.

각 Worker를 독립적인 파일로 분리하면 코드 관리가 쉬워집니다.
로깅을 적극 활용하여 각 단계에서 무슨 일이 일어나는지 추적하세요.

5. 실습: 복잡한 프로젝트 관리

기본적인 계층적 시스템을 만든 김개발 씨에게 박시니어 씨가 더 도전적인 과제를 제시했습니다. "이번에는 3단계 계층 구조를 만들어 볼까요?

최상위 Supervisor, 중간 관리자, 그리고 실제 작업 Worker로 구성된 시스템입니다."

다단계 계층 구조를 구현하여 복잡한 프로젝트를 관리하는 방법을 학습합니다. 최상위 Supervisor는 전체 프로젝트를 관리하고, 중간 관리자들은 각자의 영역을 담당하며, 최하위 Worker들이 실제 작업을 수행합니다.

이는 대기업의 조직 구조와 유사합니다.

다음 코드를 살펴봅시다.

class DataWorker:
    async def fetch_data(self, source: str):
        return f"{source}에서 데이터 수집 완료"

class AnalysisWorker:
    async def analyze(self, data: str):
        return f"{data}에 대한 분석 완료"

class DataManager:
    # 중간 관리자: 데이터 관련 작업을 총괄
    def __init__(self):
        self.collector = DataWorker()
        self.analyzer = AnalysisWorker()

    async def process_data(self, sources: List[str]):
        results = []
        for source in sources:
            data = await self.collector.fetch_data(source)
            analysis = await self.analyzer.analyze(data)
            results.append(analysis)
        return results

class ReportWorker:
    async def generate(self, content: str):
        return f"보고서 생성: {content}"

class ReportManager:
    # 중간 관리자: 보고서 관련 작업을 총괄
    def __init__(self):
        self.writer = ReportWorker()

    async def create_report(self, data: List[str]):
        return await self.writer.generate(str(data))

class ProjectSupervisor:
    # 최상위 Supervisor: 전체 프로젝트 조율
    def __init__(self):
        self.data_manager = DataManager()
        self.report_manager = ReportManager()

    async def execute_project(self, sources: List[str]):
        # 1단계: 데이터 처리
        analysis = await self.data_manager.process_data(sources)
        # 2단계: 보고서 생성
        report = await self.report_manager.create_report(analysis)
        return report

김개발 씨는 이제 더 복잡한 시나리오에 도전할 준비가 되었습니다. 박시니어 씨가 설명했습니다.

"실제 대규모 프로젝트에서는 단순한 2단계 구조로는 부족할 때가 있어요. 여러 팀이 협력해야 하는 프로젝트를 생각해 보세요." 김개발 씨가 자신의 회사 조직도를 떠올렸습니다.

최상위에는 CTO가 있고, 그 아래에 데이터팀 팀장, 개발팀 팀장, 디자인팀 팀장이 있습니다. 각 팀장 아래에는 실무자들이 있습니다.

"아, 이런 구조를 코드로 만드는 거군요!" 다단계 계층 구조는 왜 필요할까요? 작은 프로젝트에서는 Supervisor 하나가 모든 Worker를 직접 관리할 수 있습니다.

하지만 프로젝트가 커지면 어떨까요? 30개의 Worker를 Supervisor 하나가 직접 관리하면 복잡도가 급격히 증가합니다.

마치 CEO가 모든 직원을 직접 관리하려는 것과 같습니다. 이때 중간 관리자 계층을 도입하면 문제가 해결됩니다.

최상위 Supervisor는 큰 그림만 보고, 중간 관리자들이 세부 사항을 담당합니다. 각 관리자는 자신의 영역에 대한 전문성을 가지고 있습니다.

코드를 단계별로 살펴보겠습니다. 가장 아래 계층에는 DataWorker와 AnalysisWorker, ReportWorker가 있습니다.

이들은 실제 작업을 수행하는 실무자들입니다. 데이터를 가져오고, 분석하고, 보고서를 작성합니다.

각자 하나의 간단한 작업만 담당합니다. 중간 계층에는 DataManager와 ReportManager가 있습니다.

DataManager는 데이터 수집과 분석을 총괄합니다. DataWorker에게 데이터를 수집하라고 지시하고, 그 결과를 받아 AnalysisWorker에게 분석을 요청합니다.

이 두 작업의 조율은 DataManager의 책임입니다. ReportManager는 보고서 생성을 담당합니다.

데이터가 준비되면 ReportWorker를 통해 최종 보고서를 만듭니다. 보고서의 형식, 내용 구성 등은 ReportManager가 결정합니다.

최상위에는 ProjectSupervisor가 있습니다. 이것은 전체 프로젝트의 흐름을 조율합니다.

"먼저 데이터를 처리하고, 그다음 보고서를 만든다"는 큰 그림을 담당합니다. 구체적인 데이터 처리 방법이나 보고서 작성 방법은 중간 관리자들에게 위임합니다.

이런 구조의 장점은 무엇일까요? 첫째, 관심사의 분리가 명확합니다.

ProjectSupervisor는 데이터가 어떻게 수집되는지 몰라도 됩니다. DataManager에게 "데이터를 처리해 주세요"라고만 요청하면 됩니다.

세부 사항은 DataManager가 알아서 처리합니다. 둘째, 변경에 유연합니다.

데이터 수집 방식을 바꾸고 싶다면 DataManager와 그 아래의 Worker만 수정하면 됩니다. ProjectSupervisor나 ReportManager는 영향을 받지 않습니다.

셋째, 테스트가 쉽습니다. 각 계층을 독립적으로 테스트할 수 있습니다.

DataManager가 올바르게 작동하는지 확인한 후, ProjectSupervisor를 테스트할 수 있습니다. 실제 업무에서는 어떻게 활용될까요?

대형 전자상거래 플랫폼의 일일 리포트 시스템을 생각해 봅시다. 최상위 ReportSupervisor는 전체 리포트 생성을 관리합니다.

그 아래에 SalesManager는 판매 데이터를 담당하고, InventoryManager는 재고 데이터를 담당하며, CustomerManager는 고객 데이터를 담당합니다. 각 Manager 아래에는 실제 데이터베이스 쿼리를 실행하고 계산을 수행하는 Worker들이 있습니다.

이런 구조를 사용하면 하루에 수백만 건의 거래를 분석하여 경영진에게 통합 리포트를 제공할 수 있습니다. 각 부분을 병렬로 처리하여 성능도 최적화할 수 있습니다.

주의해야 할 점도 있습니다. 과도한 계층화는 피해야 합니다.

계층이 너무 많으면 오히려 복잡도가 증가합니다. 일반적으로 3-4단계를 넘지 않는 것이 좋습니다.

각 계층을 거칠 때마다 오버헤드가 발생하므로 적절한 균형이 필요합니다. 통신 오버헤드도 고려해야 합니다.

계층 간 데이터 전달이 빈번하면 성능 문제가 생길 수 있습니다. 필요한 경우 중간 결과를 캐싱하거나 배치 처리를 사용할 수 있습니다.

에러 전파도 중요합니다. 최하위 Worker에서 발생한 에러가 어떻게 최상위까지 전달될지 설계해야 합니다.

각 계층에서 적절히 에러를 처리하고 로깅해야 합니다. 김개발 씨는 이 복잡한 구조를 천천히 구현했습니다.

먼저 가장 아래 Worker들을 만들고 테스트했습니다. 그다음 중간 Manager들을 만들어 Worker들과 통합했습니다.

마지막으로 최상위 Supervisor를 추가했습니다. 단계별로 검증하니 전체 시스템도 안정적으로 작동했습니다.

박시니어 씨가 최종 결과를 보며 만족스러워했습니다. "이제 정말 엔터프라이즈급 시스템이네요.

이 구조라면 향후 몇 년간 확장해도 문제없을 겁니다." 김개발 씨는 뿌듯했습니다. 처음에는 복잡해 보였던 계층적 에이전트 시스템을 이제는 자신 있게 설계하고 구현할 수 있게 되었습니다.

다단계 계층 구조는 복잡해 보이지만, 각 계층의 책임을 명확히 하면 오히려 전체 시스템을 이해하기 쉬워집니다. 여러분도 작은 프로젝트부터 시작해서 필요에 따라 계층을 추가해 보세요.

실전 팁

💡 - 계층을 추가하기 전에 정말 필요한지 고민하세요. 단순함이 최고의 아키텍처입니다.

각 계층의 인터페이스를 명확히 정의하면 계층 간 결합도를 낮출 수 있습니다.
상위 계층은 하위 계층의 구현 세부사항을 몰라야 합니다. 추상화를 유지하세요.

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

#LLM#HierarchicalAgents#Supervisor#Worker#TaskDecomposition#LLM,계층,에이전트

Hierarchical Agents 완벽 가이드

# Hierarchical Agents 완벽 가이드 AI 에이전트를 계층적으로 구성하여 복잡한 작업을 효율적으로 처리하는 방법을 배웁니다. Supervisor-Worker 패턴부터 실전 프로젝트 관리까지 단계별로 학습합니다. --- ## 목차 1. [Supervisor-Worker 구조](#supervisor-worker-구조) 2. [태스크 분해와 위임](#태스크-분해와-위임) 3. [결과 집계](#결과-집계) 4. [실습: 계층적 에이전트 시스템](#실습:-계층적-에이전트-시스템) 5. [실습: 복잡한 프로젝트 관리](#실습:-복잡한-프로젝트-관리) --- ## 1. Supervisor-Worker 구조 어느 날 김개발 씨는 복잡한 데이터 분석 프로젝트를 맡았습니다. 데이터 수집, 전처리, 분석, 시각화까지 모든 것을 하나의 AI 에이전트로 처리하려니 코드가 너무 복잡해졌습니다. 선배 박시니어 씨가 다가와 물었습니다. "왜 하나의 에이전트로 모든 걸 하려고 해요?" **Supervisor-Worker 구조**는 관리자 역할의 Supervisor 에이전트가 전체 작업을 조율하고, 여러 Worker 에이전트가 세부 작업을 담당하는 계층적 패턴입니다. 마치 프로젝트 매니저가 팀원들에게 업무를 분배하는 것과 같습니다. 이 구조를 사용하면 복잡한 작업을 체계적으로 관리할 수 있습니다. 다음 코드를 살펴봅시다. ```python from typing import List, Dict from langchain.agents import Agent class SupervisorAgent: def __init__(self, workers: List[Agent]): # Supervisor는 여러 Worker를 관리합니다 self.workers = workers self.task_queue = [] def delegate_task(self, task: str) -> Dict: # 작업의 종류를 분석하여 적절한 Worker에게 위임 if "데이터 수집" in task: return self.workers[0].run(task) elif "분석" in task: return self.workers[1].run(task) else: return self.workers[2].run(task) ``` 김개발 씨는 입사 6개월 차 주니어 개발자입니다. 최근 AI 에이전트를 활용한 자동화 시스템을 개발하는 프로젝트에 투입되었습니다. 처음에는 간단해 보였습니다. 하나의 AI 에이전트로 모든 작업을 처리하면 되니까요. 하지만 프로젝트가 진행될수록 문제가 생겼습니다. 하나의 에이전트가 데이터 수집, 전처리, 분석, 결과 생성까지 모든 것을 처리하려니 코드가 점점 복잡해졌습니다. 에러가 발생하면 어디서 문제가 생긴 건지 찾기도 어려웠습니다. 선배 개발자 박시니어 씨가 김개발 씨의 화면을 보더니 미소를 지었습니다. "김 개발님, 조직도를 생각해 보세요. 회사에서 모든 일을 대표님이 직접 하나요?" 김개발 씨가 고개를 저었습니다. "아니요, 팀장님들이 각자 팀을 관리하고, 팀원들이 실무를 담당하죠." 박시니어 씨가 고개를 끄덕였습니다. "바로 그겁니다. AI 에이전트도 마찬가지예요." **계층적 에이전트 시스템**이란 무엇일까요? 쉽게 비유하자면, 계층적 에이전트는 회사 조직과 같습니다. 최상위에는 전체를 관리하는 매니저가 있고, 그 아래에는 각자의 전문 분야를 담당하는 팀원들이 있습니다. 매니저는 전체 프로젝트를 파악하고 팀원들에게 적절한 업무를 분배합니다. 팀원들은 자신의 전문 분야에 집중하여 최고의 결과를 만들어냅니다. 이 패턴에서 가장 핵심이 되는 것이 바로 **Supervisor-Worker 구조**입니다. 단일 에이전트 시스템이 없던 것은 아닙니다. 하지만 복잡한 작업을 처리할 때 여러 문제가 있었습니다. 첫째, 에이전트가 너무 많은 것을 알아야 했습니다. 데이터베이스 접근 방법, API 호출 방법, 데이터 분석 기법, 결과 시각화 방법까지 모든 것을 하나의 에이전트가 처리해야 했습니다. 이는 마치 한 사람이 회사의 모든 업무를 처리하는 것과 같았습니다. 둘째, 유지보수가 어려웠습니다. 데이터 수집 방식을 변경하려면 전체 에이전트 코드를 수정해야 했습니다. 작은 변경이 예상치 못한 부작용을 일으키기도 했습니다. 셋째, 확장성이 떨어졌습니다. 새로운 기능을 추가하려면 이미 복잡한 에이전트를 더 복잡하게 만들어야 했습니다. 코드는 점점 스파게티처럼 얽혀갔습니다. 바로 이런 문제를 해결하기 위해 **Supervisor-Worker 패턴**이 등장했습니다. 이 패턴을 사용하면 **책임의 분리**가 명확해집니다. Supervisor는 "무엇을 할지"를 결정하고, Worker는 "어떻게 할지"를 담당합니다. 또한 **재사용성**도 높아집니다. 데이터 수집 Worker는 다른 프로젝트에서도 그대로 사용할 수 있습니다. 무엇보다 **디버깅과 테스트**가 쉬워진다는 큰 이점이 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다. 먼저 SupervisorAgent 클래스는 여러 Worker 에이전트를 리스트로 받아 관리합니다. 이것이 핵심입니다. Supervisor는 직접 작업을 수행하지 않고, 적절한 Worker를 선택하는 역할만 합니다. delegate_task 메서드는 작업 내용을 분석하여 어떤 Worker에게 위임할지 결정합니다. 간단한 키워드 매칭을 사용했지만, 실제로는 더 정교한 분류 로직을 사용할 수 있습니다. 실제 현업에서는 어떻게 활용할까요? 예를 들어 전자상거래 플랫폼을 운영한다고 가정해봅시다. 고객 문의를 처리하는 시스템에서 Supervisor 에이전트가 문의 내용을 분석합니다. 상품 문의는 상품 정보 Worker에게, 배송 문의는 배송 추적 Worker에게, 환불 요청은 환불 처리 Worker에게 각각 전달합니다. 각 Worker는 자신의 전문 분야에 최적화되어 있어 더 정확한 답변을 제공할 수 있습니다. 하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 너무 많은 계층을 만드는 것입니다. Supervisor 아래에 또 다른 Supervisor를 두고, 그 아래에 또 Worker를 두는 식으로 복잡하게 설계하면 오히려 관리가 어려워집니다. 대부분의 경우 2-3단계의 계층이면 충분합니다. 단순함이 최고의 복잡성 해결책입니다. 다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 코드를 리팩토링하기 시작했습니다. 하나의 거대한 에이전트를 Supervisor 하나와 세 개의 Worker로 분리했습니다. 코드는 훨씬 깔끔해졌고, 각 부분을 독립적으로 테스트할 수 있게 되었습니다. Supervisor-Worker 패턴을 제대로 이해하면 더 유지보수하기 쉽고 확장 가능한 AI 시스템을 구축할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요. **실전 팁** 💡 - Supervisor는 가능한 한 단순하게 유지하세요. 복잡한 로직은 Worker에게 위임하는 것이 좋습니다. - Worker는 하나의 책임만 가지도록 설계하세요. 단일 책임 원칙을 지키면 재사용성이 높아집니다. - 처음부터 완벽한 계층 구조를 만들려 하지 마세요. 필요에 따라 점진적으로 개선하는 것이 좋습니다. --- ## 2. 태스크 분해와 위임 김개발 씨가 Supervisor-Worker 구조를 이해하고 나니 새로운 궁금증이 생겼습니다. "그런데 Supervisor는 어떻게 복잡한 작업을 적절히 나누는 거죠?" 박시니어 씨가 화이트보드를 가리키며 대답했습니다. "좋은 질문이에요. 그게 바로 태스크 분해의 핵심입니다." **태스크 분해**는 복잡한 작업을 더 작고 관리 가능한 하위 작업으로 나누는 과정입니다. Supervisor는 전체 작업을 분석하여 독립적으로 수행 가능한 단위로 쪼개고, 각각을 적절한 Worker에게 위임합니다. 이는 프로젝트 관리에서 WBS를 작성하는 것과 유사합니다. 다음 코드를 살펴봅시다. ```python from typing import List, Tuple class TaskDecomposer: def decompose(self, complex_task: str) -> List[Tuple[str, str]]: # 복잡한 작업을 하위 작업으로 분해합니다 subtasks = [] if "보고서 생성" in complex_task: # 보고서 생성 작업을 세부 단계로 분해 subtasks.append(("데이터 수집", "data_collector")) subtasks.append(("데이터 분석", "data_analyst")) subtasks.append(("그래프 생성", "visualizer")) subtasks.append(("보고서 작성", "report_writer")) return subtasks def estimate_dependencies(self, subtasks: List) -> Dict: # 각 하위 작업 간의 의존성을 파악합니다 # 예: 분석은 수집이 완료된 후에만 가능 return {"data_analyst": ["data_collector"]} ``` 김개발 씨는 이제 Supervisor와 Worker의 역할을 이해했습니다. 하지만 실제로 구현하려니 막막했습니다. "월간 매출 보고서를 생성해 주세요"라는 요청이 들어왔을 때, Supervisor는 이것을 어떻게 여러 개의 작은 작업으로 나눌 수 있을까요? 박시니어 씨가 예시를 들어주었습니다. "김 개발님이 집을 짓는다고 생각해 보세요. '집을 지어라'는 하나의 큰 작업이죠. 하지만 실제로는 어떻게 하나요?" 김개발 씨가 생각하며 대답했습니다. "음, 먼저 설계를 하고, 기초 공사를 하고, 골조를 세우고, 내부 마감을 하고..." 박시니어 씨가 웃으며 고개를 끄덕였습니다. "맞아요. 그게 바로 태스크 분해입니다." **태스크 분해**란 정확히 무엇일까요? 레시피를 생각해 봅시다. "비빔밥을 만들어라"는 하나의 목표입니다. 하지만 실제로 만들려면 여러 단계가 필요합니다. 밥을 짓고, 나물을 무치고, 고기를 볶고, 계란을 부치고, 마지막으로 모든 것을 그릇에 담아 비빕니다. 각 단계는 독립적으로 수행할 수 있으며, 일부는 동시에 진행할 수도 있습니다. 태스크 분해도 이와 같습니다. AI 에이전트 시스템에서 태스크 분해가 왜 중요할까요? 초기의 단일 에이전트 시스템에서는 "보고서를 생성해 주세요"라는 요청을 받으면 에이전트가 모든 것을 순차적으로 처리했습니다. 문제는 어느 단계에서 에러가 발생하면 처음부터 다시 시작해야 한다는 것이었습니다. 또한 일부 작업은 병렬로 처리할 수 있는데도 순차적으로만 진행했습니다. 효율성이 떨어질 수밖에 없었습니다. 더 큰 문제는 유연성 부족이었습니다. 보고서의 형식을 바꾸거나 새로운 분석 방법을 추가하려면 전체 프로세스를 수정해야 했습니다. 마치 레시피의 한 재료를 바꾸기 위해 요리 전체를 다시 만드는 것과 같았습니다. **태스크 분해**를 도입하면 이런 문제가 해결됩니다. 첫째, **병렬 처리**가 가능해집니다. 데이터를 수집하는 동안 다른 Worker가 템플릿을 준비할 수 있습니다. 둘째, **에러 처리**가 쉬워집니다. 특정 하위 작업이 실패하면 그 부분만 재시도하면 됩니다. 셋째, **재사용성**이 높아집니다. "데이터 수집" 작업은 다른 보고서를 만들 때도 그대로 사용할 수 있습니다. 위의 코드를 자세히 살펴보겠습니다. TaskDecomposer 클래스의 decompose 메서드는 복잡한 작업을 받아 하위 작업의 리스트로 반환합니다. 각 하위 작업은 작업 설명과 담당 Worker의 이름을 튜플로 가집니다. 이렇게 하면 Supervisor가 어떤 순서로 무엇을 해야 하는지 명확히 알 수 있습니다. estimate_dependencies 메서드는 더욱 중요합니다. 이것은 각 작업 간의 의존성을 파악합니다. 예를 들어 데이터를 수집하기 전에는 분석을 할 수 없습니다. 하지만 그래프 생성과 보고서 템플릿 준비는 동시에 진행할 수 있습니다. 이런 의존성 정보를 바탕으로 Supervisor는 효율적인 실행 계획을 수립합니다. 실무에서는 어떻게 활용할까요? 대형 쇼핑몰의 재고 관리 시스템을 예로 들어봅시다. "재고 부족 상품을 찾아 발주하라"는 작업이 들어옵니다. TaskDecomposer는 이것을 다음과 같이 분해합니다. 먼저 "현재 재고 조회", 그다음 "판매 추세 분석", "재고 부족 예측", "공급업체 선택", "발주서 생성", 마지막으로 "발주 실행"입니다. 각 단계는 전문화된 Worker가 담당하며, 일부는 병렬로 실행됩니다. 많은 기업에서 이런 방식으로 복잡한 비즈니스 프로세스를 자동화하고 있습니다. 특히 금융, 의료, 물류 분야에서 효과가 크다고 알려져 있습니다. 주의해야 할 점도 있습니다. 가장 흔한 실수는 **너무 작게 쪼개는 것**입니다. "보고서 제목 작성", "보고서 부제목 작성" 같은 식으로 지나치게 세분화하면 오히려 관리 오버헤드가 커집니다. 적절한 수준의 분해가 중요합니다. 일반적으로 각 하위 작업이 5-15분 정도 소요되는 것이 적당합니다. 또 다른 실수는 **의존성을 무시하는 것**입니다. 데이터를 수집하기도 전에 분석을 시작하면 에러가 발생합니다. 반드시 작업 간의 순서와 의존 관계를 명확히 해야 합니다. 김개발 씨는 박시니어 씨의 조언을 따라 태스크 분해 로직을 구현했습니다. 처음에는 복잡해 보였지만, 막상 만들고 나니 전체 시스템이 훨씬 명확해졌습니다. "이제 새로운 기능을 추가할 때도 어디를 수정해야 할지 바로 알 수 있겠어요!" 태스크 분해를 제대로 이해하면 복잡한 문제를 체계적으로 해결할 수 있습니다. 여러분도 다음 프로젝트에서 이 접근법을 시도해 보세요. **실전 팁** 💡 - 하위 작업은 독립적으로 테스트할 수 있을 만큼 명확하게 정의하세요. - 의존성 그래프를 그려보면 작업 순서를 계획하는 데 도움이 됩니다. - 처음에는 큰 단위로 나누고, 필요하면 점진적으로 세분화하세요. --- ## 3. 결과 집계 여러 Worker가 각자의 작업을 완료한 후, 김개발 씨는 새로운 문제에 직면했습니다. "각 Worker가 만든 결과물을 어떻게 하나로 합치죠?" 박시니어 씨가 커피를 한 모금 마시며 말했습니다. "그게 바로 결과 집계의 역할입니다. 가장 중요한 마지막 단계죠." **결과 집계**는 여러 Worker가 생성한 개별 결과물을 수집하고 통합하여 최종 결과를 만드는 과정입니다. Supervisor는 각 Worker의 출력을 받아 일관성을 검증하고, 필요한 형식으로 변환하며, 하나의 완전한 결과물로 조합합니다. 이는 퍼즐 조각을 맞춰 완성된 그림을 만드는 것과 같습니다. 다음 코드를 살펴봅시다. ```python from typing import List, Dict, Any class ResultAggregator: def __init__(self): self.results = {} def collect_result(self, worker_id: str, result: Any): # 각 Worker의 결과를 수집합니다 self.results[worker_id] = result def validate_results(self) -> bool: # 모든 필수 결과가 수집되었는지 확인 required = ["data_collector", "data_analyst", "visualizer"] return all(w in self.results for w in required) def aggregate(self) -> Dict: # 수집된 결과들을 하나의 최종 결과로 통합 if not self.validate_results(): raise ValueError("일부 결과가 누락되었습니다") final_report = { "data": self.results["data_collector"], "analysis": self.results["data_analyst"], "charts": self.results["visualizer"], "timestamp": "2025-01-15" } return final_report ``` 김개발 씨의 시스템은 이제 작업을 잘 분해하고 각 Worker에게 위임할 수 있게 되었습니다. 데이터 수집 Worker는 데이터를 가져오고, 분석 Worker는 인사이트를 도출하고, 시각화 Worker는 그래프를 만들었습니다. 모든 것이 순조로워 보였습니다. 하지만 문제가 생겼습니다. 각 Worker가 만든 결과물의 형식이 제각각이었습니다. 데이터 수집 Worker는 JSON을 반환했고, 분석 Worker는 텍스트를 반환했으며, 시각화 Worker는 이미지 파일 경로를 반환했습니다. 이것들을 어떻게 하나의 보고서로 만들 수 있을까요? 박시니어 씨가 설명했습니다. "레고 블록을 생각해 보세요. 각각의 블록은 그 자체로는 의미가 있지만, 이것들을 조립해야 비로소 완성된 작품이 되죠." **결과 집계**란 무엇일까요? 오케스트라를 떠올려 봅시다. 바이올린, 첼로, 플루트, 트럼펫 등 각 악기는 자신의 파트를 연주합니다. 각각은 아름답지만, 진정한 음악은 이 모든 소리가 조화를 이룰 때 탄생합니다. 지휘자는 각 악기의 소리를 듣고 전체적인 하모니를 만들어냅니다. 결과 집계도 이와 같은 역할을 합니다. 왜 결과 집계가 필요할까요? 초기 시스템에서는 각 Worker가 독립적으로 결과를 저장했습니다. 사용자는 여러 파일을 열어 각각 확인해야 했습니다. 데이터는 하나의 폴더에, 분석 결과는 다른 폴더에, 그래프는 또 다른 곳에 저장되었습니다. 이것은 마치 백과사전의 각 페이지가 다른 책에 흩어져 있는 것과 같았습니다. 더 심각한 문제는 **일관성 부족**이었습니다. 어떤 Worker는 날짜를 "2025-01-15" 형식으로 표현했고, 다른 Worker는 "15/01/2025"로 표현했습니다. 숫자도 어떤 곳은 쉼표를 사용했고, 다른 곳은 사용하지 않았습니다. 최종 사용자가 이런 불일치를 직접 처리해야 했습니다. **결과 집계**를 도입하면 이런 문제가 해결됩니다. 첫째, **통일된 형식**으로 결과를 제공할 수 있습니다. 모든 날짜는 같은 형식으로, 모든 숫자는 일관된 포맷으로 표시됩니다. 둘째, **완전성 검증**이 가능합니다. 필수적인 결과가 빠지지 않았는지 확인할 수 있습니다. 셋째, **사용자 경험**이 향상됩니다. 하나의 완성된 보고서를 받아볼 수 있습니다. 위의 코드를 단계별로 살펴보겠습니다. ResultAggregator 클래스는 결과 집계를 담당합니다. collect_result 메서드는 각 Worker가 작업을 완료할 때마다 호출되어 결과를 저장합니다. worker_id를 키로 사용하여 어떤 Worker의 결과인지 추적합니다. validate_results 메서드가 핵심입니다. 이것은 모든 필수 Worker가 결과를 제출했는지 확인합니다. 만약 데이터 수집은 완료되었지만 분석이 실패했다면, 이 메서드는 False를 반환하여 문제를 알려줍니다. 불완전한 결과를 사용자에게 제공하는 것을 방지합니다. aggregate 메서드는 실제 통합을 수행합니다. 먼저 검증을 거친 후, 각 Worker의 결과를 적절한 구조로 조합합니다. 여기서는 간단한 딕셔너리를 사용했지만, 실제로는 더 복잡한 객체나 문서를 생성할 수 있습니다. 실무에서는 어떻게 활용될까요? 고객 서비스 챗봇 시스템을 예로 들어봅시다. 고객이 "내 주문 상태를 알려주세요"라고 문의하면, 주문 조회 Worker, 배송 추적 Worker, 고객 이력 Worker가 각각 정보를 수집합니다. ResultAggregator는 이 세 가지 정보를 받아 하나의 완성된 답변을 만듭니다. "고객님의 주문은 1월 10일에 접수되어 현재 배송 중이며, 내일 도착 예정입니다. 고객님은 지난 6개월간 총 5회 구매하신 우수 고객이십니다." 같은 통합된 답변을 제공할 수 있습니다. 금융 분야에서는 더욱 중요합니다. 투자 리포트를 생성할 때, 시장 데이터 Worker, 종목 분석 Worker, 리스크 평가 Worker가 각각 작업합니다. ResultAggregator는 이 모든 정보를 검증하고 통합하여 투자자에게 일관성 있고 완전한 리포트를 제공합니다. 주의할 점이 있습니다. 흔한 실수는 **에러 처리를 무시하는 것**입니다. 어떤 Worker가 실패했을 때 어떻게 할 것인지 미리 정의해야 합니다. 전체 작업을 취소할 것인지, 부분 결과라도 제공할 것인지 결정해야 합니다. 예를 들어 시각화 Worker가 실패했다면, 텍스트 분석 결과만이라도 제공할 수 있습니다. 또 다른 주의사항은 **메모리 관리**입니다. 여러 Worker의 결과를 모두 메모리에 저장하면 큰 데이터의 경우 문제가 될 수 있습니다. 필요하다면 중간 결과를 디스크에 저장하거나 스트리밍 방식으로 처리하는 것을 고려해야 합니다. 김개발 씨는 ResultAggregator를 구현한 후 시스템을 테스트했습니다. 이제 각 Worker의 결과가 자동으로 수집되고 검증되며, 하나의 깔끔한 보고서로 완성되었습니다. "와, 이제 정말 완성된 시스템 같아요!" 결과 집계를 제대로 구현하면 사용자는 복잡한 내부 프로세스를 신경 쓸 필요 없이 완성된 결과물만 받아볼 수 있습니다. 여러분의 시스템에도 이런 통합 레이어를 추가해 보세요. **실전 팁** 💡 - 결과 검증 로직을 명확히 정의하세요. 필수 필드, 데이터 타입, 값의 범위 등을 체크합니다. - 부분 실패를 어떻게 처리할지 미리 계획하세요. 일부 Worker가 실패해도 가능한 결과는 제공할 수 있습니다. - 결과를 캐싱하여 동일한 요청에 대해 다시 계산하지 않도록 최적화할 수 있습니다. --- ## 4. 실습: 계층적 에이전트 시스템 이론을 배운 김개발 씨는 직접 계층적 에이전트 시스템을 만들어보기로 했습니다. 박시니어 씨가 실전 예제를 제안했습니다. "뉴스 요약 시스템을 만들어 볼까요? 여러 뉴스 사이트에서 기사를 수집하고, 요약하고, 분류하는 시스템입니다." **계층적 에이전트 시스템**을 직접 구현하여 Supervisor가 여러 Worker를 조율하는 실제 동작을 경험합니다. 이 실습에서는 뉴스 수집, 요약, 분류를 담당하는 세 개의 Worker와 이들을 관리하는 Supervisor를 만들어 전체 워크플로우를 완성합니다. 다음 코드를 살펴봅시다. ```python from typing import List, Dict import asyncio class NewsCollector: async def collect(self, sources: List[str]) -> List[Dict]: # 뉴스 사이트에서 기사를 수집하는 Worker articles = [] for source in sources: # 실제로는 API 호출이나 웹 스크래핑 수행 articles.append({"source": source, "content": "뉴스 내용..."}) return articles class NewsSummarizer: async def summarize(self, articles: List[Dict]) -> List[Dict]: # 기사를 요약하는 Worker for article in articles: article["summary"] = article["content"][:100] + "..." return articles class NewsClassifier: async def classify(self, articles: List[Dict]) -> List[Dict]: # 기사를 분류하는 Worker for article in articles: article["category"] = "정치" # 실제로는 AI 분류 수행 return articles class NewsSupervisor: def __init__(self): self.collector = NewsCollector() self.summarizer = NewsSummarizer() self.classifier = NewsClassifier() async def process_news(self, sources: List[str]) -> List[Dict]: # 1. 뉴스 수집 articles = await self.collector.collect(sources) # 2. 요약과 분류를 병렬로 수행 articles = await self.summarizer.summarize(articles) articles = await self.classifier.classify(articles) return articles ``` 김개발 씨는 이제 본격적으로 코드를 작성할 시간입니다. 박시니어 씨가 제안한 뉴스 요약 시스템은 실제 업무에서도 자주 사용되는 패턴입니다. 여러 소스에서 정보를 모으고, 처리하고, 분류하는 작업은 데이터 파이프라인의 전형적인 형태입니다. 김개발 씨는 먼저 전체 구조를 스케치했습니다. 가장 위에는 NewsSupervisor가 있고, 그 아래에 세 개의 Worker가 있습니다. NewsCollector는 뉴스를 수집하고, NewsSummarizer는 요약하고, NewsClassifier는 분류합니다. 간단해 보였습니다. 하지만 코드를 작성하기 시작하니 고민이 생겼습니다. "이 작업들을 순차적으로 해야 할까, 아니면 병렬로 할 수 있을까?" 박시니어 씨가 조언했습니다. "의존성을 생각해 보세요. 뉴스를 수집하기 전에 요약할 수 있나요?" 김개발 씨가 고개를 끄덕였습니다. "아, 맞아요. 수집은 먼저 해야 하고, 요약과 분류는 수집 후에 병렬로 할 수 있겠네요!" **비동기 프로그래밍**이 여기서 중요한 역할을 합니다. 일반적인 동기 프로그래밍에서는 한 작업이 끝날 때까지 기다려야 다음 작업을 시작할 수 있습니다. 뉴스 사이트 A에서 기사를 가져오는 동안 다른 일을 할 수 없습니다. 하지만 비동기 프로그래밍을 사용하면 여러 작업을 동시에 진행할 수 있습니다. 사이트 A, B, C에서 동시에 기사를 가져올 수 있습니다. Python의 asyncio 라이브러리는 이런 비동기 작업을 쉽게 만들어줍니다. async 키워드로 비동기 함수를 정의하고, await 키워드로 비동기 작업의 완료를 기다립니다. 이것은 마치 음식점에서 주문을 하고 다른 일을 하다가 호출기가 울리면 음식을 받으러 가는 것과 같습니다. 코드를 자세히 살펴보겠습니다. NewsCollector, NewsSummarizer, NewsClassifier는 각각 독립적인 Worker 클래스입니다. 각각은 하나의 명확한 책임을 가집니다. 수집은 수집만, 요약은 요약만, 분류는 분류만 담당합니다. 이것이 **단일 책임 원칙**입니다. NewsCollector의 collect 메서드를 보면 sources 리스트를 받아 각 소스에서 기사를 수집합니다. 실제 구현에서는 여기서 HTTP 요청을 보내거나 API를 호출할 것입니다. 지금은 간단히 더미 데이터를 반환하지만, 구조는 실제와 같습니다. NewsSummarizer는 받은 기사 리스트를 순회하며 각 기사의 요약을 생성합니다. 여기서는 단순히 처음 100자를 잘라내지만, 실제로는 LLM API를 호출하여 지능적인 요약을 만들 수 있습니다. NewsClassifier는 각 기사의 카테고리를 판단합니다. 정치, 경제, 사회, 문화 등으로 분류할 수 있습니다. 실제로는 텍스트 분류 모델을 사용할 것입니다. 가장 중요한 것은 NewsSupervisor입니다. 이것은 세 개의 Worker를 조율합니다. process_news 메서드를 보면 전체 워크플로우가 명확히 보입니다. 먼저 뉴스를 수집하고, 그다음 요약하고, 마지막으로 분류합니다. 각 단계가 명확하고 순서가 논리적입니다. 실제로 이 시스템을 사용하면 어떻게 될까요? 예를 들어 아침마다 주요 뉴스 사이트에서 최신 기사를 가져와 요약하고 싶다고 합시다. NewsSupervisor에게 "네이버, 다음, 조선일보"를 소스로 전달합니다. Supervisor는 NewsCollector에게 이 세 곳에서 기사를 수집하라고 지시합니다. 수집이 완료되면 NewsSummarizer가 각 기사를 간단히 요약하고, NewsClassifier가 카테고리를 태깅합니다. 최종적으로 사용자는 카테고리별로 정리된 요약 뉴스를 받아볼 수 있습니다. 이런 시스템의 장점은 **확장성**입니다. 새로운 뉴스 소스를 추가하고 싶다면? NewsCollector만 수정하면 됩니다. 요약 방식을 바꾸고 싶다면? NewsSummarizer만 수정하면 됩니다. 다른 부분에 영향을 주지 않습니다. 김개발 씨가 주의해야 할 점도 있습니다. **에러 처리**는 필수입니다. 만약 한 뉴스 사이트가 응답하지 않는다면? 전체 프로세스를 멈출 것인지, 아니면 다른 소스의 결과라도 제공할 것인지 결정해야 합니다. try-except 블록을 사용하여 각 Worker의 에러를 적절히 처리해야 합니다. **타임아웃**도 중요합니다. 어떤 뉴스 사이트가 매우 느리다면 무한정 기다릴 수 없습니다. asyncio.wait_for를 사용하여 각 작업에 타임아웃을 설정할 수 있습니다. **리소스 관리**도 고려해야 합니다. 동시에 100개의 뉴스 사이트에 접속하면 네트워크 리소스나 메모리 문제가 생길 수 있습니다. asyncio.Semaphore를 사용하여 동시 실행 수를 제한할 수 있습니다. 김개발 씨는 코드를 완성하고 테스트를 시작했습니다. 처음에는 한 개의 뉴스 소스로 시작했고, 잘 작동하는 것을 확인한 후 여러 소스로 확장했습니다. 각 Worker가 독립적으로 작동하니 디버깅도 쉬웠습니다. 박시니어 씨가 코드 리뷰를 하며 칭찬했습니다. "잘했어요. 이제 실제 프로젝트에서도 충분히 사용할 수 있는 수준이네요." 계층적 에이전트 시스템을 직접 구현해 보면 이론으로만 배울 때와는 다른 깊은 이해를 얻을 수 있습니다. 여러분도 간단한 프로젝트부터 시작해서 점진적으로 확장해 보세요. **실전 팁** 💡 - 처음에는 동기 버전으로 만들어 로직을 검증한 후, 비동기로 전환하는 것이 안전합니다. - 각 Worker를 독립적인 파일로 분리하면 코드 관리가 쉬워집니다. - 로깅을 적극 활용하여 각 단계에서 무슨 일이 일어나는지 추적하세요. --- ## 5. 실습: 복잡한 프로젝트 관리 기본적인 계층적 시스템을 만든 김개발 씨에게 박시니어 씨가 더 도전적인 과제를 제시했습니다. "이번에는 3단계 계층 구조를 만들어 볼까요? 최상위 Supervisor, 중간 관리자, 그리고 실제 작업 Worker로 구성된 시스템입니다." **다단계 계층 구조**를 구현하여 복잡한 프로젝트를 관리하는 방법을 학습합니다. 최상위 Supervisor는 전체 프로젝트를 관리하고, 중간 관리자들은 각자의 영역을 담당하며, 최하위 Worker들이 실제 작업을 수행합니다. 이는 대기업의 조직 구조와 유사합니다. 다음 코드를 살펴봅시다. ```python class DataWorker: async def fetch_data(self, source: str): return f"{source}에서 데이터 수집 완료" class AnalysisWorker: async def analyze(self, data: str): return f"{data}에 대한 분석 완료" class DataManager: # 중간 관리자: 데이터 관련 작업을 총괄 def __init__(self): self.collector = DataWorker() self.analyzer = AnalysisWorker() async def process_data(self, sources: List[str]): results = [] for source in sources: data = await self.collector.fetch_data(source) analysis = await self.analyzer.analyze(data) results.append(analysis) return results class ReportWorker: async def generate(self, content: str): return f"보고서 생성: {content}" class ReportManager: # 중간 관리자: 보고서 관련 작업을 총괄 def __init__(self): self.writer = ReportWorker() async def create_report(self, data: List[str]): return await self.writer.generate(str(data)) class ProjectSupervisor: # 최상위 Supervisor: 전체 프로젝트 조율 def __init__(self): self.data_manager = DataManager() self.report_manager = ReportManager() async def execute_project(self, sources: List[str]): # 1단계: 데이터 처리 analysis = await self.data_manager.process_data(sources) # 2단계: 보고서 생성 report = await self.report_manager.create_report(analysis) return report ``` 김개발 씨는 이제 더 복잡한 시나리오에 도전할 준비가 되었습니다. 박시니어 씨가 설명했습니다. "실제 대규모 프로젝트에서는 단순한 2단계 구조로는 부족할 때가 있어요. 여러 팀이 협력해야 하는 프로젝트를 생각해 보세요." 김개발 씨가 자신의 회사 조직도를 떠올렸습니다. 최상위에는 CTO가 있고, 그 아래에 데이터팀 팀장, 개발팀 팀장, 디자인팀 팀장이 있습니다. 각 팀장 아래에는 실무자들이 있습니다. "아, 이런 구조를 코드로 만드는 거군요!" **다단계 계층 구조**는 왜 필요할까요? 작은 프로젝트에서는 Supervisor 하나가 모든 Worker를 직접 관리할 수 있습니다. 하지만 프로젝트가 커지면 어떨까요? 30개의 Worker를 Supervisor 하나가 직접 관리하면 복잡도가 급격히 증가합니다. 마치 CEO가 모든 직원을 직접 관리하려는 것과 같습니다. 이때 **중간 관리자 계층**을 도입하면 문제가 해결됩니다. 최상위 Supervisor는 큰 그림만 보고, 중간 관리자들이 세부 사항을 담당합니다. 각 관리자는 자신의 영역에 대한 전문성을 가지고 있습니다. 코드를 단계별로 살펴보겠습니다. 가장 아래 계층에는 DataWorker와 AnalysisWorker, ReportWorker가 있습니다. 이들은 실제 작업을 수행하는 실무자들입니다. 데이터를 가져오고, 분석하고, 보고서를 작성합니다. 각자 하나의 간단한 작업만 담당합니다. 중간 계층에는 DataManager와 ReportManager가 있습니다. DataManager는 데이터 수집과 분석을 총괄합니다. DataWorker에게 데이터를 수집하라고 지시하고, 그 결과를 받아 AnalysisWorker에게 분석을 요청합니다. 이 두 작업의 조율은 DataManager의 책임입니다. ReportManager는 보고서 생성을 담당합니다. 데이터가 준비되면 ReportWorker를 통해 최종 보고서를 만듭니다. 보고서의 형식, 내용 구성 등은 ReportManager가 결정합니다. 최상위에는 ProjectSupervisor가 있습니다. 이것은 전체 프로젝트의 흐름을 조율합니다. "먼저 데이터를 처리하고, 그다음 보고서를 만든다"는 큰 그림을 담당합니다. 구체적인 데이터 처리 방법이나 보고서 작성 방법은 중간 관리자들에게 위임합니다. 이런 구조의 장점은 무엇일까요? 첫째, **관심사의 분리**가 명확합니다. ProjectSupervisor는 데이터가 어떻게 수집되는지 몰라도 됩니다. DataManager에게 "데이터를 처리해 주세요"라고만 요청하면 됩니다. 세부 사항은 DataManager가 알아서 처리합니다. 둘째, **변경에 유연**합니다. 데이터 수집 방식을 바꾸고 싶다면 DataManager와 그 아래의 Worker만 수정하면 됩니다. ProjectSupervisor나 ReportManager는 영향을 받지 않습니다. 셋째, **테스트가 쉽**습니다. 각 계층을 독립적으로 테스트할 수 있습니다. DataManager가 올바르게 작동하는지 확인한 후, ProjectSupervisor를 테스트할 수 있습니다. 실제 업무에서는 어떻게 활용될까요? 대형 전자상거래 플랫폼의 일일 리포트 시스템을 생각해 봅시다. 최상위 ReportSupervisor는 전체 리포트 생성을 관리합니다. 그 아래에 SalesManager는 판매 데이터를 담당하고, InventoryManager는 재고 데이터를 담당하며, CustomerManager는 고객 데이터를 담당합니다. 각 Manager 아래에는 실제 데이터베이스 쿼리를 실행하고 계산을 수행하는 Worker들이 있습니다. 이런 구조를 사용하면 하루에 수백만 건의 거래를 분석하여 경영진에게 통합 리포트를 제공할 수 있습니다. 각 부분을 병렬로 처리하여 성능도 최적화할 수 있습니다. 주의해야 할 점도 있습니다. **과도한 계층화**는 피해야 합니다. 계층이 너무 많으면 오히려 복잡도가 증가합니다. 일반적으로 3-4단계를 넘지 않는 것이 좋습니다. 각 계층을 거칠 때마다 오버헤드가 발생하므로 적절한 균형이 필요합니다. **통신 오버헤드**도 고려해야 합니다. 계층 간 데이터 전달이 빈번하면 성능 문제가 생길 수 있습니다. 필요한 경우 중간 결과를 캐싱하거나 배치 처리를 사용할 수 있습니다. **에러 전파**도 중요합니다. 최하위 Worker에서 발생한 에러가 어떻게 최상위까지 전달될지 설계해야 합니다. 각 계층에서 적절히 에러를 처리하고 로깅해야 합니다. 김개발 씨는 이 복잡한 구조를 천천히 구현했습니다. 먼저 가장 아래 Worker들을 만들고 테스트했습니다. 그다음 중간 Manager들을 만들어 Worker들과 통합했습니다. 마지막으로 최상위 Supervisor를 추가했습니다. 단계별로 검증하니 전체 시스템도 안정적으로 작동했습니다. 박시니어 씨가 최종 결과를 보며 만족스러워했습니다. "이제 정말 엔터프라이즈급 시스템이네요. 이 구조라면 향후 몇 년간 확장해도 문제없을 겁니다." 김개발 씨는 뿌듯했습니다. 처음에는 복잡해 보였던 계층적 에이전트 시스템을 이제는 자신 있게 설계하고 구현할 수 있게 되었습니다. 다단계 계층 구조는 복잡해 보이지만, 각 계층의 책임을 명확히 하면 오히려 전체 시스템을 이해하기 쉬워집니다. 여러분도 작은 프로젝트부터 시작해서 필요에 따라 계층을 추가해 보세요. **실전 팁** 💡 - 계층을 추가하기 전에 정말 필요한지 고민하세요. 단순함이 최고의 아키텍처입니다. - 각 계층의 인터페이스를 명확히 정의하면 계층 간 결합도를 낮출 수 있습니다. - 상위 계층은 하위 계층의 구현 세부사항을 몰라야 합니다. 추상화를 유지하세요. --- 이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

카테고리: LLM,계층,에이전트

언어: Python

태그: LLM, HierarchicalAgents, Supervisor, Worker, TaskDecomposition, LLM,계층,에이전트

작성자: AI Generated

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

CodeDeck

Hierarchical Agents 완벽 가이드

목차

1. Supervisor-Worker 구조

2. 태스크 분해와 위임

3. 결과 집계

4. 실습: 계층적 에이전트 시스템

5. 실습: 복잡한 프로젝트 관리

댓글 (0)

함께 보면 좋은 카드 뉴스

Human-in-the-Loop Agents 완벽 가이드

Agent Orchestration 완벽 가이드

Swarm Intelligence 분산 처리 완벽 가이드

Agent Debate 패턴 완벽 가이드

Multi-Agent 아키텍처 완벽 가이드