Ai 마법서

무제

45 min read

아래 내용으로 인포그래픽을 만들어라. """

7장: 다중 에이전트 협업

단일 에이전트 아키텍처는 잘 정의된 문제에는 효과적일 수 있지만, 복잡하고 다면적인 작업에 직면했을 때 그 능력은 제약을 받습니다. 다중 에이전트 협업 패턴은 다양한 전문화된 에이전트의 협력적인 앙상블로 시스템을 구성하여 이러한 한계를 해결합니다. 이 접근 방식은 고수준 목표가 이산적인 하위 문제로 분해되고, 각 하위 문제가 해당 작업에 가장 적합한 특정 도구, 데이터 액세스 또는 추론 능력을 가진 에이전트에게 할당된다는 원칙에 기초합니다.

예를 들어, 복잡한 연구 쿼리는 정보 검색을 위한 연구 에이전트, 통계 처리를 위한 데이터 분석 에이전트, 최종 보고서 생성을 위한 종합 에이전트 등 여러 에이전트에게 분해되어 할당될 수 있습니다. 이러한 시스템의 효과는 단순히 노동 분할에만 있는 것이 아니라, 에이전트 간의 통신 메커니즘에 결정적으로 의존합니다. 이는 표준화된 통신 프로토콜과 공유된 온톨로지를 요구하여 에이전트가 정보를 교환하고, 하위 작업을 위임하며, 그들의 행동을 조정하여 최종 출력이 일관성을 갖도록 보장합니다.

이 분산된 아키텍처는 향상된 모듈성, 확장성 및 복원력과 같은 여러 이점을 제공하며, 단일 에이전트의 실패가 전체 시스템 실패를 초래하지 않습니다. 이러한 협업을 통해 집단적 강점은 개별 에이전트의 잠재력을 능가하는 시너지 결과를 창출합니다.

다중 에이전트 협업 패턴 개요

다중 에이전트 협업 패턴은 공통 목표를 달성하기 위해 함께 작동하는 여러 독립적이거나 반독립적인 에이전트로 구성된 시스템을 설계하는 것을 포함합니다. 각 에이전트는 일반적으로 정의된 역할, 전반적인 목표와 일치하는 특정 목표, 그리고 잠재적으로 다른 도구 또는 지식 기반에 대한 액세스 권한을 가집니다. 이 패턴의 힘은 이러한 에이전트 간의 상호 작용과 시너지 효과에 있습니다.

협업은 다음과 같은 여러 형태로 나타날 수 있습니다.

● 순차적 인계: 한 에이전트가 작업을 완료하고 그 출력을 파이프라인의 다음 단계에 대해 다른 에이전트에게 전달합니다(복잡한 작업을 다루는 계획 패턴과 유사).

  • 병렬 처리: 여러 에이전트가 문제의 다른 부분에서 동시에 작업하고 그 결과는 나중에 결합됩니다.
  • 토론 및 합의: 다양한 관점과 정보 소스를 가진 에이전트가 옵션을 평가하고, 궁극적으로 합의 또는 더 정보에 입각한 결정에 도달하기 위해 논쟁에 참여하는 다중 에이전트 협업.
  • 계층적 구조: 관리자 에이전트가 도구 액세스 또는 플러그인 기능에 따라 동적으로 작업 에이전트에게 작업을 위임하고 그 결과를 종합할 수 있습니다. 각 에이전트는 단일 에이전트가 모든 도구를 처리하는 대신 관련 도구 그룹을 처리할 수도 있습니다.
  • 전문가 팀: 다양한 도메인에서 전문 지식을 가진 에이전트(예: 연구원, 작가, 편집자)가 복잡한 출력을 생성하기 위해 협력합니다.
  • 비평가-검토자: 에이전트는 계획, 초안 또는 답변과 같은 초기 출력을 생성합니다. 두 번째 에이전트 그룹은 이 출력을 정책 준수, 보안, 규정 준수, 정확성, 품질 및 조직 목표와의 정렬에 대해 비판적으로 평가합니다. 원래 생성자 또는 최종 에이전트가 이 피드백을 기반으로 출력을 수정합니다. 이 패턴은 코드 생성, 연구 작문, 논리 확인 및 윤리적 정렬 보장에서 특히 효과적입니다. 이 접근 방식의 장점은 향상된 강력함, 개선된 품질 및 할루시네이션 또는 오류 가능성 감소입니다.
  • 비평가-검토자: 에이전트는 계획, 초안 또는 답변과 같은 초기 출력을 생성합니다. 두 번째 에이전트 그룹은 정책 준수, 보안, 규정 준수, 정확성, 품질 및 조직 목표와의 정렬에 대해 이 출력을 비판적으로 평가합니다. 원래 생성자 또는 최종 에이전트가 이 피드백을 기반으로 출력을 수정합니다. 이 패턴은 코드 생성, 연구 작문, 논리 확인 및 윤리적 정렬 보장에서 특히 효과적입니다. 이 접근 방식의 장점은 향상된 강력함, 개선된 품질 및 할루시네이션 또는 오류 가능성 감소입니다.

다중 에이전트 시스템(그림 1 참조)은 본질적으로 에이전트 역할 및 책임의 구획화, 에이전트가 정보를 교환하는 통신 채널 설정, 협력적 노력 지시를 위한 작업 흐름 또는 상호 작용 프로토콜 공식화로 구성됩니다.

Crew AI 및 Google ADK와 같은 프레임워크는 에이전트의 사양, 작업 및 상호 작용 절차를 정의할 수 있는 구조를 제공하여 이러한 패러다임을 용이하게 합니다. 이 접근 방식은 다양한 전문 지식, 여러 이산 단계 활용 또는 동시 처리 및 에이전트 간 정보 확인의 이점을 필요로 하는 과제에 특히 효과적입니다.

실용적인 응용 및 사용 사례

다중 에이전트 협업은 모듈성, 확장성 및 향상된 지능을 가능하게 하므로 다양한 도메인에 적용할 수 있는 강력한 패턴입니다.

  • 복잡한 연구 및 분석: 여러 에이전트 팀이 연구 프로젝트를 공동으로 수행할 수 있습니다. 한 에이전트는 학술 데이터베이스 검색을 전문으로 하고, 다른 에이전트는 요약, 다른 에이전트는 추세 식별, 네 번째 에이전트는 정보를 보고서로 종합하는 역할을 할 수 있습니다. 이는 인간 연구팀이 운영하는 방식과 유사합니다.
  • 소프트웨어 개발: 여러 에이전트가 소프트웨어 구축에 협력하는 것을 상상해 보십시오. 한 에이전트는 요구사항 분석가, 다른 에이전트는 코드 생성기, 세 번째 에이전트는 테스터, 네 번째 에이전트는 문서 작성자가 될 수 있습니다. 이들은 출력을 서로 전달하여 구성 요소를 구축하고 확인합니다.
  • 창의적인 콘텐츠 생성: 마케팅 캠페인 생성에는 시장 조사 에이전트, 카피라이터 에이전트, 그래픽 디자인 에이전트(이미지 생성 도구 사용), 소셜 미디어 예약 에이전트가 모두 함께 작업할 수 있습니다.
  • 재무 분석: 다중 에이전트 시스템은 금융 시장을 분석할 수 있습니다. 에이전트는 주식 데이터 가져오기, 뉴스 감성 분석, 기술적 분석 수행, 투자 권장 사항 생성 등을 전문으로 할 수 있습니다.
  • 고객 지원 에스컬레이션: 최전선 지원 에이전트가 초기 문의를 처리하고, 문제 복잡성에 따라 전문 에이전트(예: 기술 전문가 또는 청구 전문 에이전트)에게 문제를 에스컬레이션할 수 있으며, 이는 문제 복잡성에 따른 순차적 인계를 시연합니다.
  • 공급망 최적화: 에이전트는 공급업체, 제조업체, 유통업체와 같은 공급망의 다양한 노드를 나타내며 수요 또는 중단 변화에 대응하여 재고 수준, 물류 및 일정을 최적화하기 위해 협력할 수 있습니다.
  • 네트워크 분석 및 복구: 자율 운영은 특히 장애 지점 찾기에서 에이전트 아키텍처의 큰 이점을 얻습니다. 여러 에이전트가 문제를 분류하고 해결하기 위해 협력하여 최적의 조치를 제안할 수 있습니다. 이러한 에이전트는 기존 시스템을 활용하는 동시에 기존 머신러닝 모델 및 도구를 통합할 수 있으며, 동시에 생성형 AI의 이점을 제공합니다.

전문화된 에이전트를 분할하고 상호 관계를 세심하게 조정하는 능력은 개발자가 향상된 모듈성, 확장성 및 단일 통합 에이전트의 잠재 능력을 초과하는 복잡성을 다룰 수 있는 능력을 갖춘 시스템을 구축할 수 있도록 권한을 부여합니다.

다중 에이전트 협업: 상호 관계 및 통신 구조 탐색

복잡한 다중 에이전트 시스템을 설계하는 데 있어 에이전트가 상호 작용하고 통신하는 정교한 방식을 이해하는 것은 근본적으로 중요합니다. 그림 2에 묘사된 바와 같이, 가장 단순한 단일 에이전트 시나리오에서부터 복잡하고 사용자 지정 설계된 협업 프레임워크에 이르기까지 상호 관계 및 통신 모델의 스펙트럼이 존재합니다. 각 모델은 고유한 장점과 과제를 제시하며, 이는 다중 에이전트 시스템의 전반적인 효율성, 강력함 및 적응성에 영향을 미칩니다.

  • 1. 단일 에이전트: 가장 기본적인 수준에서 “단일 에이전트”는 다른 개체와의 직접적인 상호 작용이나 통신 없이 자율적으로 작동합니다. 이 모델은 구현 및 관리가 간단하지만, 그 능력은 본질적으로 개별 에이전트의 범위와 자원으로 제한됩니다. 단일하고 자급자족하는 에이전트에 의해 해결될 수 있는 독립적인 하위 문제들로 분해될 수 있는 작업에 적합합니다.
  • 2. 네트워크: “네트워크” 모델은 여러 에이전트가 탈중앙화된 방식으로 서로 직접 상호 작용하는 협업을 향한 중요한 단계를 나타냅니다. 통신은 일반적으로 피어 투 피어(peer-to-peer)로 발생하여 정보, 리소스 및 심지어 작업의 공유를 허용합니다. 이 모델은 한 에이전트의 실패가 전체 시스템을 마비시키지 않으므로 탄력성을 조성합니다. 그러나 대규모의 구조화되지 않은 네트워크에서 통신 오버헤드를 관리하고 일관된 의사 결정을 보장하는 것은 어려울 수 있습니다.
  • 3. 감독자(Supervisor): 전용 에이전트인 “감독자”가 하위 에이전트 그룹의 활동을 감독하고 조정합니다. 감독자는 통신, 작업 할당 및 갈등 해결을 위한 중앙 허브 역할을 합니다. 이 계층적 구조는 명확한 권한 라인을 제공하며 관리를 단순화할 수 있습니다. 그러나 감독자에게 단일 실패 지점이 발생하며, 감독자가 너무 많은 하위 에이전트나 복잡한 작업으로 압도되면 병목 현상이 될 수 있습니다.
  • 4. 도구로서의 감독자: 이 모델은 감독자의 역할이 다른 에이전트에게 모든 조치를 직접 명령하고 제어하는 것이 아니라, 다른 에이전트에게 리소스, 지침 또는 분석 지원을 제공하는 것으로 초점을 옮기는 “감독자” 개념의 미묘한 확장입니다. 감독자는 다른 에이전트가 작업을 더 효과적으로 수행할 수 있도록 도구, 데이터 또는 계산 서비스를 제공할 수 있지만, 모든 조치를 지시하지는 않습니다. 이 접근 방식은 감독자의 능력을 활용하면서도 엄격한 상향식 제어를 부과하지 않도록 목표합니다.
  • 5. 계층적(Hierarchical): “계층적” 모델은 감독자 개념을 확장하여 다층적인 조직 구조를 생성합니다. 여기에는 여러 수준의 감독자가 포함되며, 상위 수준의 감독자가 하위 수준의 감독자를 감독하고, 궁극적으로 가장 낮은 계층에 운영 에이전트 컬렉션이 있습니다. 이 구조는 복잡한 문제를 하위 문제로 분해하고 각 하위 문제를 특정 계층이 관리하도록 함으로써 확장성 및 복잡성 관리에 대한 구조화된 접근 방식을 제공합니다. 이는 정의된 경계 내에서 분산된 의사 결정을 허용합니다.

그림 2: 에이전트가 다양한 방식으로 통신하고 상호 작용합니다.

  • 6. 사용자 지정(Custom): “사용자 지정” 모델은 다중 에이전트 시스템 설계에서 궁극적인 유연성을 나타냅니다. 이는 주어진 문제 또는 애플리케이션의 특정 요구 사항에 정확하게 맞게 조정된 고유한 상호 관계 및 통신 구조를 생성할 수 있도록 합니다. 여기에는 이전에 언급된 모델의 요소를 결합하거나, 시스템 설계의 고유한 제약 조건 및 기회에서 발생하는 완전히 새로운 디자인이 포함될 수 있습니다. 사용자 지정 모델은 일반적으로 특정 성능 메트릭을 최적화하거나, 통신 프로토콜, 조정 메커니즘 및 잠재적 동작을 신중하게 고려하여 설계 및 구현에 대한 깊은 이해를 요구합니다.

요약하자면, 다중 에이전트 시스템의 상호 관계 및 통신 모델 선택은 중요한 설계 결정입니다. 각 모델은 고유한 장점과 단점을 제공하며, 최적의 선택은 작업의 복잡성, 에이전트 수, 원하는 자율성 수준, 강력함에 대한 필요성, 허용 가능한 통신 오버헤드와 같은 요소에 따라 달라집니다. 다중 에이전트 시스템에 대한 미래의 발전은 이러한 모델을 계속 탐색하고 개선할 가능성이 높으며, 협력적 지능을 위한 새로운 패러다임을 개발할 것입니다.

"""

또 한번 아래 내용은 파이썬 코딩이 들어가 있는데 인포그래픽에 추가해라.

"""

실습 코드 예제 (Crew AI)

이 Python 코드는 AI 기반 크루를 사용하여 AI 트렌드에 대한 블로그 게시물을 생성하도록 CrewAI 프레임워크를 사용합니다. 이 예제는 환경을 설정하고, .env 파일에서 API 키를 로드한 다음, 연구원이 AI 트렌드를 찾고 요약하도록 하고, 작가 에이전트가 연구를 기반으로 블로그 게시물을 작성하도록 하는 두 에이전트를 정의합니다.

연구 작업의 출력이 글쓰기 작업의 입력이 되도록 두 작업이 정의됩니다. 이 에이전트와 작업은 작업이 순차적으로 실행되도록 지정하는 Crew에 조립됩니다. 크루는 에이전트, 작업 및 언어 모델(구체적으로 “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 모델 시리즈를 사용하여 성능 향상을 위해 업데이트되었습니다.
   # 최첨단(미리 보기) 기능을 위해 "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년 AI의 상위 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)

다음 코드 예제는 계층적 에이전트 구조를 설정하는 방법을 보여줍니다. 여기에는 BaseAgent에서 파생된 사용자 지정 TaskExecutor와 LlmAgent의 두 가지 유형의 에이전트가 포함됩니다. TaskExecutor는 특정, 비-LLM 작업을 위해 설계되었으며 이 예제에서는 단순히 “작업이 성공적으로 완료되었습니다” 이벤트를 생성합니다. greeter라는 LlmAgent는 지정된 모델과 친근한 인사말을 하는 지침을 사용하여 초기화됩니다. task_doer라는 사용자 지정 TaskExecutor 인스턴스가 생성됩니다. 다음으로, coordinator라는 부모 LlmAgent가 모델과 지침을 사용하여 생성됩니다. 코디네이터의 지침은 인사 요청을 Greeter에게 위임하고 작업 실행을 TaskExecutor에게 위임하도록 안내합니다. Greeter와 TaskDoer는 코디네이터의 하위 에이전트로 추가되어 부모-자식 관계를 설정합니다. 코드는 이 관계가 올바르게 설정되었는지 확인하는 어설션(assertion)을 수행합니다. 마지막으로, 계층적 에이전트 구조가 성공적으로 생성되었음을 나타내는 메시지를 인쇄합니다.

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="사용자 요청을 Greeter에게 위임하고 작업을
TaskExecutor에게 위임할 수 있는 코디네이터입니다.",
   instruction="인사를 요청받으면 Greeter에게 위임하십시오. 작업을
실행하도록 요청받으면 TaskExecutor에게 위임하십시오.",
   sub_agents=[
       greeter,
       task_doer
   ]
)
# ADK 프레임워크는 자동으로 부모-자식 관계를 설정합니다.
# 이 어설션들은 초기화 후 확인되면 통과할 것입니다.
assert greeter.parent_agent == coordinator
assert task_doer.parent_agent == coordinator
print("에이전트 계층 구조가 성공적으로 생성되었습니다.")

이 코드는 Google ADK 내에서 계층적 에이전트 구조를 설정하는 방법을 보여줍니다. 이 코드는 BaseAgent에서 파생된 TaskExecutor라는 사용자 지정 에이전트를 정의합니다. 이 에이전트는 비-LLM 작업을 나타내며, 성공적으로 완료되었음을 나타내는 이벤트를 생성합니다. 그런 다음 greeter라는 표준 LlmAgent와 task_doer라는 TaskExecutor 인스턴스를 만듭니다. 이러한 하위 에이전트는 coordinator라는 부모 LlmAgent에 할당되어 계층 구조를 형성합니다. ADK는 이러한 관계를 자동으로 설정하며, 코드는 이 관계가 올바르게 설정되었는지 확인하는 어설션을 포함합니다. 이 예제는 ADK를 사용하여 단순한 비-LLM 기능과 LLM 추론 능력을 결합하는 방법을 보여줍니다.

다음 코드 발췌는 Google ADK 프레임워크 내에서 반복적 워크플로우를 설정하기 위해 LoopAgent를 활용하는 방법을 설명합니다. 이 코드는 두 에이전트, ConditionChecker 및 ProcessingStep을 정의합니다. ConditionChecker는 세션 상태에서 “status” 값을 확인하는 사용자 지정 에이전트입니다. “status”가 “completed”이면 ConditionChecker는 루프를 중지하기 위해 이벤트를 에스컬레이션합니다. 그렇지 않으면 루프를 계속하도록 이벤트를 생성합니다. ProcessingStep은 “gemini-2.0-flash-exp” 모델을 사용하는 LlmAgent입니다. 그 지침은 작업을 수행하고 최종 단계인 경우 세션 상태를 “completed”로 설정하는 것입니다. StatusPoller라는 LoopAgent가 생성됩니다. StatusPoller는 ProcessingStep과 ConditionChecker 인스턴스를 하위 에이전트로 포함합니다. LoopAgent는 이 두 하위 에이전트를 순차적으로 실행하며, 상태가 “completed”로 설정되거나 10회 반복 횟수에 도달할 때까지 반복합니다.

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="당신은 더 긴 프로세스의 한 단계입니다. 작업을
수행하십시오. 최종 단계인 경우 세션 상태를 'completed'로 설정하여
업데이트하십시오."
)
# LoopAgent는 워크플로우를 조정합니다.
poller = LoopAgent(
   name="StatusPoller",
   max_iterations=10,
   sub_agents=[
       process_step,
       ConditionChecker() # 잘 정의된 사용자 지정 에이전트
인스턴스화.
   ]
)
# 이 폴러는 'process_step'을 실행하고 'ConditionChecker'를
반복적으로 실행합니다.

다음 코드 발췌는 선형 워크플로우를 구축하기 위해 엔지니어링된 Google ADK 내의 SequentialAgent 패턴을 설명합니다. 이 코드는 두 에이전트인 step1과 step2를 정의합니다. step1은 “Step1_Fetch”로 이름이 지정되고 그 출력은 “data” 키 아래에 세션 상태에 저장됩니다. step2는 “Step2_Process”로 이름이 지정되며, session.state[“data”]에 저장된 정보를 분석하고 요약을 제공하도록 지시받습니다. SequentialAgent인 “MyPipeline”은 이러한 하위 에이전트의 실행을 조정합니다. 파이프라인이 초기 입력으로 실행되면 Step1이 먼저 실행됩니다. Step1의 응답은 “data”로 세션 상태에 저장된 다음 Step2가 지침에 따라 상태의 정보(data)를 사용하여 실행됩니다. 이 구조는 한 에이전트의 출력이 다음 에이전트의 입력이 되는 워크플로우를 구축할 수 있도록 합니다.

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는 이러한 하위 에이전트의 실행을 조정하여 병렬로 작동할 수 있도록 합니다. 두 에이전트의 결과는 수집되어 최종 상태에 저장될 것입니다. 마지막으로, 예제는 에이전트 실행이 완료된 후 최종 상태에서 수집된 날씨 및 뉴스 데이터에 액세스하는 방법을 보여줍니다.

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 내에서 에이전트를 도구로 사용하는 패러다임을 보여주며, 한 에이전트가 다른 에이전트의 기능을 도구처럼 활용할 수 있도록 합니다. 이 코드는 두 에이전트, artist_agent와 image_generator_agent를 사용하여 이미지 생성 시스템을 정의합니다. 이미지 생성의 핵심 기능을 시뮬레이션하는 간단한 도구인 generate_image 함수가 있습니다. artist_agent는 창의적인 이미지 프롬프트를 발명하고 그 다음 image_tool 래퍼를 통해 image_generator_agent에게 작업을 위임합니다. 이 구조는 계층적 에이전트 시스템을 만드는 데 사용될 수 있으며, 상위 에이전트가 하위 에이전트의 전문 기능을 활용할 수 있도록 합니다.

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]
)

시각적 요약

그림 1: 병렬화 설계 패턴

주요 요점

다음은 주요 요점입니다.

  • 병렬화는 독립적인 작업을 동시에 실행하여 효율성을 개선하기 위한 패턴입니다.
  • 외부 자원을 기다리는 작업(API 호출과 같은)이 있을 때 특히 유용합니다.
  • 병렬화는 계산 복잡성과 비용을 증가시키므로 주의해서 사용해야 합니다.
  • LangChain 및 Google ADK와 같은 프레임워크는 병렬 실행을 정의하고 관리하기 위한 내장된 구성을 제공합니다.
  • 병렬화는 전체 지연 시간을 줄여 에이전트 시스템을 더 반응적으로 만드는 데 도움이 됩니다.

결론

이 장에서는 독립적인 하위 작업을 병렬로 실행하여 계산 워크플로우를 최적화하는 방법인 병렬화 패턴을 탐구했습니다. 이 접근 방식은 여러 모델 추론 또는 외부 서비스 호출과 관련된 복잡한 작업에서 전체 지연 시간을 줄입니다.

프레임워크는 이 패턴을 구현하기 위한 구별된 메커니즘을 제공합니다. LangChain에서는 RunnableParallel과 같은 구조가 여러 처리 체인을 동시에 정의하고 실행하는 데 사용됩니다. 대조적으로, Google ADK와 같은 프레임워크는 코디네이터 모델이 다양한 하위 작업을 할당하고 이러한 에이전트가 병렬로 작동하도록 허용하여 다중 에이전트 위임을 통해 병렬화를 달성할 수 있습니다.

순차적(체이닝) 및 조건부(라우팅) 제어 흐름에 병렬 처리를 통합함으로써 개발자는 다양한 작업을 효율적으로 관리하고 복잡한 워크플로우를 수행할 수 있는 정교하고 고성능의 계산 시스템을 구축할 수 있습니다. """ 전체를 만들면 내용이 두 개를 따로 어떻게 작업하는지 알겠지?

Graph View