7장: 멀티 에이전트 협업
단일 에이전트 아키텍처는 잘 정의된 문제에 효과적일 수 있지만, 복잡하고 다중 도메인 작업에 직면했을 때 그 능력이 종종 제약됩니다. 멀티 에이전트 협업 패턴은 시스템을 구별되고 전문화된 에이전트들의 협력적인 앙상블로 구조화함으로써 이러한 한계를 해결합니다. 이 접근 방식은 작업 분해(task decomposition) 원칙에 기반하며, 여기서 상위 수준의 목표는 개별적인 하위 문제들로 나뉩니다. 그런 다음 각 하위 문제는 해당 작업에 가장 적합한 특정 도구, 데이터 액세스 또는 추론 기능을 가진 에이전트에게 할당됩니다.
예를 들어, 복잡한 연구 질문은 정보 검색을 위한 연구 에이전트, 통계 처리를 위한 데이터 분석 에이전트, 최종 보고서 생성을 위한 종합 에이전트로 분해되어 할당될 수 있습니다. 이러한 시스템의 효능은 단순히 노동 분할에만 기인하는 것이 아니라, 에이전트 간 통신 메커니즘에 결정적으로 의존합니다. 이를 위해서는 표준화된 통신 프로토콜과 공유 온톨로지가 필요하며, 에이전트들이 데이터를 교환하고, 하위 작업을 위임하며, 최종 출력이 일관성을 유지하도록 행동을 조정할 수 있어야 합니다.
이 분산 아키텍처는 향상된 모듈성, 확장성 및 견고성을 포함한 여러 가지 이점을 제공하는데, 이는 단일 에이전트의 장애가 반드시 전체 시스템 장애를 유발하지 않기 때문입니다. 협업은 앙상블 내의 단일 에이전트의 잠재적 능력보다 집합적 성능이 뛰어난 시너지 효과를 낳을 수 있게 합니다.
멀티 에이전트 협업 패턴 개요
멀티 에이전트 협업 패턴은 여러 독립적이거나 준독립적인 에이전트들이 공통의 목표를 달성하기 위해 함께 작업하는 시스템을 설계하는 것을 포함합니다. 각 에이전트는 일반적으로 정의된 역할, 전체 목표에 부합하는 구체적인 목표, 그리고 잠재적으로 다른 도구 또는 지식 기반에 대한 접근 권한을 가집니다. 이 패턴의 강력함은 에이전트들 간의 상호작용과 시너지 효과에 있습니다.
협업은 다양한 형태로 나타날 수 있습니다.
● 순차적 인계 (Sequential Handoffs): 한 에이전트가 작업을 완료하고 그 출력을 다음 에이전트에게 파이프라인의 다음 단계로 전달합니다 (계획 패턴과 유사하지만, 명시적으로 다른 에이전트들을 포함함).
- 병렬 처리 (Parallel Processing): 여러 에이전트가 문제의 다른 부분에 대해 동시에 작업하고, 그 결과는 나중에 결합됩니다.
- 토론 및 합의 (Debate and Consensus): 다양한 관점과 정보 출처를 가진 에이전트들이 옵션을 평가하기 위해 토론에 참여하여 궁극적으로 합의나 보다 정보에 입각한 결정에 도달하는 멀티 에이전트 협업입니다.
- 계층적 구조 (Hierarchical Structures): 관리자 에이전트는 작업 도구 접근 권한이나 플러그인 기능에 따라 작업 에이전트들에게 동적으로 작업을 위임하고 그 결과를 종합할 수 있습니다. 각 에이전트는 또한 단일 에이전트가 모든 도구를 처리하는 대신 관련 도구 그룹을 처리할 수도 있습니다.
- 전문가 팀 (Expert Teams): 서로 다른 도메인(예: 연구원, 작가, 편집자)의 전문 지식을 가진 에이전트들이 복잡한 출력을 생성하기 위해 협업합니다.
- ● 비평가-검토자 (Critic-Reviewer): 에이전트들이 계획, 초안 또는 답변과 같은 초기 출력을 생성합니다. 그런 다음 두 번째 에이전트 그룹이 정책 준수, 보안, 규정 준수, 정확성, 품질 및 조직 목표와의 정렬에 대해 이 출력을 비판적으로 평가합니다. 원래 작성자 또는 최종 에이전트가 이 피드백을 기반으로 출력을 수정합니다. 이 패턴은 코드 생성, 연구 작성, 논리 확인 및 윤리적 정렬 보장에 특히 효과적입니다. 이 접근 방식의 장점은 향상된 견고성, 개선된 품질, 환각 또는 오류 가능성 감소를 포함합니다.
멀티 에이전트 시스템(그림 1 참조)은 근본적으로 에이전트 역할과 책임의 구분, 에이전트들이 정보를 교환하는 통신 채널의 확립, 그리고 그들의 협력 노력을 지시하는 작업 흐름 또는 상호작용 프로토콜의 공식화를 구성합니다.
그림 1: 멀티 에이전트 시스템의 예시
Crew AI 및 Google ADK와 같은 프레임워크는 에이전트, 작업 및 상호작용 절차의 사양을 위한 구조를 제공함으로써 이 패러다임을 용이하게 하도록 설계되었습니다. 이 접근 방식은 다양한 전문 지식이 필요하거나, 여러 개별 단계에 걸쳐 있거나, 동시 처리 및 에이전트 간 정보 확인의 이점을 활용해야 하는 과제에 특히 효과적입니다.
실제 애플리케이션 및 사용 사례
멀티 에이전트 협업은 수많은 도메인에 적용 가능한 강력한 패턴입니다.
-
복잡한 연구 및 분석: 에이전트 팀이 연구 프로젝트에서 협력할 수 있습니다. 한 에이전트는 학술 데이터베이스 검색을 전문으로 하고, 다른 에이전트는 결과 요약을, 세 번째 에이전트는 동향 파악을, 네 번째 에이전트는 정보를 보고서로 종합하는 것을 전문으로 할 수 있습니다. 이는 인간 연구팀이 운영되는 방식과 유사합니다.
-
소프트웨어 개발: 에이전트들이 소프트웨어 구축을 위해 협력하는 것을 상상해 보십시오. 한 에이전트는 요구 사항 분석가, 다른 에이전트는 코드 생성기, 세 번째 에이전트는 테스터,
-
그리고 네 번째 에이전트는 문서 작성자가 될 수 있습니다. 그들은 구성 요소를 구축하고 검증하기 위해 서로에게 출력을 전달할 수 있습니다.
-
창의적 콘텐츠 생성: 마케팅 캠페인을 만드는 것은 시장 조사 에이전트, 카피라이터 에이전트, 그래픽 디자인 에이전트(이미지 생성 도구 사용), 그리고 소셜 미디어 스케줄링 에이전트가 모두 함께 작업하는 것을 포함할 수 있습니다.
-
재무 분석: 멀티 에이전트 시스템은 금융 시장을 분석할 수 있습니다. 에이전트들은 주식 데이터 가져오기, 뉴스 심리 분석, 기술적 분석 수행, 투자 권고안 생성 등을 전문으로 할 수 있습니다.
-
고객 지원 에스컬레이션: 일선 지원 에이전트는 초기 문의를 처리하고, 필요할 때 전문 에이전트(예: 기술 전문가 또는 청구 전문가)에게 복잡한 문제를 에스컬레이션하여 문제 복잡성에 기반한 순차적 인계를 시연할 수 있습니다.
-
공급망 최적화: 에이전트들은 공급망의 다른 노드(공급업체, 제조업체, 유통업체)를 나타낼 수 있으며, 변화하는 수요나 중단에 대응하여 재고 수준, 물류 및 스케줄링을 최적화하기 위해 협력할 수 있습니다.
-
네트워크 분석 및 복구: 자율 운영은 특히 장애 지점을 찾는 데 있어 에이전트 아키텍처로부터 큰 이점을 얻습니다. 여러 에이전트가 협력하여 문제를 분류하고 복구하며 최적의 조치를 제안할 수 있습니다. 이러한 에이전트들은 기존 시스템의 이점을 활용하는 동시에 생성형 AI의 이점을 제공하기 위해 기존 머신러닝 모델 및 도구와 통합될 수도 있습니다.
전문 에이전트를 구분하고 그들의 상호 관계를 세심하게 조정하는 능력은 개발자가 향상된 모듈성, 확장성 및 단일 통합 에이전트에게는 불가능했을 복잡성을 다룰 수 있는 능력을 갖춘 시스템을 구축할 수 있도록 힘을 실어줍니다.
멀티 에이전트 협업: 상호 관계 및 통신 구조 탐색
에이전트들이 상호 작용하고 통신하는 복잡한 방식을 이해하는 것은 효과적인 멀티 에이전트 시스템을 설계하는 데 근본적입니다. 그림 2에 묘사된 바와 같이, 가장 단순한 단일 에이전트 시나리오에서부터 복잡하고 맞춤 설계된 협업 프레임워크에 이르기까지 다양한 상호 관계 및 통신 모델의 스펙트럼이 존재합니다. 각 모델은 고유한 장점과 과제를 제시하며, 멀티 에이전트 시스템의 전반적인 효율성, 견고성 및 적응성에 영향을 미칩니다.
- 1. 단일 에이전트 (Single Agent): 가장 기본적인 수준에서 “단일 에이전트”는 다른 개체와의 직접적인 상호 작용이나 통신 없이 자율적으로 작동합니다. 이 모델은 구현 및 관리가 간단하지만, 그 능력은 본질적으로 개별 에이전트의 범위와 자원에 의해 제한됩니다. 이는 단일 자급자족 에이전트에 의해 해결 가능한 독립적인 하위 문제들로 분해될 수 있는 작업에 적합합니다.
- 2. 네트워크 (Network): “네트워크” 모델은 여러 에이전트가 분산된 방식으로 서로 직접 상호 작용하는 협업을 향한 상당한 발전을 나타냅니다. 통신은 일반적으로 피어 투 피어로 발생하여 정보, 리소스 및 심지어 작업 공유를 허용합니다. 이 모델은 한 에이전트의 장애가 반드시 전체 시스템을 마비시키지 않으므로 복원력을 조성합니다. 그러나 크고 구조화되지 않은 네트워크에서 통신 오버헤드를 관리하고 일관된 의사 결정을 보장하는 것이 어려울 수 있습니다.
- 3. 감독자 (Supervisor): “감독자” 모델에서는 전담 에이전트인 “감독자”가 일련의 하위 에이전트들의 활동을 감독하고 조정합니다. 감독자는 통신, 작업 할당 및 갈등 해결을 위한 중앙 허브 역할을 합니다. 이 계층적 구조는 명확한 권한 체계를 제공하며 관리 및 제어를 단순화할 수 있습니다. 그러나 이는 단일 장애 지점(감독자)을 도입하며, 감독자가 너무 많은 수의 하위 에이전트나 복잡한 작업으로 인해 압도될 경우 병목 현상이 될 수 있습니다.
- 4. 도구로서의 감독자 (Supervisor as a Tool): 이 모델은 감독자의 역할이 직접적인 명령 및 제어보다는 다른 에이전트에게 리소스, 지침 또는 분석적 지원을 제공하는 것에 더 가깝도록 “감독자” 개념을 미묘하게 확장한 것입니다. 감독자는 다른 에이전트의 모든 행동을 지시하지 않으면서도 다른 에이전트가 작업을 더 효과적으로 수행할 수 있도록 도구, 데이터 또는 계산 서비스를 제공할 수 있습니다. 이 접근 방식은 엄격한 상향식 제어를 부과하지 않으면서 감독자의 역량을 활용하는 것을 목표로 합니다.
- 5. 계층적 (Hierarchical): “계층적” 모델은 감독자 개념을 확장하여 다단계 조직 구조를 만듭니다. 이는 여러 수준의 감독자를 포함하며, 상위 수준 감독자가 하위 수준 감독자를 감독하고, 궁극적으로 최하위 계층에 운영 에이전트 집합이 존재합니다. 이 구조는 복잡한 문제를 하위 문제들로 분해할 수 있고, 각 하위 문제는 계층의 특정 계층에 의해 관리되는 경우에 적합합니다. 이는 정의된 경계 내에서 분산된 의사 결정을 허용하여 확장성과 복잡성 관리에 구조화된 접근 방식을 제공합니다.
그림 2: 에이전트들이 다양한 방식으로 통신하고 상호 작용함.
6. 맞춤형 (Custom): “맞춤형” 모델은 멀티 에이전트 시스템 설계에서 궁극적인 유연성을 나타냅니다. 이는 주어진 문제 또는 애플리케이션의 특정 요구 사항에 정확하게 맞춰진 고유한 상호 관계 및 통신 구조를 생성할 수 있도록 합니다. 여기에는 이전에 언급된 모델들의 요소를 결합하는 하이브리드 접근 방식이나 환경의 고유한 제약 조건 및 기회에서 발생하는 완전히 새로운 설계가 포함될 수 있습니다. 맞춤형 모델은 종종 특정 성능 지수를 최적화하거나, 고도로 동적인 환경을 처리하거나, 도메인별 지식을 시스템 아키텍처에 통합해야 할 필요성에서 비롯됩니다. 맞춤형 모델을 설계하고 구현하는 것은 일반적으로 멀티 에이전트 시스템 원칙에 대한 깊은 이해와 통신 프로토콜, 조정 메커니즘 및 나타나는 행동에 대한 세심한 고려가 필요합니다.
요약하자면, 멀티 에이전트 시스템을 위한 상호 관계 및 통신 모델의 선택은 중요한 설계 결정입니다. 각 모델은 뚜렷한 장점과 단점을 제공하며, 최적의 선택은 작업의 복잡성, 에이전트 수, 원하는 자율성 수준, 견고성에 대한 필요성, 허용 가능한 통신 오버헤드와 같은 요소에 따라 달라집니다. 멀티 에이전트 시스템의 미래 발전은 이러한 모델들을 계속 탐색하고 개선할 것이며, 협력적 인텔리전스를 위한 새로운 패러다임을 개발할 것입니다.
실습 코드 (Crew AI)
이 Python 코드는 AI 트렌드에 대한 블로그 게시물을 생성하기 위해 CrewAI 프레임워크를 사용하여 AI 기반 크루를 정의합니다. 이는 .env 파일에서 API 키를 로드하여 환경을 설정하는 것으로 시작됩니다. 애플리케이션의 핵심은 두 에이전트를 정의하는 것입니다. 하나는 AI 트렌드를 찾고 요약하는 연구원이고, 다른 하나는 연구를 기반으로 블로그 게시물을 작성하는 작가입니다.
이에 따라 두 가지 작업이 정의됩니다. 하나는 트렌드를 연구하는 작업이고 다른 하나는 연구 작업의 출력에 의존하는 블로그 게시물을 작성하는 작업입니다. 그런 다음 이러한 에이전트와 작업이 크루로 조립되며, 작업이 순서대로 실행되는 순차적 프로세스가 지정됩니다. 크루는 에이전트, 작업 및 언어 모델(“gemini-2.0-flash” 모델로 구체적으로 지정)로 초기화됩니다. 메인 함수는 kickoff() 메서드를 사용하여 이 크루를 실행하여 원하는 출력을 생성하기 위해 에이전트 간의 협업을 조정합니다. 마지막으로 코드는 크루 실행의 최종 결과인 생성된 블로그 게시물을 인쇄합니다.
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from langchain_google_genai import ChatGoogleGenerativeAI
def setup_environment():
"""환경 변수를 로드하고 필수 API 키를 확인합니다."""
load_dotenv()
if not os.getenv("GOOGLE_API_KEY"):
raise ValueError("GOOGLE_API_KEY를 찾을 수 없습니다. .env 파일에 설정해 주세요.")
def main():
"""
최신 Gemini 모델을 사용하여 콘텐츠 생성을 위한 AI 크루를 초기화하고 실행합니다.
"""
setup_environment()
# 사용할 언어 모델을 정의합니다.
# 성능 및 기능 향상을 위해 Gemini 2.0 시리즈의 모델로 업데이트되었습니다.
# 최첨단(미리 보기) 기능을 사용하려면 "gemini-2.5-flash"를 사용할 수 있습니다.
llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash")
# 특정 역할과 목표를 가진 에이전트 정의
researcher = Agent(
role='선임 연구 분석가',
goal='AI의 최신 동향을 찾고 요약합니다.',
backstory="당신은 주요 동향을 파악하고 정보를 종합하는 데 재주가 있는 숙련된 연구 분석가입니다.",
verbose=True,
allow_delegation=False,
)
writer = Agent(
role='기술 콘텐츠 작가',
goal='연구 결과를 바탕으로 명확하고 매력적인 블로그 게시물을 작성합니다.',
backstory="당신은 복잡한 기술 주제를 접근하기 쉬운 콘텐츠로 변환할 수 있는 숙련된 작가입니다.",
verbose=True,
allow_delegation=False,
)
# 에이전트에 대한 작업 정의
research_task = Task(
description="2024-2025년 인공 지능의 상위 3가지 신흥 동향을 조사합니다. 실제 적용과 잠재적 영향에 중점을 둡니다.",
expected_output="주요 요점과 출처를 포함한 상위 3가지 AI 동향에 대한 자세한 요약.",
agent=researcher,
)
writing_task = Task(
description="연구 결과를 바탕으로 500단어 블로그 게시물을 작성합니다. 게시물은 일반 독자가 이해하기 쉽고 매력적이어야 합니다.",
expected_output="최신 AI 동향에 대한 완성된 500단어 블로그 게시물.",
agent=writer,
context=[research_task],
)
# 크루 생성
blog_creation_crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
process=Process.sequential,
llm=llm,
verbose=2 # 상세한 크루 실행 로그를 위해 상세 모드 설정
)
# 크루 실행
print("## Gemini 2.0 Flash로 블로그 생성 크루 실행 중... ##")
try:
result = blog_creation_crew.kickoff()
print("\n------------------\n")
print("## 크루 최종 출력 ##")
print(result)
except Exception as e:
print(f"\n예상치 못한 오류가 발생했습니다: {e}")
if __name__ == "__main__":
main()
이제 Google ADK 프레임워크 내에서 계층적, 병렬 및 순차적 조정 패러다임과 운영 도구로서의 에이전트 구현에 중점을 둔 추가 예제를 살펴보겠습니다.
실습 코드 (Google ADK)
다음 코드 예제는 부모-자식 관계를 생성하여 Google ADK 내에서 계층적 에이전트 구조를 설정하는 것을 보여줍니다. 코드는 LlmAgent와 BaseAgent에서 파생된 사용자 지정 TaskExecutor 에이전트의 두 가지 유형의 에이전트를 정의합니다. TaskExecutor는 특정 비-LLM 작업을 위해 설계되었으며 이 예제에서는 단순히 “작업이 성공적으로 완료되었습니다” 이벤트를 산출합니다. 지정된 모델과 친근한 인사 담당자 역할을 하라는 지침으로 LlmAgent인 greeter가 초기화됩니다. 사용자 지정 TaskExecutor는 task_doer로 인스턴스화됩니다. 조정자라고 불리는 부모 LlmAgent는 모델과 지침으로 생성됩니다. 조정자의 지침은 greeter에게 인사하도록 위임하고 task_doer에게 작업 실행을 위임하도록 안내합니다. greeter와 task_doer는 조정자의 하위 에이전트로 추가되어 부모-자식 관계를 설정합니다. 코드는 이 관계가 올바르게 설정되었는지 확인합니다. 마지막으로, 에이전트 계층 구조가 성공적으로 생성되었음을 나타내는 메시지를 인쇄합니다.
from google.adk.agents import LlmAgent, BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from typing import AsyncGenerator
# BaseAgent를 확장하여 사용자 지정 에이전트를 올바르게 구현합니다.
class TaskExecutor(BaseAgent):
"""사용자 지정, 비-LLM 동작을 가진 특수 에이전트입니다."""
name: str = "TaskExecutor"
description: str = "미리 정의된 작업을 실행합니다."
async def _run_async_impl(self, context: InvocationContext) ->
AsyncGenerator[Event, None]:
"""작업에 대한 사용자 지정 구현 논리입니다."""
# 여기에 사용자 지정 논리가 들어갑니다.
# 이 예제에서는 간단한 이벤트를 산출하기만 합니다.
yield Event(author=self.name, content="작업이 성공적으로 완료되었습니다.")
# 적절한 초기화로 개별 에이전트 정의
# LlmAgent는 모델을 지정해야 합니다.
greeter = LlmAgent(
name="Greeter",
model="gemini-2.0-flash-exp",
instruction="당신은 친절한 인사 담당자입니다."
)
task_doer = TaskExecutor() # 구체적인 사용자 지정 에이전트 인스턴스화
# 부모 에이전트를 생성하고 하위 에이전트 할당
# 부모 에이전트의 설명과 지침은 위임 논리를 안내해야 합니다.
coordinator = LlmAgent(
name="Coordinator",
model="gemini-2.0-flash-exp",
description="사용자에게 인사하고 작업을 실행할 수 있는 조정자입니다.",
instruction="인사를 요청받으면 Greeter에게 위임하고, 작업을 수행하라는 요청을 받으면 TaskExecutor에게 위임하십시오.",
sub_agents=[
greeter,
task_doer
]
)
# ADK 프레임워크는 부모-자식
관계를 자동으로 설정합니다.
# 초기화 후 확인하면 이 주장들은 통과될 것입니다.
assert greeter.parent_agent == coordinator
assert task_doer.parent_agent == coordinator
print("에이전트 계층 구조가 성공적으로 생성되었습니다.")
이 코드 발췌문은 Google ADK 내에서 반복적인 워크플로를 설정하기 위해 LoopAgent를 사용하는 것을 설명합니다. 코드는 두 에이전트인 ConditionChecker와 ProcessingStep을 정의합니다. ConditionChecker는 세션 상태에서 “status” 값을 확인하는 사용자 지정 에이전트입니다. “status”가 “completed”인 경우 ConditionChecker는 루프를 중지하기 위해 이벤트를 에스컬레이션합니다. 그렇지 않으면 루프를 계속하기 위해 이벤트를 산출합니다. ProcessingStep은 “gemini-2.0-flash-exp” 모델을 사용하는 LlmAgent입니다. 그 지침은 작업을 수행하고 최종 단계인 경우 세션 “status”를 “completed”로 설정하는 것입니다. StatusPoller라는 LoopAgent가 생성됩니다. StatusPoller는 max_iterations=10으로 구성됩니다. StatusPoller는 ProcessingStep과 ConditionChecker의 인스턴스 모두를 하위 에이전트로 포함합니다. LoopAgent는 최대 10번의 반복 동안 하위 에이전트들을 순차적으로 실행하며, ConditionChecker가 상태가 “completed”임을 발견하면 중지됩니다.
import asyncio
from typing import AsyncGenerator
from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.events import Event, EventActions
from google.adk.agents.invocation_context import InvocationContext
# 모범 사례: 사용자 지정 에이전트를 완전하고 자체 설명적인 클래스로 정의합니다.
class ConditionChecker(BaseAgent):
"""세션 상태에서 'completed' 상태를 확인하는 사용자 지정 에이전트입니다."""
name: str = "ConditionChecker"
description: str = "프로세스가 완료되었는지 확인하고 루프 중지를 신호합니다."
async def _run_async_impl(
self, context: InvocationContext
) -> AsyncGenerator[Event, None]:
"""상태를 확인하고 루프를 계속하거나 중지하기 위해 이벤트를 산출합니다."""
status = context.session.state.get("status", "pending")
is_done = (status == "completed")
if is_done:
# 조건이 충족되면 루프를 종료하기 위해 에스컬레이션합니다.
yield Event(author=self.name,
actions=EventActions(escalate=True))
else:
# 루프를 계속하기 위해 간단한 이벤트를 산출합니다.
yield Event(author=self.name, content="조건이 충족되지 않았으므로 루프를 계속합니다.")
# 수정: LlmAgent는 모델과 명확한 지침을 가져야 합니다.
process_step = LlmAgent(
name="ProcessingStep",
model="gemini-2.0-flash-exp",
instruction="당신은 더 긴 프로세스의 단계입니다. 작업을 수행하십시오. 최종 단계인 경우 세션 상태를 'status'를 'completed'로 설정하여 업데이트하십시오."
)
# LoopAgent가 워크플로를 조정합니다.
poller = LoopAgent(
name="StatusPoller",
max_iterations=10,
sub_agents=[
process_step,
ConditionChecker() # 잘 정의된 사용자 지정 에이전트 인스턴스화.
]
)
# 이제 이 폴러는 'process_step'을 실행한 다음
# 상태가 'completed'가 되거나 10회 반복이 지날 때까지 'ConditionChecker'를
# 반복적으로 실행합니다.
이 코드 발췌문은 Google ADK 내에서 선형 워크플로 구축을 위해 설계된 SequentialAgent 패턴을 설명합니다. 이 코드는 google.adk.agents 라이브러리를 사용하여 순차적 에이전트 파이프라인을 정의합니다. 파이프라인은 step1과 step2라는 두 개의 에이전트로 구성됩니다. step1은 “Step1_Fetch”로 이름 붙여지며 그 출력은 “data” 키 아래 세션 상태에 저장됩니다. step2는 “Step2_Process”로 이름 붙여지며 session.state[“data”]에 저장된 정보를 분석하고 요약을 제공하도록 지시됩니다. “MyPipeline”이라는 SequentialAgent가 이 하위 에이전트들의 실행을 조정합니다. 파이프라인이 초기 입력으로 실행되면 step1이 먼저 실행됩니다. step1의 응답은 “data” 키 아래 세션 상태에 저장됩니다. 그런 다음 step2가 실행되어 지침에 따라 상태에 배치된 정보를 활용합니다. 이 구조는 한 에이전트의 출력이 다음 에이전트의 입력이 되는 워크플로를 구축할 수 있도록 합니다. 이는 다단계 AI 또는 데이터 처리 파이프라인을 만드는 일반적인 패턴입니다.
from google.adk.agents import SequentialAgent, Agent
# 이 에이전트의 출력은 session.state["data"]에 저장됩니다.
step1 = Agent(name="Step1_Fetch", output_key="data")
# 이 에이전트는 이전 단계의 데이터를 사용할 것입니다.
# 이 데이터를 찾는 방법과 사용하는 방법을 지침으로 제공합니다.
step2 = Agent(
name="Step2_Process",
instruction="state['data']에서 찾은 정보를 분석하고 요약을 제공합니다."
)
pipeline = SequentialAgent(
name="MyPipeline",
sub_agents=[step1, step2]
)
# 파이프라인이 초기 입력으로 실행되면 Step1이 실행되고,
# 그 응답이 session.state["data"]에 저장된 후,
# Step2가 지침에 따라 상태의 정보를 사용하여 실행됩니다.
다음 코드 예제는 여러 에이전트 작업을 동시 실행할 수 있도록 하는 Google ADK 내의 ParallelAgent 패턴을 보여줍니다. data_gatherer는 weather_fetcher와 news_fetcher라는 두 하위 에이전트를 병렬로 실행하도록 설계되었습니다. weather_fetcher 에이전트는 주어진 위치의 날씨를 가져와 결과를 session.state[“weather_data”]에 저장하도록 지시됩니다. 마찬가지로 news_fetcher 에이전트는 주어진 주제에 대한 주요 뉴스를 검색하고 session.state[“news_data”]에 저장하도록 지시됩니다. 각 하위 에이전트는 “gemini-2.0-flash-exp” 모델을 사용하도록 구성됩니다. ParallelAgent는 이 하위 에이전트들의 실행을 조정하여 병렬로 작업할 수 있도록 합니다. weather_fetcher와 news_fetcher 모두의 결과는 에이전트 실행 완료 후 세션 상태에 수집되어 저장될 것입니다. 마지막으로,
예제는 에이전트 실행이 완료된 후 최종 상태에서 수집된 날씨 및 뉴스 데이터에 액세스하는 방법을 보여줍니다.
from google.adk.agents import Agent, ParallelAgent
# 가져오기 논리를 에이전트의 도구로 정의하는 것이 더 좋습니다.
# 이 예제의 단순성을 위해 에이전트의 지침 내에 논리를 포함할 것입니다.
# 실제 시나리오에서는 도구를 사용할 것입니다.
# 병렬로 실행될 개별 에이전트 정의
weather_fetcher = Agent(
name="weather_fetcher",
model="gemini-2.0-flash-exp",
instruction="주어진 위치의 날씨를 가져와 날씨 보고서만 반환합니다.",
output_key="weather_data" # 결과는 session.state["weather_data"]에 저장됩니다.
)
news_fetcher = Agent(
name="news_fetcher",
model="gemini-2.0-flash-exp",
instruction="주어진 주제에 대한 주요 뉴스를 가져와 해당 뉴스 기사만 반환합니다.",
output_key="news_data" # 결과는 session.state["news_data"]에 저장됩니다.
)
# 하위 에이전트를 조정하기 위한 ParallelAgent 생성
data_gatherer = ParallelAgent(
name="data_gatherer",
sub_agents=[
weather_fetcher,
news_fetcher
]
)
제공된 코드 세그먼트는 Google ADK 내에서 에이전트가 함수 호출과 유사한 방식으로 다른 에이전트의 기능을 활용할 수 있도록 하는 “도구로서의 에이전트” 패러다임을 보여줍니다. 특히, 이 코드는 Google의 LlmAgent 및 AgentTool 클래스를 사용하여 이미지 생성 시스템을 정의합니다. 이는 부모인 artist_agent와 하위 에이전트인 image_generator_agent로 구성됩니다. generate_image 함수는 모의 이미지 데이터를 반환하여 이미지 생성을 시뮬레이션하는 간단한 도구입니다. image_generator_agent는 텍스트 프롬프트를 기반으로 이 도구를 사용하는 책임을 집니다. artist_agent의 역할은 먼저 창의적인 이미지 프롬프트를 고안하는 것입니다. 그런 다음 AgentTool 래퍼를 통해 image_generator_agent를 호출합니다. AgentTool은 브리지 역할을 하여 한 에이전트가 다른 에이전트를 도구로 사용하도록 허용합니다. artist_agent가 image_tool을 호출하면 AgentTool은 artist가 발명한 프롬프트로 image_generator_agent를 호출합니다. image_generator_agent는 해당 프롬프트를 사용하여 generate_image 함수를 사용합니다. 마지막으로 생성된 이미지(또는 모의 데이터)는 에이전트를 통해 다시 반환됩니다. 이 아키텍처는 상위 수준 에이전트가 하위 수준의 전문화된 에이전트를 조정하여 작업을 수행하는 계층적 에이전트 시스템을 시연합니다.
from google.adk.agents import LlmAgent
from google.adk.tools import agent_tool
from google.genai import types
# 1. 핵심 기능에 대한 간단한 함수 도구입니다.
# 이는 조치와 추론을 분리하는 모범 사례를 따릅니다.
def generate_image(prompt: str) -> dict:
"""
텍스트 프롬프트를 기반으로 이미지를 생성합니다.
Args:
prompt: 생성할 이미지에 대한 자세한 설명입니다.
Returns:
상태와 생성된 이미지 바이트를 포함하는 사전입니다.
"""
print(f"도구: 프롬프트 '{prompt}'에 대한 이미지 생성 중")
# 실제 구현에서는 이미지 생성 API를 호출할 것입니다.
# 이 예제에서는 모의 이미지 데이터를 반환합니다.
mock_image_bytes = b"mock_image_data_for_a_cat_wearing_a_hat"
return {
"status": "success",
# 도구는 원시 바이트를 반환하며, 에이전트가 부분 생성을 처리합니다.
"image_bytes": mock_image_bytes,
"mime_type": "image/png"
}
# 2. ImageGeneratorAgent를 LlmAgent로 리팩토링합니다.
# 이제 전달된 입력을 올바르게 사용합니다.
image_generator_agent = LlmAgent(
name="ImageGen",
model="gemini-2.0-flash",
description="상세한 텍스트 프롬프트를 기반으로 이미지를 생성합니다.",
instruction=(
"당신은 이미지 생성 전문가입니다. 당신의 임무는 사용자의 요청을 받아 "
"`generate_image` 도구를 사용하여 이미지를 생성하는 것입니다. "
"사용자의 전체 요청은 도구의 'prompt' 인수로 사용되어야 합니다. "
"도구가 이미지 바이트를 반환한 후에는 반드시 이미지를 출력해야 합니다."
),
tools=[generate_image]
)
# 3. 수정된 에이전트를 AgentTool로 래핑합니다.
# 여기에 있는 설명은 부모 에이전트가 보는 내용입니다.
image_tool = agent_tool.AgentTool(
agent=image_generator_agent,
description="이 도구를 사용하여 이미지를 생성하십시오. 입력은 원하는 이미지에 대한 설명적인 프롬프트여야 합니다."
)
# 4. 부모 에이전트는 변경되지 않았습니다. 그 논리는 올바르게 작동했습니다.
artist_agent = LlmAgent(
name="Artist",
model="gemini-2.0-flash",
instruction=(
"당신은 창의적인 예술가입니다. 먼저 이미지에 대한 창의적이고 상세한 프롬프트를 만드십시오. "
"그런 다음 `ImageGen` 도구를 사용하여 당신의 프롬프트를 사용하여 이미지를 생성하십시오."
),
tools=[image_tool]
)
한눈에 보기
무엇을: 복잡한 문제는 종종 단일의 단일체 LLM 기반 에이전트의 능력을 초과합니다. 단일 에이전트는 다면적인 작업의 모든 부분을 해결하는 데 필요한 다양하고 전문화된 기술이나 특정 도구 액세스가 부족할 수 있습니다. 이러한 한계는 전체 시스템의 효율성을 감소시키는 병목 현상을 만듭니다. 결과적으로, 정교하고 다중 도메인 목표를 다루는 것은 비효율적이며 불완전하거나 차선의 결과로 이어질 수 있습니다.
왜: 멀티 에이전트 협업 패턴은 여러 협력 에이전트들로 시스템을 생성하여 표준화된 해결책을 제공합니다. 복잡한 문제는 더 작고 관리하기 쉬운 하위 문제로 분해됩니다. 그런 다음 각 하위 문제는 해당 문제를 해결하는 데 필요한 정확한 도구와 기능을 갖춘 전문 에이전트에게 할당됩니다. 이러한 에이전트들은 정의된 통신 프로토콜과 순차적 인계, 병렬 작업 스트림 또는 계층적 위임과 같은 상호작용 모델을 통해 함께 작동합니다. 이 에이전트 기반의 분산 접근 방식은 시너지 효과를 창출하여, 단일 에이전트에게는 불가능했을 결과를 그룹이 달성할 수 있도록 합니다.
경험 법칙: 작업이 단일 에이전트에게 너무 복잡하고 전문적인 기술이나 도구가 필요한 별개의 하위 작업으로 분해될 수 있을 때 이 패턴을 사용하십시오. 이는 복잡한 연구 및 분석, 소프트웨어 개발 또는 창의적 콘텐츠 생성과 같이 다양한 전문 지식이나 여러 단계가 필요한 문제에 이상적입니다.
시각적 요약
/03.-AI-Coding-Agent/02.-Agentic_Design_Patterns/Agentic_Design_Patterns-113-131/_chap_7_Figure_0.jpeg)
그림 3: 멀티 에이전트 설계 패턴
주요 요점
- 멀티 에이전트 협업은 공통의 목표를 달성하기 위해 함께 작업하는 여러 에이전트를 포함합니다.
- 이 패턴은 전문화된 역할, 분산된 작업 및 에이전트 간 통신을 활용합니다.
- 협업은 순차적 인계, 병렬 처리, 토론 또는 계층적 구조와 같은 형태를 취할 수 있습니다.
- 이 패턴은 다양한 전문 지식이나 여러 단계가 필요한 복잡한 문제에 이상적입니다.
결론
본 장에서는 멀티 에이전트 협업 패턴을 탐구하며, 시스템 내에서 여러 전문 에이전트를 조정하는 것의 이점을 시연했습니다. 우리는 다양한 협업 모델을 검토했으며, 복잡하고 다면적인 문제를 다양한 도메인에 걸쳐 해결하는 데 있어 이 패턴의 필수적인 역할을 강조했습니다. 에이전트 협업에 대한 이해는 자연스럽게 그들의 외부 환경과의 상호 작용에 대한 탐구로 이어집니다.
참고 문헌
-
- 멀티 에이전트 협업 메커니즘: LLM에 대한 설문조사, https://arxiv.org/abs/2501.06322