단일 LLM 에이전트의 한계를 넘는 Google ADK 멀티에이전트 패턴 8가지

from google.adk.agents import SequentialAgent, ParallelAgent, LoopAgent, LlmAgent
 
data_collector = LlmAgent(name="DataCollector", model="gemini-2.0-flash")
analyzer = LlmAgent(name="Analyzer", model="gemini-2.0-flash")
reporter = LlmAgent(name="Reporter", model="gemini-2.0-flash")
 
pipeline = SequentialAgent(
    name="DataPipeline",
    sub_agents=[data_collector, analyzer, reporter]
)

from google.adk.agents import SequentialAgent, LlmAgent
 
research_agent = LlmAgent(
    name="Researcher",
    model="gemini-2.0-flash",
    instruction="주어진 주제에 대한 핵심 데이터를 수집하세요."
)
 
summary_agent = LlmAgent(
    name="Summarizer",
    model="gemini-2.0-flash",
    instruction="이전 에이전트의 조사 결과를 3문단으로 요약하세요."
)
 
fact_check_agent = LlmAgent(
    name="FactChecker",
    model="gemini-2.0-flash",
    instruction="요약된 내용에서 오류나 불확실한 주장을 검토하세요."
)
 
pipeline = SequentialAgent(
    name="ResearchPipeline",
    sub_agents=[research_agent, summary_agent, fact_check_agent]
)

from google.adk.agents import LlmAgent
 
billing_agent = LlmAgent(
    name="BillingSpecialist",
    model="gemini-2.0-flash",
    instruction="결제, 환불, 구독 관련 문의를 처리합니다."
)
 
tech_agent = LlmAgent(
    name="TechSupport",
    model="gemini-2.0-flash",
    instruction="기술적 오류, 설치, 연동 관련 문의를 처리합니다."
)
 
coordinator = LlmAgent(
    name="Coordinator",
    model="gemini-2.0-flash",
    instruction="""사용자 문의를 분석하고 적절한 전문가에게 라우팅하세요.
    - 결제/환불/구독 → BillingSpecialist
    - 기술 오류/설치/연동 → TechSupport""",
    sub_agents=[billing_agent, tech_agent]
)

from google.adk.agents import SequentialAgent, ParallelAgent, LlmAgent
 
doc_search = LlmAgent(
    name="DocSearch",
    model="gemini-2.0-flash",
    instruction="공식 문서에서 관련 내용을 검색합니다."
)
 
history_search = LlmAgent(
    name="HistorySearch",
    model="gemini-2.0-flash",
    instruction="사용자 이력에서 관련 케이스를 검색합니다."
)
 
parallel_search = ParallelAgent(
    name="ParallelSearch",
    sub_agents=[doc_search, history_search]
)
 
synthesizer = LlmAgent(
    name="Synthesizer",
    model="gemini-2.0-flash",
    instruction="병렬 검색 결과를 통합해 최적의 답변을 생성합니다."
)
 
# ParallelAgent 결과를 Synthesizer로 흘려보내는 순차 래퍼가 필수
search_and_synthesize = SequentialAgent(
    name="SearchAndSynthesize",
    sub_agents=[parallel_search, synthesizer]
)

from google.adk.agents import LlmAgent
 
frontend_agent = LlmAgent(
    name="FrontendArchitect",
    model="gemini-2.0-flash",
    instruction="프론트엔드 아키텍처와 UI 컴포넌트 설계를 담당합니다."
)
 
backend_agent = LlmAgent(
    name="BackendArchitect",
    model="gemini-2.0-flash",
    instruction="백엔드 API 설계와 데이터베이스 스키마를 담당합니다."
)
 
infra_agent = LlmAgent(
    name="InfraEngineer",
    model="gemini-2.0-flash",
    instruction="배포 인프라와 CI/CD 파이프라인 설계를 담당합니다."
)
 
project_manager = LlmAgent(
    name="ProjectManager",
    model="gemini-2.0-flash",
    instruction="""프로젝트 요구사항을 분석하고 아키텍처 설계 작업을 분배합니다.
    각 전문가에게 적절한 하위 작업을 위임하고 결과를 통합하세요.
    - UI/UX 관련 → FrontendArchitect
    - API/DB 관련 → BackendArchitect
    - 인프라/배포 관련 → InfraEngineer""",
    sub_agents=[frontend_agent, backend_agent, infra_agent]
)

from google.adk.agents import LoopAgent, LlmAgent
 
generator = LlmAgent(
    name="Generator",
    model="gemini-2.0-flash",
    instruction="기술 문서 초안을 작성합니다. 이전 피드백이 있다면 반영하세요."
)
 
critic = LlmAgent(
    name="Critic",
    model="gemini-2.0-flash",
    instruction="""문서를 검토하고 피드백을 제공하세요.
    모든 기준을 충족하면 응답에 'APPROVED'를 포함하고 escalate 플래그를 설정하세요.
    기준: 정확성, 명확성, 완전성"""
)
 
refinement_loop = LoopAgent(
    name="RefinementLoop",
    sub_agents=[generator, critic],
    max_iterations=5
)

from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
 
def request_human_approval(trade_details: dict) -> dict:
    """
    개념 설명용 단순화 버전.
    실제 구현에서는 아래 흐름처럼 외부 채널을 활용합니다:
 
    [실무 패턴 pseudocode]
    approval_id = generate_uuid()
    slack_client.chat_postMessage(
        channel="#trade-approvals",
        blocks=[
            {"type": "section", "text": f"거래 승인 요청: {trade_details}"},
            {"type": "actions", "elements": [
                {"type": "button", "text": "승인", "value": f"approve:{approval_id}"},
                {"type": "button", "text": "거부", "value": f"reject:{approval_id}"}
            ]}
        ]
    )
    result = approval_store.wait_for_response(approval_id, timeout_sec=3600)
    return {"approved": result.approved, "approver": result.user}
    """
    print(f"[승인 요청] 거래 내용: {trade_details}")
    approval = input("승인하시겠습니까? (yes/no): ")
    return {"approved": approval.lower() == "yes", "details": trade_details}
 
approval_tool = FunctionTool(func=request_human_approval)
 
approval_agent = LlmAgent(
    name="ApprovalGateway",
    model="gemini-2.0-flash",
    instruction="거래 추천안을 사용자에게 제시하고 승인을 요청합니다. 미승인 시 처리를 중단합니다.",
    tools=[approval_tool]
)

from google.adk.agents import LoopAgent, LlmAgent
 
# 웹 크롤링 시나리오: 수집 완료까지 반복
data_fetcher = LlmAgent(
    name="DataFetcher",
    model="gemini-2.0-flash",
    instruction="""다음 페이지의 데이터를 수집하세요.
    더 이상 수집할 데이터가 없으면 세션 상태에 escalate 플래그를 설정하세요."""
)
 
data_processor = LlmAgent(
    name="DataProcessor",
    model="gemini-2.0-flash",
    instruction="수집된 데이터를 정제하고 저장 형식으로 변환합니다."
)
 
crawl_loop = LoopAgent(
    name="CrawlLoop",
    sub_agents=[data_fetcher, data_processor],
    max_iterations=100
)

from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
 
def search_documentation(query: str) -> str:
    return f"'{query}'에 대한 문서 검색 결과..."
 
def run_code_analysis(code: str) -> str:
    return f"코드 분석 결과..."
 
def create_ticket(title: str, description: str) -> str:
    return f"티켓 생성 완료: {title}"
 
# LLM이 런타임에 도구 조합과 순서를 스스로 결정
agentic_agent = LlmAgent(
    name="AgenticWorkflow",
    model="gemini-2.0-flash",
    instruction="""사용자 요청을 분석하고 필요한 도구를 조합해서 처리하세요.
    문제를 진단하고, 해결책을 찾고, 필요하면 티켓을 생성하는 등
    상황에 따라 판단해서 진행하면 됩니다.""",
    tools=[
        FunctionTool(func=search_documentation),
        FunctionTool(func=run_code_analysis),
        FunctionTool(func=create_ticket)
    ]
)

from google.adk.agents import SequentialAgent, ParallelAgent, LoopAgent, LlmAgent
 
# 병렬 검색 (Pattern 3: Parallel Fan-Out)
doc_searcher = LlmAgent(name="DocSearcher", model="gemini-2.0-flash",
    instruction="공식 문서에서 해결책을 검색합니다.")
history_searcher = LlmAgent(name="HistorySearcher", model="gemini-2.0-flash",
    instruction="유사 사례 이력을 검색합니다.")
 
parallel_research = ParallelAgent(
    name="ParallelResearch",
    sub_agents=[doc_searcher, history_searcher]
)
 
# 품질 보증 루프 (Pattern 5: Generator-Critic)
answer_generator = LlmAgent(name="AnswerGenerator", model="gemini-2.0-flash",
    instruction="검색 결과를 바탕으로 고객 답변을 생성합니다.")
tone_critic = LlmAgent(name="ToneCritic", model="gemini-2.0-flash",
    instruction="""답변의 톤과 정확성을 검토합니다.
    기준 통과 시 'APPROVED', 미통과 시 구체적 피드백을 제공합니다.""")
 
quality_loop = LoopAgent(
    name="QualityLoop",
    sub_agents=[answer_generator, tone_critic],
    max_iterations=3
)
 
# 기술 지원 전체 파이프라인 (Pattern 1: Sequential)
tech_support_pipeline = SequentialAgent(
    name="TechSupportPipeline",
    sub_agents=[parallel_research, quality_loop]
)
 
# 최상위 라우터 (Pattern 2: Coordinator)
billing_specialist = LlmAgent(name="BillingSpecialist", model="gemini-2.0-flash",
    instruction="결제 및 구독 관련 문의를 전담합니다.")
 
support_coordinator = LlmAgent(
    name="SupportCoordinator",
    model="gemini-2.0-flash",
    instruction="""고객 문의 유형을 분석해 적절한 팀으로 라우팅합니다.
    - 결제/환불/청구 → BillingSpecialist
    - 기술 문제/오류/사용법 → TechSupportPipeline""",
    sub_agents=[billing_specialist, tech_support_pipeline]
)

from google.adk.agents import SequentialAgent, LlmAgent
from google.adk.tools import FunctionTool
 
def request_human_approval(trade_details: dict) -> dict:
    # 실제 구현: Slack 웹훅이나 이메일로 승인 요청 후 비동기 폴링
    print(f"[승인 요청] 거래 내용: {trade_details}")
    approval = input("승인하시겠습니까? (yes/no): ")
    return {"approved": approval.lower() == "yes", "details": trade_details}
 
approval_tool = FunctionTool(func=request_human_approval)
 
data_agent = LlmAgent(name="MarketDataAgent", model="gemini-2.0-flash",
    instruction="실시간 시장 데이터와 과거 데이터를 수집합니다.")
analysis_agent = LlmAgent(name="AnalysisAgent", model="gemini-2.0-flash",
    instruction="포트폴리오와 시장 데이터를 분석합니다.")
recommendation_agent = LlmAgent(name="RecommendationAgent", model="gemini-2.0-flash",
    instruction="분석 결과를 바탕으로 거래 추천안을 생성합니다.")
 
approval_agent = LlmAgent(
    name="ApprovalGateway",
    model="gemini-2.0-flash",
    instruction="거래 추천안을 사용자에게 제시하고 승인을 요청합니다.",
    tools=[approval_tool]
)
 
execution_agent = LlmAgent(name="TradeExecutor", model="gemini-2.0-flash",
    instruction="승인된 거래만 실행합니다. 미승인 시 처리를 중단합니다.")
 
trading_pipeline = SequentialAgent(
    name="TradingPipeline",
    sub_agents=[data_agent, analysis_agent, recommendation_agent,
                approval_agent, execution_agent]
)