Google ADK 멀티에이전트 시스템: 컨텍스트 공유와 에스컬레이션 완전 가이드

# 세션 스코프 — 현재 세션에서만 유효
session.state['current_step'] = 'analysis'

# 사용자 스코프 — 같은 사용자의 모든 세션에서 공유
session.state['user:preferred_language'] = 'ko'

# 앱 스코프 — 모든 사용자와 세션에서 공유
session.state['app:global_config'] = 'v2'

# 임시 스코프 — 현재 호출(invocation)에서만 유효, 호출 완료 후 폐기
session.state['temp:intermediate_result'] = {'score': 0.95}

from google.adk.agents import LlmAgent, SequentialAgent

agent_a = LlmAgent(
    name="AgentA",
    model="gemini-2.0-flash",
    instruction="프랑스의 수도를 찾아주세요.",
    output_key="capital_city"  # 결과를 state['capital_city']에 저장
)

agent_b = LlmAgent(
    name="AgentB",
    model="gemini-2.0-flash",
    instruction="{capital_city}에 대해 자세히 알려주세요."
    # {capital_city} → state['capital_city'] 값이 자동 주입됨
)

pipeline = SequentialAgent(
    name="CityInfoPipeline",
    sub_agents=[agent_a, agent_b]
)

from google.adk.agents import LlmAgent

billing_agent = LlmAgent(
    name="Billing",
    model="gemini-2.0-flash",
    description="결제 관련 문의를 처리합니다."
)

support_agent = LlmAgent(
    name="Support",
    model="gemini-2.0-flash",
    description="기술 지원 요청을 처리합니다."
)

coordinator = LlmAgent(
    name="HelpDesk",
    model="gemini-2.0-flash",
    instruction="결제 문제는 Billing에게, 기술 문제는 Support에게 전달하세요.",
    sub_agents=[billing_agent, support_agent]
)

from google.adk.agents import LlmAgent
from google.adk.tools import agent_tool

summarizer = LlmAgent(
    name="Summarizer",
    model="gemini-2.0-flash",
    description="텍스트를 요약합니다."
)

research_agent = LlmAgent(
    name="Researcher",
    model="gemini-2.0-flash",
    instruction="주제를 조사하고, Summarizer 도구를 사용해 결과를 요약하세요.",
    tools=[agent_tool.AgentTool(agent=summarizer)]
)

from google.adk.agents import SequentialAgent, LlmAgent

# Step1이 temp: 상태에 데이터를 저장
step1 = LlmAgent(
    name="Analyzer",
    model="gemini-2.0-flash",
    instruction="데이터를 분석하세요.",
    output_key="temp:analysis_result"
)

# Step2가 같은 temp: 상태에서 데이터를 읽음
step2 = LlmAgent(
    name="Reporter",
    model="gemini-2.0-flash",
    instruction="{temp:analysis_result}를 바탕으로 보고서를 작성하세요."
)

pipeline = SequentialAgent(
    name="AnalysisPipeline",
    sub_agents=[step1, step2]
)

from google.adk.agents import ParallelAgent, SequentialAgent, LlmAgent

# 병렬 실행 시 서로 다른 output_key를 사용해야 경합 조건을 방지
fetch_weather = LlmAgent(
    name="WeatherFetcher",
    model="gemini-2.0-flash",
    instruction="날씨 정보를 가져오세요.",
    output_key="weather_data"  # 고유한 키
)

fetch_news = LlmAgent(
    name="NewsFetcher",
    model="gemini-2.0-flash",
    instruction="뉴스를 가져오세요.",
    output_key="news_data"  # 고유한 키
)

gather = ParallelAgent(
    name="InfoGatherer",
    sub_agents=[fetch_weather, fetch_news]
)

synthesizer = LlmAgent(
    name="Synthesizer",
    model="gemini-2.0-flash",
    instruction="{weather_data}와 {news_data}를 종합하세요."
)

workflow = SequentialAgent(
    name="GatherAndSynthesize",
    sub_agents=[gather, synthesizer]
)

from google.adk.tools import ToolContext

def search_database(query: str, tool_context: ToolContext) -> dict:
    # State에서 사용자 설정 읽기
    user_lang = tool_context.state.get("user:preferred_language", "en")
    
    # 도구 실행 로직
    results = perform_search(query, language=user_lang)
    
    # State에 결과 저장 (자동으로 EventActions.state_delta에 추적됨)
    tool_context.state["last_search_query"] = query
    tool_context.state["temp:search_result_count"] = len(results)
    
    return {"results": results, "count": len(results)}

from google.adk.agents.context import Context
from google.adk.models import LlmRequest
from google.genai import types

def before_model_callback(context: Context, request: LlmRequest):
    call_count = context.state.get("model_calls", 0)
    context.state["model_calls"] = call_count + 1
    
    # 특정 조건에서 모델 호출을 가로채기
    if context.state.get("temp:skip_model"):
        return types.Content(
            parts=[types.Part(text="캐시된 응답을 반환합니다.")]
        )
    
    return None  # 정상적으로 모델 호출 진행

from google.adk.tools import ToolContext

async def intelligent_search(query: str, tool_context: ToolContext) -> dict:
    # 과거 세션 메모리에서 관련 정보 검색
    relevant_memories = await tool_context.search_memory(
        f"{query}와 관련된 과거 대화"
    )
    
    # 세션 아티팩트 목록 조회
    artifacts = await tool_context.list_artifacts()
    
    # 특정 아티팩트 로드
    config = await tool_context.load_artifact("search_config.json")
    
    # 결과를 아티팩트로 저장
    await tool_context.save_artifact(
        "search_results.json",
        types.Part(text=json.dumps(results))
    )
    
    return {"results": results, "memory_context": relevant_memories}

from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.events import Event, EventActions
from google.adk.agents.invocation_context import InvocationContext

class QualityGate(BaseAgent):
    async def _run_async_impl(self, ctx: InvocationContext):
        status = ctx.session.state.get("quality_status", "fail")
        should_stop = (status == "pass")
        yield Event(
            author=self.name,
            actions=EventActions(escalate=should_stop)
        )

code_refiner = LlmAgent(
    name="CodeRefiner",
    model="gemini-2.0-flash",
    instruction="코드를 개선하세요. 현재 코드: {current_code}",
    output_key="current_code"
)

quality_checker = LlmAgent(
    name="QualityChecker",
    model="gemini-2.0-flash",
    instruction="{current_code}의 품질을 평가하세요. 'pass' 또는 'fail'로 답하세요.",
    output_key="quality_status"
)

refinement_loop = LoopAgent(
    name="RefinementLoop",
    max_iterations=5,
    sub_agents=[code_refiner, quality_checker, QualityGate(name="Gate")]
)

billing_agent = LlmAgent(
    name="Billing",
    model="gemini-2.0-flash",
    description="결제 관련 문의를 처리합니다.",
    instruction="""결제 관련 질문에 답하세요.
    기술 지원이 필요한 질문이면 Support 에이전트로 전환하세요.
    일반적인 문의면 HelpDesk(부모)로 전환하세요."""
)

support_agent = LlmAgent(
    name="Support",
    model="gemini-2.0-flash",
    description="기술 지원을 제공합니다.",
    instruction="""기술 문제를 해결하세요.
    결제 관련 문의면 Billing 에이전트로 전환하세요."""
)

coordinator = LlmAgent(
    name="HelpDesk",
    model="gemini-2.0-flash",
    instruction="사용자 요청을 분석해 적절한 에이전트에게 전달하세요.",
    sub_agents=[billing_agent, support_agent]
)

specialist = LlmAgent(
    name="DataAnalyst",
    model="gemini-2.0-flash",
    description="데이터 분석 전문가입니다.",
    instruction="요청된 데이터 분석을 수행하세요.",
    fallback_to_parent=True  # 작업 완료 후 자동으로 부모에게 복귀
)

coordinator = LlmAgent(
    name="ProjectManager",
    model="gemini-2.0-flash",
    instruction="프로젝트 관리를 담당합니다. 데이터 분석이 필요하면 DataAnalyst에게 위임하세요.",
    sub_agents=[specialist]
)

from google.adk.agents import (
    LlmAgent, SequentialAgent, ParallelAgent, LoopAgent, BaseAgent
)
from google.adk.tools import agent_tool, ToolContext
from google.adk.events import Event, EventActions


# === 도구 정의: ToolContext를 통한 상태 접근 ===

def lookup_customer(customer_id: str, tool_context: ToolContext) -> dict:
    """고객 정보 조회 도구 — State를 통해 결과를 공유"""
    customer = db.get_customer(customer_id)
    tool_context.state["user:customer_tier"] = customer["tier"]
    tool_context.state["temp:customer_name"] = customer["name"]
    return customer

def check_order_status(order_id: str, tool_context: ToolContext) -> dict:
    """주문 상태 확인 — 메모리 검색으로 과거 맥락 활용"""
    past_context = tool_context.search_memory(f"주문 {order_id} 관련 이력")
    status = db.get_order(order_id)
    return {"status": status, "history": past_context}


# === 전문 서브에이전트 ===

billing_specialist = LlmAgent(
    name="BillingSpecialist",
    model="gemini-2.0-flash",
    description="결제, 환불, 청구서 관련 문제를 처리합니다.",
    instruction="""결제 관련 문제를 해결하세요.
    고객 등급은 {user:customer_tier}입니다.
    기술 문제라면 TechSupport로 전환하세요.
    해결 완료되면 요약을 작성하세요.""",
    output_key="resolution_summary",
    fallback_to_parent=True  # 완료 후 자동 복귀
)

tech_support = LlmAgent(
    name="TechSupport",
    model="gemini-2.0-flash",
    description="기술 지원과 계정 접근 문제를 처리합니다.",
    instruction="""기술 문제를 해결하세요.
    결제 문제라면 BillingSpecialist로 전환하세요.""",
    output_key="resolution_summary",
    fallback_to_parent=True
)


# === 품질 검증 루프 ===

class ResolutionValidator(BaseAgent):
    """해결 결과 검증 — escalate로 루프 탈출"""
    async def _run_async_impl(self, ctx):
        score = ctx.session.state.get("satisfaction_score", 0)
        yield Event(
            author=self.name,
            actions=EventActions(escalate=(score >= 4))
        )

satisfaction_check = LlmAgent(
    name="SatisfactionChecker",
    model="gemini-2.0-flash",
    instruction="""고객 만족도를 1-5점으로 평가하세요.
    해결 내용: {resolution_summary}""",
    output_key="satisfaction_score"
)

quality_loop = LoopAgent(
    name="QualityAssurance",
    max_iterations=3,
    sub_agents=[satisfaction_check, ResolutionValidator(name="Validator")]
)


# === 루트 에이전트: 전체 오케스트레이션 ===

root_agent = LlmAgent(
    name="CustomerServiceHub",
    model="gemini-2.0-flash",
    instruction="""고객 지원 허브입니다.
    1. 먼저 고객 정보를 조회하세요.
    2. 문제 유형에 따라 적절한 전문가에게 전달하세요.
    3. 해결 후 품질을 검증하세요.""",
    tools=[lookup_customer, check_order_status],
    sub_agents=[billing_specialist, tech_support]
)