Multi-Agent Orchestration 완벽 가이드

여러 AI 에이전트가 협력하여 복잡한 문제를 해결하는 Multi-Agent Orchestration의 핵심 개념을 다룹니다. 하위 에이전트 생성부터 분산 시스템까지, 실무에서 바로 적용할 수 있는 패턴을 소설처럼 쉽게 설명합니다.

---

목차

1. 하위_에이전트_생성_및_관리
2. 에이전트_간_통신_프로토콜
3. 역할_기반_에이전트_설계
4. 협업_패턴_투표_합의_위임
5. 에이전트_풀_관리
6. 분산_에이전트_시스템

---

1. 하위 에이전트 생성 및 관리

김개발 씨는 AI 스타트업에 입사한 지 얼마 되지 않은 주니어 개발자입니다. 어느 날 팀장님이 다가와 말했습니다.

"우리 서비스에 여러 AI 에이전트가 협력하는 시스템을 구축해야 해요. 먼저 하위 에이전트를 생성하고 관리하는 부분부터 시작해볼까요?"

하위 에이전트 생성 및 관리는 마치 회사에서 팀장이 팀원들을 채용하고 업무를 배정하는 것과 같습니다. 메인 오케스트레이터가 필요에 따라 전문화된 에이전트들을 생성하고, 그들의 생명주기를 관리합니다.

이를 통해 복잡한 작업을 여러 전문가에게 분담시킬 수 있습니다.

다음 코드를 살펴봅시다.

from abc import ABC, abstractmethod
from typing import Dict, Any
import uuid

class BaseAgent(ABC):
    def __init__(self, name: str, capabilities: list):
        self.id = str(uuid.uuid4())
        self.name = name
        self.capabilities = capabilities
        self.status = "idle"

    @abstractmethod
    async def execute(self, task: Dict[str, Any]) -> Dict[str, Any]:
        pass

class AgentFactory:
    # 에이전트 타입별 클래스 등록
    _registry: Dict[str, type] = {}

    @classmethod
    def register(cls, agent_type: str, agent_class: type):
        cls._registry[agent_type] = agent_class

    @classmethod
    def create(cls, agent_type: str, **kwargs) -> BaseAgent:
        # 등록된 타입에 따라 적절한 에이전트 인스턴스 생성
        if agent_type not in cls._registry:
            raise ValueError(f"Unknown agent type: {agent_type}")
        return cls._registry[agent_type](**kwargs)

김개발 씨는 처음 듣는 용어에 약간 당황했습니다. "하위 에이전트요?

그게 뭔가요?" 팀장님이 웃으며 설명을 시작했습니다. "회사를 생각해봐요.

사장님 혼자서 모든 일을 할 수 있을까요? 당연히 불가능하죠.

그래서 각 분야의 전문가들을 채용합니다. 마케팅 담당자, 개발자, 디자이너처럼요.

하위 에이전트도 마찬가지입니다." Multi-Agent 시스템에서 오케스트레이터는 사장님과 같은 역할을 합니다. 혼자서 모든 작업을 처리하는 대신, 전문화된 하위 에이전트들을 생성하여 각자의 역할을 맡깁니다.

그렇다면 왜 이런 구조가 필요할까요? 단일 에이전트가 모든 작업을 처리하면 코드가 거대해지고 복잡해집니다.

새로운 기능을 추가할 때마다 기존 코드를 수정해야 하고, 버그가 발생하면 전체 시스템이 멈출 수 있습니다. 마치 1인 기업이 성장의 한계에 부딪히는 것과 같습니다.

팩토리 패턴을 사용하면 이 문제를 우아하게 해결할 수 있습니다. 위 코드의 AgentFactory 클래스를 살펴보겠습니다.

먼저 _registry라는 딕셔너리가 있습니다. 여기에 에이전트 타입과 해당 클래스를 등록해둡니다.

마치 인사팀이 채용 가능한 직군 목록을 관리하는 것과 같습니다. register 메서드는 새로운 에이전트 타입을 시스템에 등록합니다.

나중에 "번역 에이전트"나 "분석 에이전트" 같은 새로운 타입이 필요하면 이 메서드로 추가하면 됩니다. create 메서드가 실제로 에이전트를 생성하는 핵심입니다.

요청받은 타입이 등록되어 있는지 확인하고, 해당 클래스의 인스턴스를 만들어 반환합니다. BaseAgent 클래스는 모든 에이전트의 부모 클래스입니다.

고유 ID, 이름, 능력 목록, 상태 정보를 기본으로 가집니다. 추상 메서드 execute는 각 하위 클래스에서 반드시 구현해야 하는 핵심 로직입니다.

실무에서는 에이전트의 상태 관리가 매우 중요합니다. idle, working, error, terminated 같은 상태를 추적하여 시스템의 건강 상태를 모니터링할 수 있습니다.

주의할 점이 있습니다. 에이전트를 무한정 생성하면 메모리 문제가 발생합니다.

따라서 에이전트 풀을 관리하거나, 사용이 끝난 에이전트를 적절히 정리하는 로직이 필요합니다. 김개발 씨가 고개를 끄덕였습니다.

"아, 그러니까 필요할 때 전문가를 부르고, 일이 끝나면 정리하는 거군요!" 팀장님이 미소 지었습니다. "정확해요.

이제 다음으로 이 에이전트들이 어떻게 서로 대화하는지 알아볼까요?"

실전 팁

💡 - 에이전트 생성 시 고유 ID를 부여하여 추적과 디버깅을 용이하게 하세요

• 팩토리 패턴을 사용하면 새로운 에이전트 타입 추가가 간편해집니다
• 에이전트 상태를 항상 추적하여 시스템 모니터링에 활용하세요

---

2. 에이전트 간 통신 프로토콜

하위 에이전트를 만드는 법을 배운 김개발 씨에게 새로운 과제가 주어졌습니다. "에이전트들끼리 어떻게 대화하게 만들죠?" 박시니어 씨가 옆에서 힌트를 줍니다.

"사람들이 어떻게 소통하는지 생각해봐요. 편지, 전화, 회의...

에이전트도 마찬가지예요."

에이전트 간 통신 프로토콜은 에이전트들이 서로 정보를 주고받는 표준화된 방식입니다. 마치 회사에서 공식 문서 양식을 정해두고 모든 부서가 같은 형식으로 소통하는 것과 같습니다.

명확한 메시지 구조와 라우팅 규칙이 있어야 혼란 없이 협업할 수 있습니다.

다음 코드를 살펴봅시다.

from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum
from datetime import datetime
import asyncio

class MessageType(Enum):
    REQUEST = "request"
    RESPONSE = "response"
    BROADCAST = "broadcast"
    ERROR = "error"

@dataclass
class AgentMessage:
    sender_id: str
    receiver_id: str
    msg_type: MessageType
    payload: Dict[str, Any]
    correlation_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: datetime = field(default_factory=datetime.now)

class MessageBus:
    def __init__(self):
        self._subscribers: Dict[str, asyncio.Queue] = {}

    async def subscribe(self, agent_id: str) -> asyncio.Queue:
        # 에이전트별 메시지 큐 생성
        self._subscribers[agent_id] = asyncio.Queue()
        return self._subscribers[agent_id]

    async def publish(self, message: AgentMessage):
        # 수신자에게 메시지 전달
        if message.receiver_id in self._subscribers:
            await self._subscribers[message.receiver_id].put(message)

"사람들이 회사에서 어떻게 소통하나요?" 박시니어 씨의 질문에 김개발 씨가 대답했습니다. "이메일이나 슬랙으로요.

아, 가끔 회의도 하고요." "맞아요. 그런데 이메일을 보낼 때 아무렇게나 쓰나요?" 김개발 씨가 고개를 저었습니다.

"아니요, 받는 사람, 제목, 본문 형식이 정해져 있죠." 바로 이것이 통신 프로토콜의 핵심입니다. 에이전트들도 정해진 형식에 맞춰 메시지를 주고받아야 합니다.

그렇지 않으면 서로 무슨 말인지 이해하지 못하는 혼란이 발생합니다. 위 코드에서 AgentMessage 클래스가 바로 그 표준 양식입니다.

마치 회사의 공문서 템플릿과 같습니다. sender_id는 "누가 보냈는지"를 나타냅니다.

receiver_id는 "누구에게 보내는지"입니다. msg_type은 메시지의 성격을 정의합니다.

요청인지, 응답인지, 전체 공지인지, 에러 알림인지 구분합니다. payload가 실제 내용을 담는 곳입니다.

딕셔너리 형태로 어떤 데이터든 담을 수 있습니다. correlation_id는 매우 중요한 필드입니다.

요청과 응답을 연결해주는 고리 역할을 합니다. 예를 들어 분석 에이전트가 데이터 에이전트에게 "사용자 정보 좀 줘"라고 요청했다고 합시다.

나중에 데이터 에이전트가 응답을 보낼 때, correlation_id가 같으면 "아, 이게 아까 그 요청에 대한 답이구나"라고 알 수 있습니다. MessageBus 클래스는 우체국과 같습니다.

모든 메시지가 이곳을 거쳐 전달됩니다. subscribe 메서드로 에이전트가 자신의 우편함을 등록합니다.

asyncio.Queue를 사용하여 비동기적으로 메시지를 받을 수 있습니다. publish 메서드는 메시지를 적절한 수신자의 우편함에 넣어줍니다.

왜 이렇게 중앙 집중식 메시지 버스를 사용할까요? 에이전트들이 서로 직접 연결되면 연결 관계가 복잡해집니다.

10개의 에이전트가 모두 연결되려면 45개의 연결이 필요합니다. 하지만 메시지 버스를 사용하면 각 에이전트는 버스와만 연결하면 됩니다.

10개의 연결로 충분합니다. 실무에서는 RabbitMQ나 Redis Pub/Sub 같은 전문 메시지 브로커를 사용하기도 합니다.

하지만 기본 원리는 동일합니다. 김개발 씨가 질문했습니다.

"그런데 메시지가 유실되면 어떻게 해요?" 좋은 질문입니다. 실제 시스템에서는 메시지 확인(ACK) 메커니즘, 재전송 로직, 데드레터 큐 같은 안전장치를 추가합니다.

실전 팁

💡 - correlation_id를 활용하여 요청-응답 쌍을 추적하세요

• 메시지 타입을 Enum으로 정의하면 오타로 인한 버그를 방지할 수 있습니다
• 비동기 큐를 사용하면 에이전트가 블로킹 없이 메시지를 처리할 수 있습니다

---

3. 역할 기반 에이전트 설계

김개발 씨의 팀에서 새로운 프로젝트가 시작되었습니다. 고객 문의를 자동으로 처리하는 AI 시스템입니다.

팀장님이 화이트보드에 그림을 그리며 말했습니다. "각 에이전트에게 명확한 역할을 부여해야 해요.

마치 축구팀에서 골키퍼, 수비수, 공격수가 각자 맡은 바가 있듯이요."

역할 기반 에이전트 설계는 각 에이전트에게 명확한 책임과 권한을 부여하는 설계 방식입니다. 마치 회사의 조직도처럼 누가 무엇을 담당하는지 명확히 정의합니다.

이를 통해 작업 분배가 효율적이 되고, 각 에이전트가 자신의 전문 영역에 집중할 수 있습니다.

다음 코드를 살펴봅시다.

from enum import Enum
from typing import List, Callable

class AgentRole(Enum):
    COORDINATOR = "coordinator"  # 작업 분배 및 조율
    ANALYZER = "analyzer"        # 데이터 분석
    EXECUTOR = "executor"        # 실제 작업 수행
    VALIDATOR = "validator"      # 결과 검증

class RoleBasedAgent(BaseAgent):
    def __init__(self, name: str, role: AgentRole, capabilities: list):
        super().__init__(name, capabilities)
        self.role = role
        self._permissions: List[str] = []
        self._assign_permissions()

    def _assign_permissions(self):
        # 역할에 따른 권한 자동 부여
        permission_map = {
            AgentRole.COORDINATOR: ["delegate", "monitor", "escalate"],
            AgentRole.ANALYZER: ["read_data", "compute", "report"],
            AgentRole.EXECUTOR: ["write_data", "call_api", "transform"],
            AgentRole.VALIDATOR: ["read_data", "verify", "approve"]
        }
        self._permissions = permission_map.get(self.role, [])

    def can_perform(self, action: str) -> bool:
        # 해당 액션 수행 권한 확인
        return action in self._permissions

"축구팀을 생각해보세요." 팀장님의 설명이 이어졌습니다. "11명 모두가 공을 쫓아다니면 어떻게 될까요?" 김개발 씨가 웃으며 대답했습니다.

"엉망진창이 되겠죠. 수비할 사람도 없고요." "바로 그거예요.

그래서 역할 분담이 필요합니다. 골키퍼는 골대를 지키고, 수비수는 상대 공격을 막고, 미드필더는 볼을 배급하고, 공격수는 골을 넣습니다." Multi-Agent 시스템에서도 마찬가지입니다.

모든 에이전트가 똑같은 일을 하면 비효율적이고 충돌이 발생합니다. 각자 맡은 바가 있어야 합니다.

위 코드에서 AgentRole Enum은 네 가지 핵심 역할을 정의합니다. COORDINATOR는 감독과 같습니다.

전체 작업을 조망하고, 적절한 에이전트에게 작업을 분배합니다. 직접 일을 하기보다는 지휘하는 역할입니다.

ANALYZER는 분석가입니다. 데이터를 읽고 패턴을 찾고 인사이트를 도출합니다.

복잡한 계산이나 판단이 필요한 작업을 담당합니다. EXECUTOR는 실행자입니다.

실제로 뭔가를 변경하는 역할입니다. API를 호출하거나, 데이터를 수정하거나, 외부 시스템과 상호작용합니다.

VALIDATOR는 검증자입니다. 다른 에이전트가 수행한 작업이 올바른지 확인합니다.

품질 관리 담당자라고 생각하면 됩니다. _assign_permissions 메서드를 주목하세요.

역할에 따라 자동으로 권한이 부여됩니다. COORDINATOR는 delegate(위임), monitor(모니터링), escalate(상위 보고) 권한을 가집니다.

EXECUTOR는 write_data(데이터 쓰기), call_api(API 호출) 권한이 있습니다. 이렇게 권한을 분리하면 보안이 강화됩니다.

분석만 하는 에이전트가 실수로 데이터를 수정하는 일을 방지할 수 있습니다. can_perform 메서드는 특정 액션을 수행할 권한이 있는지 확인합니다.

작업 실행 전에 항상 이 검사를 수행하면 무단 접근을 막을 수 있습니다. 실무에서는 이런 역할들을 더 세분화하기도 합니다.

예를 들어 "읽기 전용 ANALYZER"와 "고급 분석 ANALYZER"를 구분할 수 있습니다. 비즈니스 요구사항에 맞게 유연하게 설계하면 됩니다.

박시니어 씨가 덧붙였습니다. "처음부터 완벽하게 역할을 나누려고 하지 마세요.

시작은 단순하게 하고, 시스템이 커지면서 점차 세분화하면 됩니다."

실전 팁

💡 - 역할은 단일 책임 원칙(SRP)을 따르도록 설계하세요

• 권한은 최소 권한 원칙에 따라 꼭 필요한 것만 부여하세요
• 역할 변경이 필요하면 새 에이전트를 생성하는 것이 기존 에이전트 수정보다 안전합니다

---

4. 협업 패턴 투표 합의 위임

프로젝트가 본격적으로 진행되면서 새로운 문제가 생겼습니다. 여러 에이전트가 서로 다른 의견을 낼 때 어떻게 결정을 내려야 할까요?

김개발 씨가 고민하자 박시니어 씨가 말했습니다. "민주주의 사회에서 결정을 내리는 방법을 생각해봐요.

투표도 하고, 토론 끝에 합의도 하고, 전문가에게 위임도 하잖아요."

협업 패턴은 여러 에이전트가 함께 결정을 내리는 방법론입니다. 투표는 다수결로 빠르게 결정하고, 합의는 모든 에이전트가 동의할 때까지 논의하며, 위임은 전문가 에이전트에게 결정권을 맡깁니다.

상황에 따라 적절한 패턴을 선택해야 합니다.

다음 코드를 살펴봅시다.

from typing import List, Dict, Any
from collections import Counter

class CollaborationPattern:
    @staticmethod
    async def voting(agents: List[BaseAgent], question: Dict) -> Any:
        # 다수결 투표 패턴
        votes = []
        for agent in agents:
            vote = await agent.execute({"action": "vote", "question": question})
            votes.append(vote["choice"])
        # 가장 많은 표를 받은 선택 반환
        return Counter(votes).most_common(1)[0][0]

    @staticmethod
    async def consensus(agents: List[BaseAgent], proposal: Dict, max_rounds: int = 5) -> bool:
        # 합의 도출 패턴
        for round_num in range(max_rounds):
            responses = []
            for agent in agents:
                result = await agent.execute({"action": "evaluate", "proposal": proposal})
                responses.append(result["agree"])

            if all(responses):  # 모두 동의하면 합의 성공
                return True
            # 반대 의견 수집하여 제안 수정
            proposal = await CollaborationPattern._refine_proposal(proposal, responses)
        return False  # 합의 실패

    @staticmethod
    async def delegation(coordinator: BaseAgent, specialists: Dict[str, BaseAgent], task: Dict) -> Any:
        # 위임 패턴: 적절한 전문가에게 작업 위임
        task_type = task.get("type", "general")
        if task_type in specialists:
            return await specialists[task_type].execute(task)
        return await coordinator.execute(task)  # 기본은 코디네이터가 처리

"회사에서 중요한 결정을 어떻게 내리나요?" 박시니어 씨의 질문에 김개발 씨가 곰곰이 생각했습니다. "음...

때에 따라 다른 것 같아요. 점심 메뉴 같은 건 그냥 투표하고, 중요한 프로젝트 방향은 다 같이 논의해서 합의하고, 기술적인 결정은 팀장님한테 맡기기도 하고요." "훌륭해요!

에이전트 시스템에서도 똑같습니다." 첫 번째 패턴은 **투표(Voting)**입니다. 가장 단순하고 빠른 방법입니다.

여러 에이전트에게 질문을 던지고, 다수결로 결정합니다. 위 코드의 voting 메서드를 보면, 모든 에이전트로부터 투표를 받아 Counter로 집계합니다.

most_common(1)은 가장 많은 표를 받은 선택지를 반환합니다. 투표는 언제 사용할까요?

빠른 결정이 필요할 때, 또는 모든 선택지가 비슷한 가치를 가질 때 적합합니다. 예를 들어 "이 텍스트의 감정이 긍정인가 부정인가?"를 판단할 때 여러 분석 에이전트의 투표를 활용할 수 있습니다.

두 번째 패턴은 **합의(Consensus)**입니다. 모든 참여자가 동의해야 결정이 확정됩니다.

consensus 메서드는 최대 max_rounds번까지 합의를 시도합니다. 한 라운드에서 모든 에이전트가 동의하면 성공입니다.

반대 의견이 있으면 제안을 수정하고 다시 시도합니다. 합의는 시간이 오래 걸리지만, 모두가 납득하는 결정을 내릴 수 있습니다.

중요한 의사결정이나 되돌리기 어려운 작업에 적합합니다. 예를 들어 "이 코드를 프로덕션에 배포해도 될까?"라는 결정은 합의 패턴이 좋습니다.

세 번째 패턴은 **위임(Delegation)**입니다. 해당 분야의 전문가에게 결정권을 넘깁니다.

delegation 메서드는 작업 타입을 확인하고, 해당 분야 전문가가 있으면 그에게 작업을 맡깁니다. 전문가가 없으면 기본적으로 코디네이터가 처리합니다.

위임은 효율성이 높습니다. 번역 작업은 번역 에이전트에게, 이미지 분석은 비전 에이전트에게 맡기면 각자의 강점을 살릴 수 있습니다.

실무에서는 이 세 가지 패턴을 혼합해서 사용합니다. 먼저 위임으로 적절한 에이전트 그룹을 선택하고, 그 그룹 내에서 투표나 합의로 최종 결정을 내리는 식입니다.

"어떤 패턴을 선택해야 할지 어떻게 알 수 있어요?" 김개발 씨의 질문에 박시니어 씨가 답했습니다. "결정의 중요도, 시간 제약, 전문성 필요 여부를 고려하세요.

가볍고 빠른 결정은 투표, 무거운 결정은 합의, 전문 지식이 필요하면 위임입니다."

실전 팁

💡 - 투표 시 동점이 나오면 어떻게 할지 미리 정해두세요 (랜덤, 코디네이터 결정 등)

• 합의가 무한 루프에 빠지지 않도록 최대 라운드 수를 설정하세요
• 위임 대상이 없을 때의 기본 처리자(fallback)를 항상 정의해두세요

---

5. 에이전트 풀 관리

시스템이 점점 커지면서 동시에 처리해야 할 작업이 늘어났습니다. 김개발 씨가 에이전트를 마구 생성했더니 서버 메모리가 위험 수준까지 올라갔습니다.

팀장님이 긴급 회의를 소집했습니다. "에이전트를 무한정 만들 수는 없어요.

수영장의 레인처럼 정해진 수의 에이전트를 효율적으로 관리해야 합니다."

에이전트 풀 관리는 미리 생성해둔 에이전트들을 재사용하여 자원을 효율적으로 활용하는 기법입니다. 마치 회사에서 정규직 직원을 유지하며 업무를 배분하는 것과 같습니다.

필요할 때마다 새로 채용하고 해고하는 것보다 훨씬 효율적입니다.

다음 코드를 살펴봅시다.

import asyncio
from typing import Optional

class AgentPool:
    def __init__(self, agent_factory: AgentFactory, agent_type: str, pool_size: int = 5):
        self._factory = agent_factory
        self._agent_type = agent_type
        self._pool_size = pool_size
        self._available: asyncio.Queue = asyncio.Queue()
        self._in_use: set = set()

    async def initialize(self):
        # 풀 초기화: 미리 에이전트 생성
        for _ in range(self._pool_size):
            agent = self._factory.create(self._agent_type, name=f"{self._agent_type}_worker")
            await self._available.put(agent)

    async def acquire(self, timeout: float = 30.0) -> Optional[BaseAgent]:
        # 사용 가능한 에이전트 획득
        try:
            agent = await asyncio.wait_for(self._available.get(), timeout=timeout)
            self._in_use.add(agent.id)
            agent.status = "working"
            return agent
        except asyncio.TimeoutError:
            return None  # 타임아웃 시 None 반환

    async def release(self, agent: BaseAgent):
        # 에이전트 반환
        if agent.id in self._in_use:
            self._in_use.remove(agent.id)
            agent.status = "idle"
            await self._available.put(agent)

    @property
    def available_count(self) -> int:
        return self._available.qsize()

"데이터베이스 커넥션 풀 알아요?" 팀장님의 질문에 김개발 씨가 고개를 끄덕였습니다. "네, DB 연결을 미리 만들어두고 재사용하는 거요." "에이전트 풀도 똑같은 개념이에요.

매번 새로 만들지 않고, 미리 만들어둔 에이전트를 돌려쓰는 겁니다." 에이전트를 생성하는 것은 비용이 드는 작업입니다. 메모리를 할당하고, 초기화 작업을 수행하고, 필요한 리소스를 로드해야 합니다.

작업이 끝날 때마다 에이전트를 삭제하면 이 비용을 반복해서 지불해야 합니다. 에이전트 풀은 이 문제를 해결합니다.

시스템 시작 시 일정 수의 에이전트를 미리 생성해두고, 필요할 때 빌려 쓰고, 작업이 끝나면 반환합니다. 위 코드의 AgentPool 클래스를 살펴보겠습니다.

initialize 메서드는 풀을 초기화합니다. pool_size만큼 에이전트를 미리 생성하여 _available 큐에 넣어둡니다.

이 작업은 시스템 시작 시 한 번만 수행됩니다. acquire 메서드는 에이전트를 "빌리는" 것입니다.

_available 큐에서 하나를 꺼내고, _in_use 집합에 추가하여 사용 중임을 표시합니다. 중요한 점은 타임아웃이 있다는 것입니다.

모든 에이전트가 사용 중이면 최대 timeout 초까지 기다립니다. release 메서드는 에이전트를 "반환"합니다.

_in_use에서 제거하고 다시 _available 큐에 넣습니다. 상태도 idle로 바꿉니다.

왜 asyncio.Queue를 사용할까요? 멀티스레딩 환경에서 **경쟁 조건(race condition)**을 방지하기 위해서입니다.

두 작업이 동시에 같은 에이전트를 가져가려 하면 문제가 생깁니다. Queue는 이를 안전하게 처리합니다.

available_count 프로퍼티로 현재 사용 가능한 에이전트 수를 확인할 수 있습니다. 모니터링이나 스케일링 결정에 유용합니다.

실무에서는 동적 스케일링을 추가하기도 합니다. 부하가 높을 때 풀 크기를 늘리고, 한가할 때 줄이는 것입니다.

클라우드 환경에서는 이런 탄력적인 관리가 비용 절감에 큰 도움이 됩니다. 주의할 점이 있습니다.

에이전트를 빌린 후 반환하지 않으면 리소스 누수가 발생합니다. 예외가 발생해도 반드시 release가 호출되도록 try-finally나 컨텍스트 매니저를 사용하세요.

"아, 그래서 제가 만든 에이전트들이 메모리를 다 먹은 거군요." 김개발 씨가 깨달았습니다. 팀장님이 고개를 끄덕였습니다.

"맞아요. 이제 풀을 사용해서 다시 구현해보세요."

실전 팁

💡 - 풀 크기는 예상 동시 작업 수와 가용 메모리를 고려하여 설정하세요

• 에이전트 반환을 잊지 않도록 컨텍스트 매니저 패턴을 활용하세요
• 풀 사용률을 모니터링하여 적절한 크기를 찾아가세요

---

6. 분산 에이전트 시스템

프로젝트가 성공하여 사용자가 급증했습니다. 단일 서버로는 감당이 안 됩니다.

CTO님이 말했습니다. "이제 여러 서버에 에이전트를 분산 배치해야 합니다.

마치 전국에 지사를 두는 것처럼요. 본사에서 모든 지사를 조율하면서도, 각 지사가 독립적으로 운영될 수 있어야 합니다."

분산 에이전트 시스템은 여러 서버나 노드에 에이전트를 배치하여 확장성과 내결함성을 확보하는 아키텍처입니다. 중앙 코디네이터가 작업을 분배하고, 각 노드의 에이전트들이 독립적으로 작업을 수행합니다.

한 노드가 실패해도 전체 시스템은 계속 동작합니다.

다음 코드를 살펴봅시다.

from typing import Dict, List
import hashlib

class DistributedCoordinator:
    def __init__(self, node_id: str):
        self.node_id = node_id
        self._nodes: Dict[str, "NodeInfo"] = {}
        self._local_agents: Dict[str, BaseAgent] = {}

    async def register_node(self, node_info: "NodeInfo"):
        # 새 노드 등록
        self._nodes[node_info.node_id] = node_info

    async def route_task(self, task: Dict) -> str:
        # 일관된 해싱으로 작업을 적절한 노드로 라우팅
        task_key = task.get("key", str(task))
        target_node = self._consistent_hash(task_key)

        if target_node == self.node_id:
            return await self._execute_locally(task)
        else:
            return await self._forward_to_node(target_node, task)

    def _consistent_hash(self, key: str) -> str:
        # 일관된 해싱으로 노드 선택
        hash_val = int(hashlib.md5(key.encode()).hexdigest(), 16)
        node_ids = sorted(self._nodes.keys())
        return node_ids[hash_val % len(node_ids)]

    async def handle_node_failure(self, failed_node_id: str):
        # 노드 실패 처리: 작업 재분배
        if failed_node_id in self._nodes:
            pending_tasks = self._nodes[failed_node_id].pending_tasks
            del self._nodes[failed_node_id]
            # 실패한 노드의 작업을 다른 노드로 재분배
            for task in pending_tasks:
                await self.route_task(task)

"전국에 지사가 있는 큰 회사를 상상해보세요." CTO님의 설명이 시작되었습니다. "서울 본사, 부산 지사, 대전 지사, 대구 지사가 있다고 합시다.

고객이 문의하면 어디로 연결해야 할까요?" 김개발 씨가 대답했습니다. "고객 위치에 따라 가까운 지사로 연결하면 되지 않을까요?" "좋아요.

그런데 갑자기 부산 지사에 불이 나서 운영이 중단되면요?" "다른 지사에서 부산 고객도 담당해야겠네요." 바로 이것이 분산 시스템의 핵심입니다. 작업을 여러 노드에 분산하고, 일부가 실패해도 전체 시스템이 계속 동작해야 합니다.

위 코드의 DistributedCoordinator는 분산 환경의 코디네이터입니다. 각 노드에 하나씩 존재하며, 서로 협력하여 작업을 처리합니다.

register_node 메서드로 새로운 노드를 시스템에 등록합니다. 노드가 추가되면 자동으로 작업 분산 대상이 됩니다.

route_task가 핵심 메서드입니다. 작업이 들어오면 어느 노드에서 처리할지 결정합니다.

**일관된 해싱(Consistent Hashing)**을 사용하여 같은 키의 작업은 항상 같은 노드로 라우팅됩니다. 왜 일관된 해싱을 사용할까요?

단순히 노드 수로 나머지 연산을 하면, 노드가 추가되거나 제거될 때 거의 모든 작업의 배치가 바뀝니다. 일관된 해싱은 변경을 최소화합니다.

_consistent_hash 메서드를 보면, 작업 키를 해시하여 숫자로 변환하고, 노드 목록에서 해당 위치의 노드를 선택합니다. 단순화된 구현이지만 원리를 이해하기에 충분합니다.

handle_node_failure는 노드 실패를 처리합니다. 실패한 노드를 목록에서 제거하고, 그 노드가 처리하던 작업들을 다른 노드로 재분배합니다.

이것이 **내결함성(Fault Tolerance)**입니다. 실무에서는 더 복잡한 요소들이 있습니다.

서비스 디스커버리로 노드를 자동 발견하고, 헬스 체크로 노드 상태를 모니터링하고, 리더 선출로 코디네이터를 동적으로 결정합니다. 네트워크 파티션도 고려해야 합니다.

본사와 지사 사이의 통신이 끊어지면 어떻게 할까요? CAP 정리에 따르면 일관성, 가용성, 분할 내성 중 두 가지만 보장할 수 있습니다.

비즈니스 요구사항에 맞게 선택해야 합니다. "분산 시스템은 어렵지만, 규모가 커지면 피할 수 없어요." CTO님이 말했습니다.

"중요한 건 단순하게 시작해서 점차 복잡성을 추가하는 겁니다. 처음부터 완벽한 분산 시스템을 만들려고 하면 실패하기 쉬워요." 김개발 씨가 결심했습니다.

"먼저 단일 서버에서 잘 동작하게 만들고, 그다음에 분산으로 확장해봐야겠어요."

실전 팁

💡 - 분산 시스템은 단일 서버에서 충분히 검증한 후 도입하세요

• 일관된 해싱을 사용하면 노드 추가/제거 시 재분배를 최소화할 수 있습니다
• 네트워크 장애를 항상 가정하고 타임아웃, 재시도 로직을 구현하세요

---

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