에이전트 개발 키트를 사용한 Agent Platform 메모리 뱅크 빠른 시작

Agent Platform 메모리 뱅크를 사용하면 에이전트가 세션 전반에서 장기 메모리를 관리할 수 있습니다. 에이전트 개발 키트 (ADK)와 함께 사용하면 에이전트가 메모리 뱅크 호출을 자동으로 조정하여 사용자 상호작용을 기반으로 메모리를 저장하고 검색할 수 있습니다.

이 문서에서는 ADK 에이전트를 만들고, 메모리 뱅크를 사용하도록 구성하고, 에이전트와 상호작용하여 메모리를 생성하고 액세스하는 방법을 설명합니다.

ADK 없이 API를 직접 호출하는 방법에 대한 자세한 내용은 메모리 뱅크 API 빠른 시작을 참조하세요.

과 같은 오류가 발생할 수 있습니다.

ADK 메모리 서비스 및 메모리 뱅크로 메모리 관리

VertexAiMemoryBankService 는 ADK의 BaseMemoryService로 정의된 메모리 뱅크용 ADK 래퍼입니다. 메모리 서비스와 상호작용하여 메모리를 읽고 쓰는 콜백과 도구를 정의할 수 있습니다.

VertexAiMemoryBankService 인터페이스에는 다음이 포함됩니다.

  • memory_service.add_session_to_memory 트리거합니다 GenerateMemories 메모리 뱅크에 요청 제공된 adk.Session의 모든 이벤트를 소스 콘텐츠로 사용하여. 콜백에서 callback_context.add_session_to_memory를 사용하여 이 메서드 호출을 조정할 수 있습니다.

    from google.adk.agents.callback_context import CallbackContext
    
    async def add_session_to_memory_callback(callback_context: CallbackContext):
        await callback_context.add_session_to_memory()
        return None
    
  • memory_service.add_events_to_memory 이벤트 하위 집합을 사용하여 메모리 뱅크에 GenerateMemories 요청을 트리거합니다. 콜백에서 callback_context.add_events_to_memory를 사용하여 이 메서드 호출을 조정할 수 있습니다.

    from google.adk.agents.callback_context import CallbackContext
    
    async def add_events_to_memory_callback(callback_context: CallbackContext):
        await callback_context.add_events_to_memory(events=callback_context.session.events[-5:-1])
        return None
    
  • memory_service.search_memory는 메모리 뱅크에 관련 메모리를 가져오기 위한 RetrieveMemories 요청을 트리거합니다. 현재 user_idapp_name. 기본 제공 메모리 도구 (LoadMemoryTool 또는 PreloadMemoryTool) 또는 tool_context.search_memory를 호출하는 커스텀 도구를 사용하여 이 메서드 호출을 조정할 수 있습니다.

시작하기 전에

이 튜토리얼에 설명된 단계를 완료하려면 먼저 메모리 뱅크 설정 페이지의 시작하기 섹션에 있는 단계를 수행해야 합니다.

환경 변수 설정하기

ADK를 사용하려면 환경 변수를 설정하세요.

import os

os.environ["GOOGLE_GENAI_USE_ENTERPRISE"] = "TRUE"
os.environ["GOOGLE_CLOUD_PROJECT"] = "PROJECT_ID"
os.environ["GOOGLE_CLOUD_LOCATION"] = "LOCATION"

다음을 바꿉니다.

ADK 에이전트 만들기

메모리 지원 에이전트를 만들려면 메모리 서비스 호출을 조정하는 도구와 콜백을 설정합니다.

메모리 생성 콜백 정의

메모리 생성 호출을 조정하려면 메모리 생성을 트리거하는 콜백 함수를 만듭니다. 백그라운드에서 처리할 이벤트 하위 집합 (callback_context.add_events_to_memory) 또는 세션의 모든 이벤트 (callback_context.add_session_to_memory)를 전송할 수 있습니다.

from google.adk.agents.callback_context import CallbackContext

async def generate_memories_callback(callback_context: CallbackContext):
    # Option 1 (Recommended): Send events to Memory Bank for memory generation,
    # which is ideal for incremental processing of events.
    await callback_context.add_events_to_memory(
      events=callback_context.session.events[-5:-1])

    # Option 2: Send the full session to Memory Bank for memory generation.
    # It's recommended to only call this at the end of a session to minimize
    # how many times a single event is re-processed.
    await callback_context.add_session_to_memory()

    return None

메모리 검색 도구 정의

ADK 에이전트를 개발할 때는, 에이전트가 메모리를 언제 가져오고 프롬프트에 메모리를 어떻게 포함할지를 제어하는 메모리 도구를 포함해야 합니다.

PreloadMemoryTool을 사용하면 에이전트가 각 턴의 시작 시점에 메모리를 가져오고 가져온 메모리를 시스템 지침에 포함합니다. 이는 사용자에 대한 기준 컨텍스트를 설정하는 데 유용합니다. LoadMemoryTool을 사용하면 모델이 메모리가 사용자 쿼리에 답변하는 데 필요하다고 판단할 때 이 도구를 호출합니다.

from google import adk
from google.adk.tools.load_memory_tool import LoadMemoryTool
from google.adk.tools.preload_memory_tool import PreloadMemoryTool

memory_retrieval_tools = [
  # Option 1: Retrieve memories at the start of every turn.
  PreloadMemoryTool(),
  # Option 2: Retrieve memories via tool calls. The model will only call this tool
  # when it decides that memories are necessary to respond to the user query.
  LoadMemoryTool()
]

agent = adk.Agent(
    model="gemini-2.5-flash",
    name='stateful_agent',
    instruction="""You are a Vehicle Voice Agent, designed to assist users with information and in-vehicle actions.

1.  **Direct Action:** If a user requests a specific vehicle function (e.g., "turn on the AC"), execute it immediately using the corresponding tool. You don't have the outcome of the actual tool execution, so provide a hypothetical tool execution outcome.
2.  **Information Retrieval:** Respond concisely to general information requests with your own knowledge (e.g., restaurant recommendation).
3.  **Clarity:** When necessary, try to seek clarification to better understand the user's needs and preference before taking an action.
4.  **Brevity:** Limit responses to under 30 words.
""",
    tools=memory_retrieval_tools,
    after_agent_callback=generate_memories_callback
)

또는 메모리를 검색하는 자체 커스텀 도구를 만들 수 있습니다. 이는 에이전트에 메모리를 검색할 시점을 안내하려는 경우에 유용합니다.

from google import adk
from google.adk.tools import ToolContext, FunctionTool

async def search_memories(query: str, tool_context: ToolContext):
  """Query this tool when you need to fetch information about user preferences."""
  return await tool_context.search_memory(query)

agent = adk.Agent(
    model="gemini-2.5-flash",
    name='stateful_agent',
    instruction="""...""",
    tools=[FunctionTool(func=search_memories)],
    after_agent_callback=generate_memories_callback
)

ADK 메모리 뱅크 메모리 서비스 및 런타임 정의

메모리 지원 에이전트를 만든 후에는 메모리 서비스에 연결해야 합니다. ADK 메모리 서비스 구성 프로세스는 에이전트, 도구, 콜백의 실행을 조정하는 ADK 에이전트가 실행되는 위치에 따라 다릅니다.

Agent Runtime 인스턴스 만들기

먼저 메모리 뱅크에 사용할 Agent Runtime 인스턴스를 만들어야 합니다. Agent Runtime 런타임을 사용하여 에이전트를 배포하는 경우 이 단계는 선택사항입니다. 메모리 뱅크 동작 맞춤설정에 대한 자세한 내용은 메모리 뱅크 설정 페이지의 메모리 뱅크용 Agent Runtime 인스턴스 구성 섹션을 참조하세요.

import vertexai

client = vertexai.Client(
  project="PROJECT_ID",
  location="LOCATION"
)
# If you don't have an Agent Runtime instance already, create an Agent Platform
# Memory Bank instance using the default configuration.
agent_engine = client.agent_engines.create()

# Optionally, print out the resource name. You will need the
# resource name if you want to interact with your Runtime instance later on.
print(agent_engine.api_resource.name)

agent_engine_id = agent_engine.api_resource.name.split("/")[-1]

다음을 바꿉니다.

  • PROJECT_ID: 프로젝트 ID입니다.
  • LOCATION: 리전입니다. 메모리 뱅크에서 지원되는 리전을 참조하세요.

ADK 런타임 만들기

에이전트가 메모리 뱅크를 ADK 메모리 서비스로 사용하도록 Agent Runtime ID를 런타임 또는 배포 스크립트에 전달합니다.

로컬 러너

adk.Runner 는 일반적으로 Colab과 같은 로컬 환경에서 사용됩니다. 이 경우 메모리 서비스와 러너를 직접 만들어야 합니다.

import asyncio

from google.adk.memory import VertexAiMemoryBankService
from google.adk.sessions import VertexAiSessionService
from google.genai import types

memory_service = VertexAiMemoryBankService(
    project="PROJECT_ID",
    location="LOCATION",
    agent_engine_id="AGENT_ENGINE_ID",
)

# You can use any ADK session service. This example uses Sessions.
session_service = VertexAiSessionService(
    project="PROJECT_ID",
    location="LOCATION",
    agent_engine_id="AGENT_ENGINE_ID",
)

runner = adk.Runner(
    agent=agent,
    app_name="APP_NAME",
    session_service=session_service,
    memory_service=memory_service
)

async def call_agent(query, session, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run_async(
    user_id=user_id, session_id=session, new_message=content)

  async for event in events:
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

다음을 바꿉니다.

  • PROJECT_ID: 프로젝트 ID입니다.
  • LOCATION: 리전입니다. 메모리 뱅크에서 지원되는 리전을 참조하세요.
  • APP_NAME: ADK 앱 이름입니다. 앱 이름은 생성된 메모리의 scope 딕셔너리에 포함되므로 메모리가 사용자와 앱 모두에서 격리됩니다.
  • AGENT_ENGINE_ID: 메모리 뱅크 및 Agent Platform 세션에 사용할 Agent Runtime ID입니다. 예를 들어 projects/my-project/locations/us-central1/reasoningEngines/456456입니다.

에이전트 런타임

Agent Runtime ADK 템플릿 (AdkApp)은 로컬 환경과 Agent Runtime 모두에서 ADK 에이전트를 배포하는 데 사용할 수 있습니다. Agent Platform에 배포되는 경우 Agent Runtime ADK 템플릿VertexAiMemoryBankService를 기본 메모리 서비스로 사용하며, 런타임과 동일한 런타임 인스턴스를 메모리 뱅크에 사용합니다. 따라서 한 단계로 메모리 뱅크 인스턴스를 만들고 런타임에 배포할 수 있습니다.

Agent Runtime 인스턴스 설정 및 메모리 뱅크 동작을 맞춤설정하는 방법을 포함하여 자세한 내용은 Agent Runtime 구성 을 참조하세요.

다음 코드를 사용하여 메모리 지원 ADK 에이전트를 Agent Runtime에 배포합니다.

import asyncio

import vertexai
from vertexai.agent_engines import AdkApp

client = vertexai.Client(
  project="PROJECT_ID",
  location="LOCATION"
)

adk_app = AdkApp(agent=agent)

# Create a new resource with your agent deployed to Agent Runtime.
# The Agent Runtime instance will also include an empty Memory Bank.
agent_engine = client.agent_engines.create(
      agent_engine=adk_app,
      config={
            "staging_bucket": "STAGING_BUCKET",
            "requirements": ["google-cloud-aiplatform[agent_engines,adk]"]
      }
)

# Alternatively, update an existing resource to deploy your agent to Agent Platform.
# Your agent will have access to the Runtime instance's existing memories.
agent_engine = client.agent_engines.update(
      name=agent_engine.api_resource.name,
      agent_engine=adk_app,
      config={
            "staging_bucket": "STAGING_BUCKET",
            "requirements": ["google-cloud-aiplatform[agent_engines,adk]"]
      }
)

async def call_agent(query, session_id, user_id):
    async for event in agent_engine.async_stream_query(
        user_id=user_id,
        session_id=session_id,
        message=query,
    ):
        print(event)

다음을 바꿉니다.

로컬에서 실행될 때 ADK 템플릿은 InMemoryMemoryService를 기본 메모리 서비스로 사용합니다. 하지만 기본 메모리 서비스를 재정의하여 VertexAiMemoryBankService를 사용할 수도 있습니다.

def memory_bank_service_builder():
    return VertexAiMemoryBankService(
        project="PROJECT_ID",
        location="LOCATION",
        agent_engine_id="AGENT_ENGINE_ID"
    )

adk_app = AdkApp(
      agent=adk_agent,
      # Override the default memory service.
      memory_service_builder=memory_bank_service_builder
)

async def call_agent(query, session_id, user_id):
  # adk_app is a local agent. If you want to deploy it to Agent Runtime,
  # use `client.agent_engines.create(...)` or `client.agent_engines.update(...)`
  # and call the returned Agent Runtime instance instead.
  async for event in adk_app.async_stream_query(
      user_id=user_id,
      session_id=session_id,
      message=query,
  ):
      print(event)

다음을 바꿉니다.

Cloud Run

Cloud Run에 에이전트를 배포하려면 ADK 문서의 안내를 참조하여 Cloud Run에 배포할 에이전트를 정의하는 방법을 알아보세요.

adk deploy cloud_run \
    ...
    --memory_service_uri=agentengine://AGENT_ENGINE_ID

GKE

Google Kubernetes Engine (GKE)에 에이전트를 배포하려면 ADK 문서의 안내를 참조하여 GKE에 배포할 에이전트를 정의하는 방법을 알아보세요.

adk deploy gke \
    ...
    --memory_service_uri=agentengine://AGENT_ENGINE_ID

ADK 웹

ADK 웹 인터페이스를 사용하면 브라우저에서 직접 에이전트를 테스트할 수 있습니다.

export GOOGLE_CLOUD_PROJECT="PROJECT_ID"
export GOOGLE_CLOUD_LOCATION="LOCATION"

adk web --memory_service_uri=agentengine://AGENT_ENGINE_ID

다음을 바꿉니다.

에이전트와 상호작용

에이전트를 정의하고 메모리 뱅크를 설정한 후에는 에이전트와 상호작용할 수 있습니다. 에이전트를 초기화할 때 메모리 생성을 트리거하는 콜백을 제공한 경우 에이전트가 호출될 때마다 메모리 생성이 트리거됩니다.

메모리는 에이전트를 실행하는 데 사용되는 사용자 ID 및 앱 이름에 해당하는 범위 {"user_id": USER_ID, "app_name": APP_NAME}를 사용하여 저장됩니다.

에이전트와 상호작용하는 방법은 실행 환경에 따라 다릅니다.

로컬 러너

# Use `asyncio.run(session_service.create(...))` if you're running this
# code as a standard Python script.
session = await session_service.create_session(
    app_name="APP_NAME",
    user_id="USER_ID"
)

# Use `asyncio.run(call_agent(...))` if you're running this code as a
# standard Python script.
await call_agent(
    "Can you fix the temperature?",
    session.id,
    "USER_ID"
)

다음을 바꿉니다.

  • APP_NAME: 러너의 앱 이름입니다.
  • USER_ID: 사용자의 식별자입니다. 이 세션에서 생성된 메모리는 이 불투명 식별자를 키로 사용하여 저장됩니다. 생성된 메모리의 범위는 {"user_id": "USER_ID"}로 저장됩니다.

에이전트 런타임

ADK 템플릿을 사용하는 경우 Agent Runtime을 호출하여 메모리 및 세션과 상호작용할 수 있습니다.

# Use `asyncio.run(agent_engine.async_create_session(...))` if you're
# running this code as a standard Python script.
session = await agent_engine.async_create_session(user_id="USER_ID")

# Use `asyncio.run(call_agent(...))` if you're running this code as a
# standard Python script.
await call_agent(
    "Can you fix the temperature?",
    session.get("id"),
    "USER_ID"
)

다음을 바꿉니다.

  • USER_ID: 사용자의 식별자입니다. 이 세션에서 생성된 메모리는 이 불투명 식별자를 키로 사용하여 저장됩니다. 생성된 메모리의 범위는 {"user_id": "USER_ID"}로 저장됩니다.

Cloud Run

ADK Cloud Run 배포 문서의 에이전트 테스트 섹션을 참조하세요.

GKE

ADK GKE 배포 문서의 에이전트 테스트 섹션을 참조하세요.

ADK 웹

ADK 웹을 사용하려면 http://localhost:8000에서 로컬 서버로 이동합니다.

기본적으로 ADK 웹은 사용자 ID를 user로 설정합니다. 기본 사용자 ID를 재정의하려면 쿼리 매개변수에 userId를 포함합니다. 예: http://localhost:8000?userId=YOUR_USER_ID.

자세한 내용은 ADK 문서의 ADK 웹 페이지를 참조하세요.

상호작용 예시

첫 번째 세션

PreloadMemoryTool을 사용한 경우 에이전트는 각 턴의 시작 시점에 메모리를 불러와 사용자가 이전에 에이전트에 전달한 선호 정보를 확인합니다. 에이전트가 사용자와 처음 상호작용하는 동안에는 검색할 수 있는 메모리가 없습니다. 따라서 에이전트는 다음 예와 같이 사용자의 선호도(예: 선호하는 온도)를 알지 못합니다.

  1. 첫 번째 턴:

    • 사용자: "온도를 수정해 줘."

    • (도구 호출): ADK가 메모리를 가져오려고 시도하지만 사용 가능한 메모리가 없습니다.

    • 모델: "선호하는 온도는 몇 도야?"

    • (콜백): ADK가 메모리 생성을 트리거합니다. 메모리가 추출되지 않습니다.

  2. 두 번째 턴:

    • 사용자: 22도가 편안해.

    • (도구 호출): ADK가 메모리를 가져오려고 시도하지만 사용 가능한 메모리가 없습니다.

    • 모델: 알겠습니다. 온도를 22도로 업데이트했습니다.

    • (콜백): ADK가 메모리 생성을 트리거합니다. '온도 22도가 좋아' 메모리가 생성됩니다.

두 번째 세션

추출된 메모리는 동일한 앱 이름과 사용자 ID의 다음 세션에서 사용할 수 있습니다. 사용자가 기존 메모리와 유사하거나 모순되는 정보를 제공하는 경우 새 정보가 기존 메모리와 통합됩니다.

  1. 첫 번째 턴

    • 사용자: 온도를 수정해 줘. 너무 불편해!

    • (도구 호출): ADK가 메모리를 가져오려고 시도합니다. '온도 22도가 좋아' 메모리가 검색됩니다.

    • 모델: 알겠습니다. 온도를 22도로 업데이트했습니다.

    • (콜백): ADK가 메모리 생성을 트리거합니다. 사용자가 유지할 만한 의미 있는 정보를 공유하지 않았으므로 메모리가 추출되지 않습니다.

  2. 두 번째 턴

    • 사용자: 아침에는 더 따뜻한 게 좋아.

    • (도구 호출): ADK가 메모리를 가져오려고 시도합니다. '온도 22도가 좋아' 메모리가 검색됩니다.

    • 모델: 알겠습니다. 온도를 더 따뜻하게 했습니다.

    • (콜백): ADK가 메모리 생성을 트리거합니다. 기존 메모리 '온도 22도가 좋아'가 '일반적으로 온도가 22도인 것을 좋아하지만 아침에는 더 따뜻한 게 좋아'로 업데이트됩니다.

리전 런타임으로 멀티 리전 메모리 뱅크 사용

기본적으로 에이전트와 메모리 뱅크는 동일한 리전에 배포됩니다. 하지만 리전 런타임 (예: us-central1)으로 멀티 리전 메모리 뱅크 (예: us)를 사용하도록 분리할 수 있습니다. 이 구성을 사용하면 여러 리전 배포에서 중앙 메모리 뱅크를 유지할 수 있습니다.

멀티 리전 메모리 뱅크를 사용하려면 기본 ADK 메모리 서비스 빌더를 재정의하여 멀티 리전 위치와 해당 런타임 ID를 가리켜야 합니다.

import vertexai
from google.adk.memory import VertexAiMemoryBankService
from vertexai.agent_engines import AdkApp

# Create the Memory Bank store in a multi-region location (for example, 'us')
client_mb = vertexai.Client(project="PROJECT_ID", location="us")
mb_engine = client_mb.agent_engines.create()
mb_engine_id = mb_engine.api_resource.name.split(\"/\")[-1]


# Point your memory service to the 'us' location, 'us' Memory Bank
def memory_bank_service_builder():
    return VertexAiMemoryBankService(
        project="PROJECT_ID",
        location="us",
        agent_engine_id=mb_engine_id
    )

# Create the AdkApp with the overridden builder
adk_app = AdkApp(
    agent=agent,
    memory_service_builder=memory_bank_service_builder
)

# Deploy the runtime to a specific region (for example, 'us-central1')
client_runtime = vertexai.Client(project="PROJECT_ID", location="us-central1")
agent_engine = client_runtime.agent_engines.create(
    agent=adk_app,
    config={
        "staging_bucket": "STAGING_BUCKET",
        "requirements": ["google-cloud-aiplatform[agent_engines,adk]"]
    }
)

다음을 바꿉니다.

  • PROJECT_ID: 프로젝트 ID입니다.
  • STAGING_BUCKET: Agent Runtime을 스테이징하는 데 사용할 Cloud Storage 버킷입니다.

정리

이 프로젝트에 사용된 모든 리소스를 정리하려면 빠른 시작에 사용한 프로젝트 Google Cloud 를 삭제하면 됩니다.

또는 다음과 같이 이 튜토리얼에서 만든 개별 리소스를 삭제하면 됩니다.

  1. 다음 코드 샘플을 사용하여 Agent Runtime 인스턴스를 삭제합니다. 그러면 해당 런타임에 속한 세션 또는 메모리도 삭제됩니다.

    agent_engine.delete(force=True)
    
  2. 로컬에서 만든 파일을 삭제합니다.

다음 단계

빠른 시작

메모리 뱅크 API를 시작하여 장기 메모리를 관리합니다.