Agent Communication 완벽 가이드

여러 AI 에이전트가 서로 소통하며 협력하는 방법을 배웁니다. 프로토콜, 메시지 포맷, 협업 패턴까지 실무에 필요한 모든 것을 다룹니다. 초급 개발자도 쉽게 이해할 수 있도록 실전 예제와 함께 설명합니다.

1. 에이전트 간 프로토콜

어느 날 김개발 씨가 AI 에이전트 시스템을 개발하던 중 문제에 부딪혔습니다. "에이전트 두 개가 서로 대화를 못 하네요?" 박시니어 씨가 다가와 말합니다.

"에이전트 간 프로토콜을 정의하지 않아서 그래요."

에이전트 간 프로토콜은 여러 AI 에이전트가 서로 통신할 때 지켜야 하는 규칙입니다. 마치 사람들이 대화할 때 한국어나 영어 같은 공통 언어를 사용하는 것처럼, 에이전트들도 서로 이해할 수 있는 약속된 형식이 필요합니다.

이를 통해 다양한 에이전트가 협력하여 복잡한 작업을 수행할 수 있습니다.

다음 코드를 살펴봅시다.

from enum import Enum
from dataclasses import dataclass

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

@dataclass
class AgentProtocol:
    # 프로토콜 버전 정의
    version: str = "1.0"
    # 메시지 타입 지정
    message_type: MessageType = MessageType.REQUEST

    def validate(self, message: dict) -> bool:
        # 필수 필드 검증
        required_fields = ["sender", "receiver", "content"]
        return all(field in message for field in required_fields)

김개발 씨는 입사 3개월 차 주니어 개발자입니다. 최근 회사에서 멀티 에이전트 시스템을 도입하기로 했고, 김개발 씨는 처음으로 이런 시스템을 만들어야 했습니다.

열심히 코드를 작성했지만, 에이전트 A가 보낸 메시지를 에이전트 B가 이해하지 못하는 상황이 계속 발생했습니다. 선배 개발자 박시니어 씨가 김개발 씨의 화면을 들여다보며 말합니다.

"아, 프로토콜이 없네요. 그래서 에이전트들이 서로 말을 못 알아듣는 거예요." 그렇다면 에이전트 간 프로토콜이란 정확히 무엇일까요?

쉽게 비유하자면, 프로토콜은 마치 외교 문서의 양식과 같습니다. 나라마다 언어는 다르지만, 외교 문서는 정해진 형식을 따릅니다.

제목, 발신자, 수신자, 본문, 날짜 같은 필수 항목이 있고, 이를 지켜야 상대방이 문서를 제대로 이해할 수 있습니다. 에이전트 간 프로토콜도 마찬가지로 메시지의 구조와 규칙을 정의합니다.

프로토콜이 없던 시절에는 어땠을까요? 초기 멀티 에이전트 시스템에서는 각 에이전트가 제멋대로 메시지를 보냈습니다.

어떤 에이전트는 JSON 형식으로, 어떤 에이전트는 XML 형식으로 데이터를 전송했습니다. 더 큰 문제는 필수 정보가 빠진 메시지를 받았을 때였습니다.

수신자는 메시지를 해석할 수 없어 오류가 발생하고, 전체 시스템이 멈춰버리는 일도 빈번했습니다. 바로 이런 문제를 해결하기 위해 에이전트 간 프로토콜이 등장했습니다.

프로토콜을 사용하면 모든 에이전트가 동일한 규칙으로 메시지를 주고받을 수 있습니다. 또한 메시지의 유효성을 검증하여 잘못된 데이터가 시스템에 들어오는 것을 방지할 수 있습니다.

무엇보다 새로운 에이전트를 추가할 때도 프로토콜만 따르면 되므로 확장성이 뛰어납니다. 위의 코드를 한 줄씩 살펴보겠습니다.

먼저 MessageType을 Enum으로 정의하여 메시지 타입을 명확히 구분합니다. REQUEST는 요청, RESPONSE는 응답, BROADCAST는 전체 공지를 의미합니다.

다음으로 AgentProtocol 클래스는 프로토콜의 버전과 메시지 타입을 관리합니다. validate 메서드는 메시지에 필수 필드가 모두 포함되어 있는지 검증합니다.

실제 현업에서는 어떻게 활용할까요? 예를 들어 고객 서비스 시스템을 개발한다고 가정해봅시다.

문의 접수 에이전트, 답변 생성 에이전트, 품질 검수 에이전트가 협력하여 고객 문의를 처리합니다. 이때 프로토콜을 정의하면 각 에이전트가 명확한 형식으로 정보를 전달하여 오류 없이 작업을 완수할 수 있습니다.

많은 AI 스타트업에서 이런 패턴을 적극적으로 사용하고 있습니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수 중 하나는 프로토콜을 너무 복잡하게 만드는 것입니다. 필드가 너무 많으면 유지보수가 어렵고, 에이전트 개발 시간도 늘어납니다.

따라서 처음에는 필수 필드만 정의하고, 필요에 따라 점진적으로 확장하는 것이 좋습니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

박시니어 씨의 조언에 따라 프로토콜을 정의한 김개발 씨는 에이전트 간 통신이 안정적으로 작동하는 것을 확인했습니다. "이제 에이전트들이 서로 대화를 잘하네요!" 에이전트 간 프로토콜을 제대로 이해하면 더 안정적이고 확장 가능한 멀티 에이전트 시스템을 구축할 수 있습니다.

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

실전 팁

💡 - 프로토콜 버전을 명시하여 향후 업데이트 시 호환성을 관리하세요

필수 필드는 최소화하고, 선택 필드는 확장 가능하게 설계하세요
메시지 검증 로직을 반드시 구현하여 잘못된 데이터를 조기에 차단하세요

2. 메시지 포맷

프로토콜을 정의한 김개발 씨는 이제 실제 메시지를 주고받아야 했습니다. "메시지를 어떤 형식으로 만들어야 하나요?" 박시니어 씨가 화이트보드에 그림을 그리며 설명하기 시작했습니다.

메시지 포맷은 에이전트 간 통신에서 주고받는 데이터의 구조를 정의합니다. 마치 편지에 발신자, 수신자, 제목, 본문이 있듯이, 에이전트 메시지도 정해진 구조를 따릅니다.

표준화된 포맷을 사용하면 에이전트들이 서로의 메시지를 정확히 해석하고 적절히 응답할 수 있습니다.

다음 코드를 살펴봅시다.

from datetime import datetime
from typing import Any, Optional

@dataclass
class AgentMessage:
    # 발신자 에이전트 ID
    sender: str
    # 수신자 에이전트 ID
    receiver: str
    # 메시지 내용
    content: Any
    # 메시지 타입
    msg_type: MessageType
    # 타임스탬프 (자동 생성)
    timestamp: str = None
    # 선택적 메타데이터
    metadata: Optional[dict] = None

    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.now().isoformat()

김개발 씨는 프로토콜은 만들었지만, 이제 실제로 어떤 데이터를 어떻게 담아서 보내야 할지 막막했습니다. 에이전트 A에서 에이전트 B로 작업을 요청하려면 구체적으로 어떤 정보가 필요할까요?

박시니어 씨가 화이트보드에 도식을 그리며 설명합니다. "메시지는 기본적으로 누가, 누구에게, 무엇을, 언제 보냈는지가 명확해야 해요." 메시지 포맷이란 무엇일까요?

쉽게 비유하자면, 메시지 포맷은 마치 택배 송장과 같습니다. 택배 송장에는 보내는 사람, 받는 사람, 물품 내용, 발송 날짜가 명확히 적혀 있습니다.

이 정보가 있어야 택배 기사님이 정확한 곳에 물건을 배달할 수 있습니다. 에이전트 메시지도 마찬가지로 필수 정보를 구조화하여 담아야 수신 에이전트가 올바르게 처리할 수 있습니다.

메시지 포맷이 정해지지 않았을 때는 어떤 문제가 있었을까요? 초기에는 개발자마다 자기 방식대로 메시지를 만들었습니다.

어떤 개발자는 딕셔너리에 "from"과 "to"를 사용했고, 다른 개발자는 "sender"와 "receiver"를 사용했습니다. 더 혼란스러운 것은 타임스탬프였습니다.

어떤 메시지는 Unix timestamp를, 다른 메시지는 ISO 8601 형식을 사용했습니다. 이런 불일치는 디버깅을 어렵게 만들고 버그를 양산했습니다.

바로 이런 혼란을 막기 위해 표준 메시지 포맷이 필요합니다. 표준 포맷을 사용하면 모든 에이전트가 동일한 구조로 메시지를 생성하고 파싱할 수 있습니다.

또한 타임스탬프를 자동으로 생성하여 개발자가 신경 쓸 부분을 줄여줍니다. 무엇보다 메타데이터 필드를 통해 확장 가능한 구조를 제공합니다.

위의 코드를 한 줄씩 살펴보겠습니다. AgentMessage 클래스는 dataclass로 정의되어 간결하면서도 명확합니다.

sender와 receiver는 문자열로 에이전트 ID를 나타냅니다. content는 Any 타입으로 다양한 데이터를 담을 수 있습니다.

msg_type은 앞서 정의한 MessageType Enum을 사용합니다. timestamp는 post_init 메서드에서 자동으로 현재 시간을 ISO 형식으로 생성합니다.

metadata는 Optional로 추가 정보를 담을 수 있습니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 번역 서비스 시스템을 만든다고 가정해봅시다. 텍스트 추출 에이전트가 문서에서 텍스트를 뽑아내고, 번역 에이전트에게 전달합니다.

이때 AgentMessage를 사용하면 sender는 "text_extractor", receiver는 "translator", content는 추출된 텍스트, metadata에는 원본 언어와 목표 언어 정보를 담을 수 있습니다. 명확한 포맷 덕분에 번역 에이전트는 메시지를 받자마자 필요한 정보를 정확히 추출하여 작업을 수행합니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 content 필드에 너무 많은 정보를 중첩하여 담는 것입니다.

content 안에 또 다른 딕셔너리를 여러 단계로 중첩하면 파싱이 복잡해지고 오류가 발생하기 쉽습니다. 따라서 content는 가능한 한 단순하게 유지하고, 추가 정보는 metadata에 담는 것이 좋습니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 표준 메시지 포맷을 적용했고, 에이전트 간 통신이 훨씬 명확해졌습니다.

"이제 메시지를 보고 바로 무슨 내용인지 알 수 있어요!" 메시지 포맷을 제대로 설계하면 에이전트 시스템의 안정성과 유지보수성이 크게 향상됩니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - timestamp는 항상 ISO 8601 형식을 사용하여 국제 표준을 따르세요

content는 단순하게 유지하고, 복잡한 정보는 metadata로 분리하세요
dataclass를 사용하면 코드가 간결해지고 타입 힌트를 활용할 수 있습니다

3. 협업 패턴

메시지를 주고받을 수 있게 된 김개발 씨는 이제 에이전트들이 협력하여 작업을 수행하도록 만들어야 했습니다. "에이전트들이 어떻게 협업해야 하나요?" 박시니어 씨가 웃으며 말합니다.

"협업 패턴을 배워야 할 때네요."

협업 패턴은 여러 에이전트가 함께 작업을 수행하는 방식을 정의합니다. 마스터-워커 패턴, 파이프라인 패턴, 브로드캐스트 패턴 등이 있으며, 각 패턴은 특정 상황에 적합합니다.

올바른 협업 패턴을 선택하면 효율적이고 안정적인 멀티 에이전트 시스템을 구축할 수 있습니다.

다음 코드를 살펴봅시다.

from typing import List

class MasterWorkerPattern:
    def __init__(self, master_id: str):
        self.master_id = master_id
        self.workers: List[str] = []

    def add_worker(self, worker_id: str):
        # 워커 에이전트 등록
        self.workers.append(worker_id)

    def distribute_task(self, task: str) -> List[AgentMessage]:
        # 마스터가 각 워커에게 작업 분배
        messages = []
        for worker in self.workers:
            msg = AgentMessage(
                sender=self.master_id,
                receiver=worker,
                content=task,
                msg_type=MessageType.REQUEST
            )
            messages.append(msg)
        return messages

김개발 씨는 이제 에이전트들이 메시지를 주고받을 수 있게 되었습니다. 하지만 실제 업무를 처리하려면 에이전트들이 어떤 순서로, 어떻게 협력해야 하는지 막막했습니다.

예를 들어 대용량 데이터를 처리할 때 한 에이전트가 모든 작업을 하는 것은 비효율적이었습니다. 박시니어 씨가 회의실로 김개발 씨를 데려가 설명을 시작했습니다.

"팀 프로젝트를 할 때 어떻게 일을 나누나요? 팀장이 작업을 분배하고, 팀원들이 각자 맡은 부분을 처리하고, 결과를 다시 팀장에게 보고하죠.

에이전트도 마찬가지예요." 협업 패턴이란 무엇일까요? 쉽게 비유하자면, 협업 패턴은 마치 레스토랑 주방의 운영 방식과 같습니다.

헤드 셰프가 주문을 받아 각 파트(전채, 메인, 디저트)에 작업을 분배하고, 각 파트의 셰프들이 자기 역할을 수행한 후 완성된 요리를 헤드 셰프에게 전달합니다. 이처럼 협업 패턴은 에이전트들이 어떤 역할을 맡고 어떻게 소통할지를 정의합니다.

협업 패턴이 없던 시절에는 어땠을까요? 초기 멀티 에이전트 시스템에서는 모든 에이전트가 무작위로 통신했습니다.

에이전트 A가 에이전트 B, C, D에게 동시에 메시지를 보내고, 각 에이전트는 또 다른 에이전트들에게 메시지를 전달했습니다. 결과적으로 메시지가 폭증하고, 누가 무엇을 처리해야 하는지 불분명했습니다.

작업이 중복되거나 누락되는 일이 빈번했습니다. 바로 이런 혼란을 막기 위해 협업 패턴이 등장했습니다.

협업 패턴을 사용하면 각 에이전트의 역할과 책임이 명확해집니다. 또한 작업 흐름이 예측 가능하여 디버깅과 모니터링이 쉬워집니다.

무엇보다 시스템 확장 시 패턴만 따르면 되므로 새로운 에이전트를 추가하기가 간편합니다. 위의 코드를 한 줄씩 살펴보겠습니다.

MasterWorkerPattern 클래스는 마스터-워커 패턴을 구현합니다. master_id는 중앙 에이전트의 ID이고, workers는 워커 에이전트 목록입니다.

add_worker 메서드로 워커를 등록하고, distribute_task 메서드는 마스터가 모든 워커에게 동일한 작업을 분배합니다. 각 워커에게 AgentMessage를 생성하여 반환하므로, 호출자는 이 메시지들을 실제로 전송하면 됩니다.

실제 현업에서는 어떻게 활용할까요? 예를 들어 이미지 처리 서비스를 개발한다고 가정해봅시다.

사용자가 100장의 이미지를 업로드하면, 마스터 에이전트가 이미지를 10장씩 10개의 워커 에이전트에게 분배합니다. 각 워커는 할당받은 이미지를 처리하고 결과를 마스터에게 반환합니다.

마스터는 모든 결과를 수집하여 사용자에게 전달합니다. 이런 패턴은 처리 속도를 10배 가까이 높일 수 있습니다.

다른 패턴도 알아봅시다. 파이프라인 패턴은 데이터가 순차적으로 여러 에이전트를 거치는 구조입니다.

예를 들어 음성 인식 시스템에서 음성 파일 → 노이즈 제거 에이전트 → 텍스트 변환 에이전트 → 문법 교정 에이전트 순으로 처리됩니다. 브로드캐스트 패턴은 한 에이전트가 모든 에이전트에게 동시에 메시지를 보내는 구조로, 시스템 전체에 공지사항을 전달할 때 유용합니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 모든 상황에 마스터-워커 패턴을 적용하는 것입니다.

순차 처리가 필요한 작업에는 파이프라인 패턴이 더 적합하고, 단순 공지에는 브로드캐스트 패턴이 효율적입니다. 따라서 작업의 특성을 파악하고 적절한 패턴을 선택해야 합니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 데이터 처리 작업에 마스터-워커 패턴을 적용했고, 처리 속도가 크게 향상되었습니다.

"패턴을 사용하니 코드도 깔끔하고 성능도 좋아졌어요!" 협업 패턴을 제대로 이해하면 효율적이고 확장 가능한 멀티 에이전트 시스템을 설계할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - 작업 특성에 맞는 패턴을 선택하세요 (병렬 처리는 마스터-워커, 순차 처리는 파이프라인)

마스터 에이전트가 병목이 되지 않도록 워커 수를 적절히 조절하세요
각 패턴의 장단점을 이해하고, 상황에 맞게 조합하여 사용하세요

4. 실습 에이전트 통신 시스템

이론을 배운 김개발 씨는 이제 실제로 에이전트 통신 시스템을 구축해야 했습니다. "직접 만들어보면서 배우는 게 제일 빠르죠." 박시니어 씨가 새 파일을 열며 함께 코딩을 시작했습니다.

에이전트 통신 시스템은 프로토콜, 메시지 포맷, 메시지 큐를 종합하여 실제로 동작하는 시스템을 구현합니다. 에이전트가 메시지를 보내고, 받고, 처리하는 전체 흐름을 실습합니다.

이를 통해 멀티 에이전트 시스템의 핵심 메커니즘을 체득할 수 있습니다.

다음 코드를 살펴봅시다.

from queue import Queue
from threading import Thread

class CommunicationSystem:
    def __init__(self):
        # 각 에이전트별 메시지 큐
        self.message_queues = {}
        self.protocol = AgentProtocol()

    def register_agent(self, agent_id: str):
        # 에이전트 등록 및 큐 생성
        self.message_queues[agent_id] = Queue()

    def send_message(self, message: AgentMessage):
        # 메시지 유효성 검증 후 전송
        msg_dict = message.__dict__
        if self.protocol.validate(msg_dict):
            self.message_queues[message.receiver].put(message)

    def receive_message(self, agent_id: str) -> AgentMessage:
        # 해당 에이전트의 큐에서 메시지 수신
        return self.message_queues[agent_id].get()

김개발 씨는 지금까지 배운 프로토콜, 메시지 포맷, 협업 패턴을 모두 이해했습니다. 하지만 이것들을 실제로 어떻게 조합하여 동작하는 시스템을 만들어야 할지 막막했습니다.

마치 재료는 다 준비했지만 요리 순서를 모르는 것과 같았습니다. 박시니어 씨가 화면을 공유하며 함께 코딩을 시작했습니다.

"자, 이제 실제로 에이전트들이 통신할 수 있는 시스템을 만들어볼까요?" 에이전트 통신 시스템이란 무엇일까요? 쉽게 비유하자면, 통신 시스템은 마치 우체국과 같습니다.

각 집(에이전트)마다 우편함이 있고, 우체국은 편지(메시지)를 올바른 우편함에 배달합니다. 보내는 사람이 주소를 잘못 썼다면 우체국은 반송하고, 주소가 정확하면 해당 우편함에 편지를 넣어줍니다.

에이전트 통신 시스템도 마찬가지로 메시지를 검증하고, 올바른 수신자에게 전달하는 역할을 합니다. 통신 시스템이 없던 시절에는 어땠을까요?

초기에는 에이전트들이 직접 서로를 호출했습니다. 에이전트 A가 에이전트 B의 메서드를 직접 호출하는 방식이었습니다.

이렇게 하면 두 에이전트가 강하게 결합되어 하나를 수정하면 다른 하나도 수정해야 했습니다. 더 큰 문제는 에이전트가 동시에 여러 메시지를 받을 때였습니다.

메시지가 뒤섞이거나 처리 순서가 꼬여서 예상치 못한 버그가 발생했습니다. 바로 이런 문제를 해결하기 위해 중앙 집중식 통신 시스템이 등장했습니다.

통신 시스템을 사용하면 에이전트들이 직접 연결되지 않고 시스템을 통해 통신합니다. 또한 메시지 큐를 사용하여 각 에이전트가 자기 속도에 맞춰 메시지를 처리할 수 있습니다.

무엇보다 메시지 검증을 중앙에서 수행하여 잘못된 메시지가 시스템에 유입되는 것을 방지합니다. 위의 코드를 한 줄씩 살펴보겠습니다.

CommunicationSystem 클래스는 전체 통신을 관리합니다. message_queues는 딕셔너리로 각 에이전트 ID를 키로, Queue 객체를 값으로 저장합니다.

register_agent 메서드는 새 에이전트를 시스템에 등록하고 전용 큐를 생성합니다. send_message 메서드는 메시지를 AgentMessage 객체에서 딕셔너리로 변환하고 프로토콜 검증을 통과하면 수신자의 큐에 메시지를 추가합니다.

receive_message 메서드는 해당 에이전트의 큐에서 메시지를 꺼내 반환합니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 실시간 주문 처리 시스템을 만든다고 가정해봅시다. 주문 접수 에이전트, 재고 확인 에이전트, 결제 처리 에이전트, 배송 준비 에이전트가 있습니다.

CommunicationSystem을 사용하면 각 에이전트를 등록하고, 주문 정보가 메시지로 각 에이전트를 거치며 처리됩니다. 한 에이전트가 느려져도 큐에 메시지가 쌓이기만 할 뿐 전체 시스템이 멈추지 않습니다.

메시지 큐의 장점을 더 살펴봅시다. Queue는 스레드 안전하므로 여러 에이전트가 동시에 메시지를 보내도 문제없이 처리됩니다.

또한 FIFO(First In First Out) 방식으로 메시지 순서가 보장됩니다. 에이전트가 바쁠 때는 큐에 메시지가 쌓였다가, 여유가 생기면 순서대로 처리됩니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 메시지 큐의 크기를 제한하지 않는 것입니다.

수신 에이전트가 처리 속도를 따라가지 못하면 큐가 무한정 커져서 메모리가 고갈될 수 있습니다. 따라서 Queue 생성 시 maxsize를 설정하여 큐 크기를 제한하고, 큐가 가득 찬 경우의 처리 로직을 구현해야 합니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨와 함께 통신 시스템을 구현한 김개발 씨는 실제로 에이전트들이 메시지를 주고받으며 작업을 처리하는 것을 확인했습니다.

"이제 전체 구조가 이해됐어요!" 에이전트 통신 시스템을 직접 구현해보면 멀티 에이전트 시스템의 핵심을 체득할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - Queue의 maxsize를 설정하여 메모리 고갈을 방지하세요

메시지 검증은 send_message에서 수행하여 잘못된 메시지를 조기에 차단하세요
프로덕션 환경에서는 Redis나 RabbitMQ 같은 메시지 브로커를 사용하는 것이 좋습니다

5. 실습 협업 태스크 수행

통신 시스템을 완성한 김개발 씨는 이제 실제로 여러 에이전트가 협업하여 복잡한 작업을 수행하도록 만들어야 했습니다. "이제 진짜 일을 시켜볼까요?" 박시니어 씨가 실전 예제를 보여주기 시작했습니다.

협업 태스크 수행은 여러 에이전트가 통신 시스템을 통해 메시지를 주고받으며 하나의 큰 작업을 완수하는 것입니다. 마스터 에이전트가 작업을 분배하고, 워커 에이전트들이 각자 처리한 후 결과를 반환하는 전체 흐름을 실습합니다.

이를 통해 실무에서 사용할 수 있는 완전한 멀티 에이전트 시스템을 구축할 수 있습니다.

다음 코드를 살펴봅시다.

class WorkerAgent:
    def __init__(self, agent_id: str, comm_system: CommunicationSystem):
        self.agent_id = agent_id
        self.comm_system = comm_system
        comm_system.register_agent(agent_id)

    def process_task(self, task: str) -> str:
        # 실제 작업 수행 (여기서는 단순화)
        return f"Processed: {task}"

    def run(self):
        # 메시지 대기 및 처리
        while True:
            msg = self.comm_system.receive_message(self.agent_id)
            result = self.process_task(msg.content)
            # 결과를 마스터에게 응답
            response = AgentMessage(
                sender=self.agent_id,
                receiver=msg.sender,
                content=result,
                msg_type=MessageType.RESPONSE
            )
            self.comm_system.send_message(response)

김개발 씨는 통신 시스템까지 만들었지만, 이제 실제로 에이전트들이 일을 하도록 만들어야 했습니다. 이론과 시스템은 준비되었지만, 구체적으로 어떻게 작업을 분배하고 결과를 수집해야 할지 고민되었습니다.

박시니어 씨가 실전 예제를 보여주며 설명했습니다. "지금부터 텍스트 분석 시스템을 만들어볼 거예요.

마스터가 긴 문서를 여러 조각으로 나누고, 워커들이 각 조각을 분석한 후 결과를 마스터에게 돌려주는 시스템입니다." 협업 태스크 수행이란 무엇일까요? 쉽게 비유하자면, 협업 태스크는 마치 대형 건설 프로젝트와 같습니다.

프로젝트 매니저(마스터)가 전체 건물 설계를 받아 각 층을 다른 건설팀(워커)에게 할당합니다. 1층은 A팀, 2층은 B팀, 3층은 C팀이 동시에 작업합니다.

각 팀이 자기 층을 완성하면 매니저에게 보고하고, 매니저는 모든 층이 완성되었는지 확인합니다. 에이전트 협업도 이와 동일한 구조입니다.

협업 시스템이 없던 시절에는 어떤 문제가 있었을까요? 초기에는 하나의 에이전트가 모든 작업을 순차적으로 처리했습니다.

100개의 파일을 분석해야 한다면 첫 번째 파일부터 마지막 파일까지 차례로 처리했습니다. 시간이 오래 걸렸고, 에이전트 하나가 고장 나면 전체 작업이 중단되었습니다.

자원 활용도 비효율적이었습니다. CPU 코어가 8개인 서버에서 1개만 사용하는 셈이었습니다.

바로 이런 비효율을 개선하기 위해 협업 태스크 수행이 등장했습니다. 협업 태스크를 사용하면 작업을 병렬로 처리하여 시간을 단축할 수 있습니다.

또한 한 워커가 실패해도 다른 워커는 계속 작업을 수행하므로 안정성이 높아집니다. 무엇보다 서버의 모든 리소스를 효율적으로 활용할 수 있습니다.

위의 코드를 한 줄씩 살펴보겠습니다. WorkerAgent 클래스는 실제로 작업을 수행하는 에이전트입니다.

__init__에서 자신을 통신 시스템에 등록합니다. process_task 메서드는 실제 작업 로직을 담고 있으며, 여기서는 단순히 "Processed: " 문자열을 붙여 반환합니다.

run 메서드는 무한 루프로 메시지를 대기하고, 메시지가 도착하면 작업을 처리한 후 결과를 응답 메시지로 발신자에게 전송합니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 이메일 스팸 필터링 시스템을 만든다고 가정해봅시다. 하루에 수백만 개의 이메일이 들어옵니다.

마스터 에이전트가 이메일을 100개씩 묶어 10개의 워커 에이전트에게 분배합니다. 각 워커는 할당받은 이메일을 분석하여 스팸 여부를 판단하고 결과를 마스터에게 반환합니다.

마스터는 모든 결과를 데이터베이스에 저장합니다. 이렇게 하면 단일 에이전트로 처리할 때보다 10배 빠르게 작업을 완료할 수 있습니다.

마스터 에이전트는 어떻게 구현할까요? 마스터는 전체 작업을 작은 단위로 나누고, 각 워커에게 분배합니다.

워커들로부터 응답을 받으면 결과를 수집하고, 모든 작업이 완료되었는지 확인합니다. 필요하다면 실패한 작업을 재시도하거나 다른 워커에게 재할당할 수도 있습니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 워커의 run 메서드에 예외 처리를 하지 않는 것입니다.

작업 중 오류가 발생하면 워커 전체가 멈춰버립니다. 따라서 try-except 블록으로 감싸고, 오류가 발생하면 오류 메시지를 마스터에게 전송하여 마스터가 적절히 대응할 수 있도록 해야 합니다.

또 다른 주의사항은 무한 루프 종료 조건입니다. 위 코드의 run 메서드는 while True로 영원히 실행됩니다.

실제 시스템에서는 종료 신호를 받으면 루프를 빠져나오도록 구현해야 합니다. 예를 들어 특별한 타입의 메시지(SHUTDOWN)를 받으면 break하는 방식입니다.

스레딩을 활용하면 더 효과적입니다. 각 워커를 별도 스레드에서 실행하면 여러 워커가 동시에 작동합니다.

Thread 객체를 생성하여 target으로 worker.run을 지정하고 start() 메서드를 호출하면 됩니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

박시니어 씨와 함께 마스터와 워커를 구현한 김개발 씨는 실제로 텍스트 분석 작업이 병렬로 처리되는 것을 확인했습니다. "와, 정말 10배 빨라졌어요!" 협업 태스크 수행을 직접 구현해보면 멀티 에이전트 시스템의 진가를 체험할 수 있습니다.

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

실전 팁

💡 - 워커의 run 메서드에 예외 처리를 반드시 추가하여 안정성을 확보하세요

무한 루프에 종료 조건을 구현하여 우아한 종료(graceful shutdown)를 지원하세요
Thread를 사용하여 워커를 병렬 실행하면 성능이 크게 향상됩니다

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

#Python#Agent#Communication#Protocol#MultiAgent#LLM,통신,MultiAgent

Agent Communication 완벽 가이드

# Agent Communication 완벽 가이드 여러 AI 에이전트가 서로 소통하며 협력하는 방법을 배웁니다. 프로토콜, 메시지 포맷, 협업 패턴까지 실무에 필요한 모든 것을 다룹니다. 초급 개발자도 쉽게 이해할 수 있도록 실전 예제와 함께 설명합니다. --- ## 목차 1. [에이전트_간_프로토콜](#에이전트-간-프로토콜) 2. [메시지_포맷](#메시지-포맷) 3. [협업_패턴](#협업-패턴) 4. [실습_에이전트_통신_시스템](#실습-에이전트-통신-시스템) 5. [실습_협업_태스크_수행](#실습-협업-태스크-수행) --- ## 1. 에이전트_간_프로토콜 어느 날 김개발 씨가 AI 에이전트 시스템을 개발하던 중 문제에 부딪혔습니다. "에이전트 두 개가 서로 대화를 못 하네요?" 박시니어 씨가 다가와 말합니다. "에이전트 간 프로토콜을 정의하지 않아서 그래요." **에이전트 간 프로토콜**은 여러 AI 에이전트가 서로 통신할 때 지켜야 하는 규칙입니다. 마치 사람들이 대화할 때 한국어나 영어 같은 공통 언어를 사용하는 것처럼, 에이전트들도 서로 이해할 수 있는 약속된 형식이 필요합니다. 이를 통해 다양한 에이전트가 협력하여 복잡한 작업을 수행할 수 있습니다. 다음 코드를 살펴봅시다. ```python from enum import Enum from dataclasses import dataclass class MessageType(Enum): REQUEST = "request" RESPONSE = "response" BROADCAST = "broadcast" @dataclass class AgentProtocol: # 프로토콜 버전 정의 version: str = "1.0" # 메시지 타입 지정 message_type: MessageType = MessageType.REQUEST def validate(self, message: dict) -> bool: # 필수 필드 검증 required_fields = ["sender", "receiver", "content"] return all(field in message for field in required_fields) ``` 김개발 씨는 입사 3개월 차 주니어 개발자입니다. 최근 회사에서 멀티 에이전트 시스템을 도입하기로 했고, 김개발 씨는 처음으로 이런 시스템을 만들어야 했습니다. 열심히 코드를 작성했지만, 에이전트 A가 보낸 메시지를 에이전트 B가 이해하지 못하는 상황이 계속 발생했습니다. 선배 개발자 박시니어 씨가 김개발 씨의 화면을 들여다보며 말합니다. "아, 프로토콜이 없네요. 그래서 에이전트들이 서로 말을 못 알아듣는 거예요." 그렇다면 **에이전트 간 프로토콜**이란 정확히 무엇일까요? 쉽게 비유하자면, 프로토콜은 마치 외교 문서의 양식과 같습니다. 나라마다 언어는 다르지만, 외교 문서는 정해진 형식을 따릅니다. 제목, 발신자, 수신자, 본문, 날짜 같은 필수 항목이 있고, 이를 지켜야 상대방이 문서를 제대로 이해할 수 있습니다. 에이전트 간 프로토콜도 마찬가지로 메시지의 구조와 규칙을 정의합니다. 프로토콜이 없던 시절에는 어땠을까요? 초기 멀티 에이전트 시스템에서는 각 에이전트가 제멋대로 메시지를 보냈습니다. 어떤 에이전트는 JSON 형식으로, 어떤 에이전트는 XML 형식으로 데이터를 전송했습니다. 더 큰 문제는 필수 정보가 빠진 메시지를 받았을 때였습니다. 수신자는 메시지를 해석할 수 없어 오류가 발생하고, 전체 시스템이 멈춰버리는 일도 빈번했습니다. 바로 이런 문제를 해결하기 위해 **에이전트 간 프로토콜**이 등장했습니다. 프로토콜을 사용하면 모든 에이전트가 동일한 규칙으로 메시지를 주고받을 수 있습니다. 또한 메시지의 유효성을 검증하여 잘못된 데이터가 시스템에 들어오는 것을 방지할 수 있습니다. 무엇보다 새로운 에이전트를 추가할 때도 프로토콜만 따르면 되므로 확장성이 뛰어납니다. 위의 코드를 한 줄씩 살펴보겠습니다. 먼저 MessageType을 Enum으로 정의하여 메시지 타입을 명확히 구분합니다. REQUEST는 요청, RESPONSE는 응답, BROADCAST는 전체 공지를 의미합니다. 다음으로 AgentProtocol 클래스는 프로토콜의 버전과 메시지 타입을 관리합니다. validate 메서드는 메시지에 필수 필드가 모두 포함되어 있는지 검증합니다. 실제 현업에서는 어떻게 활용할까요? 예를 들어 고객 서비스 시스템을 개발한다고 가정해봅시다. 문의 접수 에이전트, 답변 생성 에이전트, 품질 검수 에이전트가 협력하여 고객 문의를 처리합니다. 이때 프로토콜을 정의하면 각 에이전트가 명확한 형식으로 정보를 전달하여 오류 없이 작업을 완수할 수 있습니다. 많은 AI 스타트업에서 이런 패턴을 적극적으로 사용하고 있습니다. 하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 프로토콜을 너무 복잡하게 만드는 것입니다. 필드가 너무 많으면 유지보수가 어렵고, 에이전트 개발 시간도 늘어납니다. 따라서 처음에는 필수 필드만 정의하고, 필요에 따라 점진적으로 확장하는 것이 좋습니다. 다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 조언에 따라 프로토콜을 정의한 김개발 씨는 에이전트 간 통신이 안정적으로 작동하는 것을 확인했습니다. "이제 에이전트들이 서로 대화를 잘하네요!" 에이전트 간 프로토콜을 제대로 이해하면 더 안정적이고 확장 가능한 멀티 에이전트 시스템을 구축할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요. **실전 팁** 💡 - 프로토콜 버전을 명시하여 향후 업데이트 시 호환성을 관리하세요 - 필수 필드는 최소화하고, 선택 필드는 확장 가능하게 설계하세요 - 메시지 검증 로직을 반드시 구현하여 잘못된 데이터를 조기에 차단하세요 --- ## 2. 메시지_포맷 프로토콜을 정의한 김개발 씨는 이제 실제 메시지를 주고받아야 했습니다. "메시지를 어떤 형식으로 만들어야 하나요?" 박시니어 씨가 화이트보드에 그림을 그리며 설명하기 시작했습니다. **메시지 포맷**은 에이전트 간 통신에서 주고받는 데이터의 구조를 정의합니다. 마치 편지에 발신자, 수신자, 제목, 본문이 있듯이, 에이전트 메시지도 정해진 구조를 따릅니다. 표준화된 포맷을 사용하면 에이전트들이 서로의 메시지를 정확히 해석하고 적절히 응답할 수 있습니다. 다음 코드를 살펴봅시다. ```python from datetime import datetime from typing import Any, Optional @dataclass class AgentMessage: # 발신자 에이전트 ID sender: str # 수신자 에이전트 ID receiver: str # 메시지 내용 content: Any # 메시지 타입 msg_type: MessageType # 타임스탬프 (자동 생성) timestamp: str = None # 선택적 메타데이터 metadata: Optional[dict] = None def __post_init__(self): if self.timestamp is None: self.timestamp = datetime.now().isoformat() ``` 김개발 씨는 프로토콜은 만들었지만, 이제 실제로 어떤 데이터를 어떻게 담아서 보내야 할지 막막했습니다. 에이전트 A에서 에이전트 B로 작업을 요청하려면 구체적으로 어떤 정보가 필요할까요? 박시니어 씨가 화이트보드에 도식을 그리며 설명합니다. "메시지는 기본적으로 누가, 누구에게, 무엇을, 언제 보냈는지가 명확해야 해요." **메시지 포맷**이란 무엇일까요? 쉽게 비유하자면, 메시지 포맷은 마치 택배 송장과 같습니다. 택배 송장에는 보내는 사람, 받는 사람, 물품 내용, 발송 날짜가 명확히 적혀 있습니다. 이 정보가 있어야 택배 기사님이 정확한 곳에 물건을 배달할 수 있습니다. 에이전트 메시지도 마찬가지로 필수 정보를 구조화하여 담아야 수신 에이전트가 올바르게 처리할 수 있습니다. 메시지 포맷이 정해지지 않았을 때는 어떤 문제가 있었을까요? 초기에는 개발자마다 자기 방식대로 메시지를 만들었습니다. 어떤 개발자는 딕셔너리에 "from"과 "to"를 사용했고, 다른 개발자는 "sender"와 "receiver"를 사용했습니다. 더 혼란스러운 것은 타임스탬프였습니다. 어떤 메시지는 Unix timestamp를, 다른 메시지는 ISO 8601 형식을 사용했습니다. 이런 불일치는 디버깅을 어렵게 만들고 버그를 양산했습니다. 바로 이런 혼란을 막기 위해 **표준 메시지 포맷**이 필요합니다. 표준 포맷을 사용하면 모든 에이전트가 동일한 구조로 메시지를 생성하고 파싱할 수 있습니다. 또한 타임스탬프를 자동으로 생성하여 개발자가 신경 쓸 부분을 줄여줍니다. 무엇보다 메타데이터 필드를 통해 확장 가능한 구조를 제공합니다. 위의 코드를 한 줄씩 살펴보겠습니다. AgentMessage 클래스는 dataclass로 정의되어 간결하면서도 명확합니다. sender와 receiver는 문자열로 에이전트 ID를 나타냅니다. content는 Any 타입으로 다양한 데이터를 담을 수 있습니다. msg_type은 앞서 정의한 MessageType Enum을 사용합니다. timestamp는 __post_init__ 메서드에서 자동으로 현재 시간을 ISO 형식으로 생성합니다. metadata는 Optional로 추가 정보를 담을 수 있습니다. 실제 현업에서는 어떻게 활용할까요? 예를 들어 번역 서비스 시스템을 만든다고 가정해봅시다. 텍스트 추출 에이전트가 문서에서 텍스트를 뽑아내고, 번역 에이전트에게 전달합니다. 이때 AgentMessage를 사용하면 sender는 "text_extractor", receiver는 "translator", content는 추출된 텍스트, metadata에는 원본 언어와 목표 언어 정보를 담을 수 있습니다. 명확한 포맷 덕분에 번역 에이전트는 메시지를 받자마자 필요한 정보를 정확히 추출하여 작업을 수행합니다. 하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 content 필드에 너무 많은 정보를 중첩하여 담는 것입니다. content 안에 또 다른 딕셔너리를 여러 단계로 중첩하면 파싱이 복잡해지고 오류가 발생하기 쉽습니다. 따라서 content는 가능한 한 단순하게 유지하고, 추가 정보는 metadata에 담는 것이 좋습니다. 다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 표준 메시지 포맷을 적용했고, 에이전트 간 통신이 훨씬 명확해졌습니다. "이제 메시지를 보고 바로 무슨 내용인지 알 수 있어요!" 메시지 포맷을 제대로 설계하면 에이전트 시스템의 안정성과 유지보수성이 크게 향상됩니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요. **실전 팁** 💡 - timestamp는 항상 ISO 8601 형식을 사용하여 국제 표준을 따르세요 - content는 단순하게 유지하고, 복잡한 정보는 metadata로 분리하세요 - dataclass를 사용하면 코드가 간결해지고 타입 힌트를 활용할 수 있습니다 --- ## 3. 협업_패턴 메시지를 주고받을 수 있게 된 김개발 씨는 이제 에이전트들이 협력하여 작업을 수행하도록 만들어야 했습니다. "에이전트들이 어떻게 협업해야 하나요?" 박시니어 씨가 웃으며 말합니다. "협업 패턴을 배워야 할 때네요." **협업 패턴**은 여러 에이전트가 함께 작업을 수행하는 방식을 정의합니다. 마스터-워커 패턴, 파이프라인 패턴, 브로드캐스트 패턴 등이 있으며, 각 패턴은 특정 상황에 적합합니다. 올바른 협업 패턴을 선택하면 효율적이고 안정적인 멀티 에이전트 시스템을 구축할 수 있습니다. 다음 코드를 살펴봅시다. ```python from typing import List class MasterWorkerPattern: def __init__(self, master_id: str): self.master_id = master_id self.workers: List[str] = [] def add_worker(self, worker_id: str): # 워커 에이전트 등록 self.workers.append(worker_id) def distribute_task(self, task: str) -> List[AgentMessage]: # 마스터가 각 워커에게 작업 분배 messages = [] for worker in self.workers: msg = AgentMessage( sender=self.master_id, receiver=worker, content=task, msg_type=MessageType.REQUEST ) messages.append(msg) return messages ``` 김개발 씨는 이제 에이전트들이 메시지를 주고받을 수 있게 되었습니다. 하지만 실제 업무를 처리하려면 에이전트들이 어떤 순서로, 어떻게 협력해야 하는지 막막했습니다. 예를 들어 대용량 데이터를 처리할 때 한 에이전트가 모든 작업을 하는 것은 비효율적이었습니다. 박시니어 씨가 회의실로 김개발 씨를 데려가 설명을 시작했습니다. "팀 프로젝트를 할 때 어떻게 일을 나누나요? 팀장이 작업을 분배하고, 팀원들이 각자 맡은 부분을 처리하고, 결과를 다시 팀장에게 보고하죠. 에이전트도 마찬가지예요." **협업 패턴**이란 무엇일까요? 쉽게 비유하자면, 협업 패턴은 마치 레스토랑 주방의 운영 방식과 같습니다. 헤드 셰프가 주문을 받아 각 파트(전채, 메인, 디저트)에 작업을 분배하고, 각 파트의 셰프들이 자기 역할을 수행한 후 완성된 요리를 헤드 셰프에게 전달합니다. 이처럼 협업 패턴은 에이전트들이 어떤 역할을 맡고 어떻게 소통할지를 정의합니다. 협업 패턴이 없던 시절에는 어땠을까요? 초기 멀티 에이전트 시스템에서는 모든 에이전트가 무작위로 통신했습니다. 에이전트 A가 에이전트 B, C, D에게 동시에 메시지를 보내고, 각 에이전트는 또 다른 에이전트들에게 메시지를 전달했습니다. 결과적으로 메시지가 폭증하고, 누가 무엇을 처리해야 하는지 불분명했습니다. 작업이 중복되거나 누락되는 일이 빈번했습니다. 바로 이런 혼란을 막기 위해 **협업 패턴**이 등장했습니다. 협업 패턴을 사용하면 각 에이전트의 역할과 책임이 명확해집니다. 또한 작업 흐름이 예측 가능하여 디버깅과 모니터링이 쉬워집니다. 무엇보다 시스템 확장 시 패턴만 따르면 되므로 새로운 에이전트를 추가하기가 간편합니다. 위의 코드를 한 줄씩 살펴보겠습니다. MasterWorkerPattern 클래스는 마스터-워커 패턴을 구현합니다. master_id는 중앙 에이전트의 ID이고, workers는 워커 에이전트 목록입니다. add_worker 메서드로 워커를 등록하고, distribute_task 메서드는 마스터가 모든 워커에게 동일한 작업을 분배합니다. 각 워커에게 AgentMessage를 생성하여 반환하므로, 호출자는 이 메시지들을 실제로 전송하면 됩니다. 실제 현업에서는 어떻게 활용할까요? 예를 들어 이미지 처리 서비스를 개발한다고 가정해봅시다. 사용자가 100장의 이미지를 업로드하면, 마스터 에이전트가 이미지를 10장씩 10개의 워커 에이전트에게 분배합니다. 각 워커는 할당받은 이미지를 처리하고 결과를 마스터에게 반환합니다. 마스터는 모든 결과를 수집하여 사용자에게 전달합니다. 이런 패턴은 처리 속도를 10배 가까이 높일 수 있습니다. 다른 패턴도 알아봅시다. **파이프라인 패턴**은 데이터가 순차적으로 여러 에이전트를 거치는 구조입니다. 예를 들어 음성 인식 시스템에서 음성 파일 → 노이즈 제거 에이전트 → 텍스트 변환 에이전트 → 문법 교정 에이전트 순으로 처리됩니다. **브로드캐스트 패턴**은 한 에이전트가 모든 에이전트에게 동시에 메시지를 보내는 구조로, 시스템 전체에 공지사항을 전달할 때 유용합니다. 하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 모든 상황에 마스터-워커 패턴을 적용하는 것입니다. 순차 처리가 필요한 작업에는 파이프라인 패턴이 더 적합하고, 단순 공지에는 브로드캐스트 패턴이 효율적입니다. 따라서 작업의 특성을 파악하고 적절한 패턴을 선택해야 합니다. 다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 데이터 처리 작업에 마스터-워커 패턴을 적용했고, 처리 속도가 크게 향상되었습니다. "패턴을 사용하니 코드도 깔끔하고 성능도 좋아졌어요!" 협업 패턴을 제대로 이해하면 효율적이고 확장 가능한 멀티 에이전트 시스템을 설계할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요. **실전 팁** 💡 - 작업 특성에 맞는 패턴을 선택하세요 (병렬 처리는 마스터-워커, 순차 처리는 파이프라인) - 마스터 에이전트가 병목이 되지 않도록 워커 수를 적절히 조절하세요 - 각 패턴의 장단점을 이해하고, 상황에 맞게 조합하여 사용하세요 --- ## 4. 실습_에이전트_통신_시스템 이론을 배운 김개발 씨는 이제 실제로 에이전트 통신 시스템을 구축해야 했습니다. "직접 만들어보면서 배우는 게 제일 빠르죠." 박시니어 씨가 새 파일을 열며 함께 코딩을 시작했습니다. **에이전트 통신 시스템**은 프로토콜, 메시지 포맷, 메시지 큐를 종합하여 실제로 동작하는 시스템을 구현합니다. 에이전트가 메시지를 보내고, 받고, 처리하는 전체 흐름을 실습합니다. 이를 통해 멀티 에이전트 시스템의 핵심 메커니즘을 체득할 수 있습니다. 다음 코드를 살펴봅시다. ```python from queue import Queue from threading import Thread class CommunicationSystem: def __init__(self): # 각 에이전트별 메시지 큐 self.message_queues = {} self.protocol = AgentProtocol() def register_agent(self, agent_id: str): # 에이전트 등록 및 큐 생성 self.message_queues[agent_id] = Queue() def send_message(self, message: AgentMessage): # 메시지 유효성 검증 후 전송 msg_dict = message.__dict__ if self.protocol.validate(msg_dict): self.message_queues[message.receiver].put(message) def receive_message(self, agent_id: str) -> AgentMessage: # 해당 에이전트의 큐에서 메시지 수신 return self.message_queues[agent_id].get() ``` 김개발 씨는 지금까지 배운 프로토콜, 메시지 포맷, 협업 패턴을 모두 이해했습니다. 하지만 이것들을 실제로 어떻게 조합하여 동작하는 시스템을 만들어야 할지 막막했습니다. 마치 재료는 다 준비했지만 요리 순서를 모르는 것과 같았습니다. 박시니어 씨가 화면을 공유하며 함께 코딩을 시작했습니다. "자, 이제 실제로 에이전트들이 통신할 수 있는 시스템을 만들어볼까요?" **에이전트 통신 시스템**이란 무엇일까요? 쉽게 비유하자면, 통신 시스템은 마치 우체국과 같습니다. 각 집(에이전트)마다 우편함이 있고, 우체국은 편지(메시지)를 올바른 우편함에 배달합니다. 보내는 사람이 주소를 잘못 썼다면 우체국은 반송하고, 주소가 정확하면 해당 우편함에 편지를 넣어줍니다. 에이전트 통신 시스템도 마찬가지로 메시지를 검증하고, 올바른 수신자에게 전달하는 역할을 합니다. 통신 시스템이 없던 시절에는 어땠을까요? 초기에는 에이전트들이 직접 서로를 호출했습니다. 에이전트 A가 에이전트 B의 메서드를 직접 호출하는 방식이었습니다. 이렇게 하면 두 에이전트가 강하게 결합되어 하나를 수정하면 다른 하나도 수정해야 했습니다. 더 큰 문제는 에이전트가 동시에 여러 메시지를 받을 때였습니다. 메시지가 뒤섞이거나 처리 순서가 꼬여서 예상치 못한 버그가 발생했습니다. 바로 이런 문제를 해결하기 위해 **중앙 집중식 통신 시스템**이 등장했습니다. 통신 시스템을 사용하면 에이전트들이 직접 연결되지 않고 시스템을 통해 통신합니다. 또한 메시지 큐를 사용하여 각 에이전트가 자기 속도에 맞춰 메시지를 처리할 수 있습니다. 무엇보다 메시지 검증을 중앙에서 수행하여 잘못된 메시지가 시스템에 유입되는 것을 방지합니다. 위의 코드를 한 줄씩 살펴보겠습니다. CommunicationSystem 클래스는 전체 통신을 관리합니다. message_queues는 딕셔너리로 각 에이전트 ID를 키로, Queue 객체를 값으로 저장합니다. register_agent 메서드는 새 에이전트를 시스템에 등록하고 전용 큐를 생성합니다. send_message 메서드는 메시지를 AgentMessage 객체에서 딕셔너리로 변환하고 프로토콜 검증을 통과하면 수신자의 큐에 메시지를 추가합니다. receive_message 메서드는 해당 에이전트의 큐에서 메시지를 꺼내 반환합니다. 실제 현업에서는 어떻게 활용할까요? 예를 들어 실시간 주문 처리 시스템을 만든다고 가정해봅시다. 주문 접수 에이전트, 재고 확인 에이전트, 결제 처리 에이전트, 배송 준비 에이전트가 있습니다. CommunicationSystem을 사용하면 각 에이전트를 등록하고, 주문 정보가 메시지로 각 에이전트를 거치며 처리됩니다. 한 에이전트가 느려져도 큐에 메시지가 쌓이기만 할 뿐 전체 시스템이 멈추지 않습니다. 메시지 큐의 장점을 더 살펴봅시다. Queue는 스레드 안전하므로 여러 에이전트가 동시에 메시지를 보내도 문제없이 처리됩니다. 또한 FIFO(First In First Out) 방식으로 메시지 순서가 보장됩니다. 에이전트가 바쁠 때는 큐에 메시지가 쌓였다가, 여유가 생기면 순서대로 처리됩니다. 하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 메시지 큐의 크기를 제한하지 않는 것입니다. 수신 에이전트가 처리 속도를 따라가지 못하면 큐가 무한정 커져서 메모리가 고갈될 수 있습니다. 따라서 Queue 생성 시 maxsize를 설정하여 큐 크기를 제한하고, 큐가 가득 찬 경우의 처리 로직을 구현해야 합니다. 다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨와 함께 통신 시스템을 구현한 김개발 씨는 실제로 에이전트들이 메시지를 주고받으며 작업을 처리하는 것을 확인했습니다. "이제 전체 구조가 이해됐어요!" 에이전트 통신 시스템을 직접 구현해보면 멀티 에이전트 시스템의 핵심을 체득할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요. **실전 팁** 💡 - Queue의 maxsize를 설정하여 메모리 고갈을 방지하세요 - 메시지 검증은 send_message에서 수행하여 잘못된 메시지를 조기에 차단하세요 - 프로덕션 환경에서는 Redis나 RabbitMQ 같은 메시지 브로커를 사용하는 것이 좋습니다 --- ## 5. 실습_협업_태스크_수행 통신 시스템을 완성한 김개발 씨는 이제 실제로 여러 에이전트가 협업하여 복잡한 작업을 수행하도록 만들어야 했습니다. "이제 진짜 일을 시켜볼까요?" 박시니어 씨가 실전 예제를 보여주기 시작했습니다. **협업 태스크 수행**은 여러 에이전트가 통신 시스템을 통해 메시지를 주고받으며 하나의 큰 작업을 완수하는 것입니다. 마스터 에이전트가 작업을 분배하고, 워커 에이전트들이 각자 처리한 후 결과를 반환하는 전체 흐름을 실습합니다. 이를 통해 실무에서 사용할 수 있는 완전한 멀티 에이전트 시스템을 구축할 수 있습니다. 다음 코드를 살펴봅시다. ```python class WorkerAgent: def __init__(self, agent_id: str, comm_system: CommunicationSystem): self.agent_id = agent_id self.comm_system = comm_system comm_system.register_agent(agent_id) def process_task(self, task: str) -> str: # 실제 작업 수행 (여기서는 단순화) return f"Processed: {task}" def run(self): # 메시지 대기 및 처리 while True: msg = self.comm_system.receive_message(self.agent_id) result = self.process_task(msg.content) # 결과를 마스터에게 응답 response = AgentMessage( sender=self.agent_id, receiver=msg.sender, content=result, msg_type=MessageType.RESPONSE ) self.comm_system.send_message(response) ``` 김개발 씨는 통신 시스템까지 만들었지만, 이제 실제로 에이전트들이 일을 하도록 만들어야 했습니다. 이론과 시스템은 준비되었지만, 구체적으로 어떻게 작업을 분배하고 결과를 수집해야 할지 고민되었습니다. 박시니어 씨가 실전 예제를 보여주며 설명했습니다. "지금부터 텍스트 분석 시스템을 만들어볼 거예요. 마스터가 긴 문서를 여러 조각으로 나누고, 워커들이 각 조각을 분석한 후 결과를 마스터에게 돌려주는 시스템입니다." **협업 태스크 수행**이란 무엇일까요? 쉽게 비유하자면, 협업 태스크는 마치 대형 건설 프로젝트와 같습니다. 프로젝트 매니저(마스터)가 전체 건물 설계를 받아 각 층을 다른 건설팀(워커)에게 할당합니다. 1층은 A팀, 2층은 B팀, 3층은 C팀이 동시에 작업합니다. 각 팀이 자기 층을 완성하면 매니저에게 보고하고, 매니저는 모든 층이 완성되었는지 확인합니다. 에이전트 협업도 이와 동일한 구조입니다. 협업 시스템이 없던 시절에는 어떤 문제가 있었을까요? 초기에는 하나의 에이전트가 모든 작업을 순차적으로 처리했습니다. 100개의 파일을 분석해야 한다면 첫 번째 파일부터 마지막 파일까지 차례로 처리했습니다. 시간이 오래 걸렸고, 에이전트 하나가 고장 나면 전체 작업이 중단되었습니다. 자원 활용도 비효율적이었습니다. CPU 코어가 8개인 서버에서 1개만 사용하는 셈이었습니다. 바로 이런 비효율을 개선하기 위해 **협업 태스크 수행**이 등장했습니다. 협업 태스크를 사용하면 작업을 병렬로 처리하여 시간을 단축할 수 있습니다. 또한 한 워커가 실패해도 다른 워커는 계속 작업을 수행하므로 안정성이 높아집니다. 무엇보다 서버의 모든 리소스를 효율적으로 활용할 수 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다. WorkerAgent 클래스는 실제로 작업을 수행하는 에이전트입니다. __init__에서 자신을 통신 시스템에 등록합니다. process_task 메서드는 실제 작업 로직을 담고 있으며, 여기서는 단순히 "Processed: " 문자열을 붙여 반환합니다. run 메서드는 무한 루프로 메시지를 대기하고, 메시지가 도착하면 작업을 처리한 후 결과를 응답 메시지로 발신자에게 전송합니다. 실제 현업에서는 어떻게 활용할까요? 예를 들어 이메일 스팸 필터링 시스템을 만든다고 가정해봅시다. 하루에 수백만 개의 이메일이 들어옵니다. 마스터 에이전트가 이메일을 100개씩 묶어 10개의 워커 에이전트에게 분배합니다. 각 워커는 할당받은 이메일을 분석하여 스팸 여부를 판단하고 결과를 마스터에게 반환합니다. 마스터는 모든 결과를 데이터베이스에 저장합니다. 이렇게 하면 단일 에이전트로 처리할 때보다 10배 빠르게 작업을 완료할 수 있습니다. 마스터 에이전트는 어떻게 구현할까요? 마스터는 전체 작업을 작은 단위로 나누고, 각 워커에게 분배합니다. 워커들로부터 응답을 받으면 결과를 수집하고, 모든 작업이 완료되었는지 확인합니다. 필요하다면 실패한 작업을 재시도하거나 다른 워커에게 재할당할 수도 있습니다. 하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 워커의 run 메서드에 예외 처리를 하지 않는 것입니다. 작업 중 오류가 발생하면 워커 전체가 멈춰버립니다. 따라서 try-except 블록으로 감싸고, 오류가 발생하면 오류 메시지를 마스터에게 전송하여 마스터가 적절히 대응할 수 있도록 해야 합니다. 또 다른 주의사항은 무한 루프 종료 조건입니다. 위 코드의 run 메서드는 while True로 영원히 실행됩니다. 실제 시스템에서는 종료 신호를 받으면 루프를 빠져나오도록 구현해야 합니다. 예를 들어 특별한 타입의 메시지(SHUTDOWN)를 받으면 break하는 방식입니다. 스레딩을 활용하면 더 효과적입니다. 각 워커를 별도 스레드에서 실행하면 여러 워커가 동시에 작동합니다. Thread 객체를 생성하여 target으로 worker.run을 지정하고 start() 메서드를 호출하면 됩니다. 다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨와 함께 마스터와 워커를 구현한 김개발 씨는 실제로 텍스트 분석 작업이 병렬로 처리되는 것을 확인했습니다. "와, 정말 10배 빨라졌어요!" 협업 태스크 수행을 직접 구현해보면 멀티 에이전트 시스템의 진가를 체험할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요. **실전 팁** 💡 - 워커의 run 메서드에 예외 처리를 반드시 추가하여 안정성을 확보하세요 - 무한 루프에 종료 조건을 구현하여 우아한 종료(graceful shutdown)를 지원하세요 - Thread를 사용하여 워커를 병렬 실행하면 성능이 크게 향상됩니다 --- 이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

카테고리: LLM,통신,MultiAgent

언어: Python

태그: Python, Agent, Communication, Protocol, MultiAgent, LLM,통신,MultiAgent

작성자: AI Generated

CodeDeck

Agent Communication 완벽 가이드

목차

1. 에이전트 간 프로토콜

2. 메시지 포맷

3. 협업 패턴

4. 실습 에이전트 통신 시스템

5. 실습 협업 태스크 수행

댓글 (0)

함께 보면 좋은 카드 뉴스

Agent Debate 패턴 완벽 가이드

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

Hierarchical Agents 완벽 가이드

Memory Systems 완벽 가이드 - LLM 에이전트의 기억력 설계

Tool-Augmented Agents 완벽 가이드