에이전트 개발 키트로 세션 관리

이 페이지에서는 에이전트 개발 키트 (ADK) 에이전트를 Agent Platform 세션에 연결하고 로컬 및 프로덕션 환경에서 관리형 세션을 사용하는 방법을 설명합니다.

시작하기 전에

이 안내에서는 ADK 에이전트와 지원 러너 및 배포 코드를 정의하기 위해 다음과 같은 기본 프로젝트 파일 구조를 사용합니다.

my_agent/
    agent.py      # main agent code
    runner.py     # code for interacting with the agent
    deploy.py     # code for deploying the agent to Google Cloud

환경 설정의 필요한 역할 가져오기 및 인증 단계를 수행하여 환경이 설정되었는지 확인합니다.

환경 변수 설정하기

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

import os

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

다음을 바꿉니다.

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

Vertex AI Agent Engine 인스턴스 만들기

Agent Platform 세션에 액세스하려면 먼저 Vertex AI Agent Engine 인스턴스를 사용해야 합니다. 세션을 사용하기 위해 코드를 배포할 필요가 없습니다. 이전에 Agent Engine을 사용한 적이 있다면 코드를 배포하지 않고 Vertex AI Agent Engine 인스턴스를 만드는 데 몇 초밖에 걸리지 않습니다. Agent Engine을 처음 사용하는 경우 시간이 더 걸릴 수 있습니다.

Google Cloud 프로젝트 

import vertexai

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

# If you don't have an Agent Engine instance already, create an instance.
agent_engine = client.agent_engines.create()

# Print the agent engine ID, you will need it in the later steps to initialize
# the ADK `VertexAiSessionService`.
print(agent_engine.api_resource.name.split("/")[-1])

다음을 바꿉니다.

  • PROJECT_ID: 프로젝트 ID입니다.

  • LOCATION: 리전 세션에서 지원되는 리전을 참고하세요.

ADK 에이전트 개발

ADK 에이전트를 만들려면 에이전트 개발 키트의 안내를 따르거나 다음 코드를 사용하여 고정 인사말로 사용자를 맞이하는 에이전트를 만듭니다. 이 코드를 agent.py 파일에 저장합니다.

# file: my_agent/agent.py
from google import adk

def greetings(query: str):
  """Tool to greet user."""
  if 'hello' in query.lower():
    return {"greeting": "Hello, world"}
  else:
    return {"greeting": "Goodbye, world"}

# Define an ADK agent
root_agent = adk.Agent(
    model="gemini-2.0-flash",
    name='my_agent',
    instruction="You are an Agent that greet users, always use greetings tool to respond.",
    tools=[greetings]
)

ADK 실행기 설정

ADK 런타임은 에이전트, 도구, 콜백의 실행을 조정하고 세션 읽기 및 쓰기 호출을 조정합니다. Agent Platform 세션과 연결되는 VertexAiSessionService로 러너를 초기화합니다. 이 코드를 runner.py 파일에 저장합니다.

Google Cloud 프로젝트 

# file: my_agent/runner.py
import agent # Import from your agent.py
from google.adk import Runner
from google.adk.sessions import VertexAiSessionService
from google.genai import types

app_name="APP_NAME"
user_id="USER_ID"

# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
      project="PROJECT_ID",
      location="LOCATION",
      agent_engine_id="AGENT_ENGINE_ID"
)
runner = Runner(
    agent=agent.root_agent,
    app_name=app_name,
    session_service=session_service)

# Helper method to send query to the runner
async def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  async for event in runner.run_async(
      user_id=user_id, session_id=session_id, new_message=content):
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

다음을 바꿉니다.

  • APP_NAME: 에이전트 애플리케이션의 이름입니다.

  • USER_ID: 128자(영문 기준)로 제한된 사용자 ID를 선택합니다. 예를 들면 user-123입니다.

  • AGENT_ENGINE_ID: Vertex AI Agent Engine 인스턴스의 리소스 ID입니다.

  • 배포된 에이전트의 경우 리소스 ID가 GOOGLE_CLOUD_AGENT_ENGINE_ID 환경 변수로 표시됩니다.

  • 로컬 에이전트의 경우 agent_engine.api_resource.name.split("/")[-1]을 사용하여 리소스 ID를 검색할 수 있습니다.

에이전트와 상호작용

에이전트를 정의하고 Agent Platform 세션을 설정한 후 에이전트와 상호작용하여 세션 기록과 상태가 유지되는지 확인할 수 있습니다.

ADK UI

ADK 사용자 인터페이스로 에이전트를 테스트하고 Agent Platform 세션에 연결하려면 터미널에서 adk web 명령어를 실행하세요.

adk web는 일반적으로 .env 파일에서 환경 변수를 읽지만 --session_service_uri 플래그를 사용하면 이 파일을 무시합니다. 대신 명령어를 실행하기 전에 터미널 세션에서 GOOGLE_CLOUD_PROJECT 및 GOOGLE_CLOUD_LOCATION 환경 변수를 설정해야 합니다.

export GOOGLE_CLOUD_PROJECT="PROJECT_ID"
export GOOGLE_CLOUD_LOCATION="LOCATION"
adk web --session_service_uri=agentengine://AGENT_ENGINE_ID
# Sample output
+-----------------------------------------------------------------------------+
| ADK Web Server started                                                      |
|                                                                             |
| For local testing, access at http://localhost:8000.                         |
+-----------------------------------------------------------------------------+

INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

ADK UI

Python

ADK Python 코드를 사용하여 세션과 상태를 관리합니다. runner.py 파일 끝에 다음 코드를 추가하여 에이전트와 상호작용합니다.

다음 스니펫에는 간결성을 위해 최상위 await 호출이 포함되어 있습니다. 이 코드를 Python 스크립트로 실행하려면 스니펫을 async 함수 내에 배치하고 asyncio.run()를 사용하여 실행합니다(예 참고).

import asyncio

async def main():
  # Place one or more snippets here.
  # For example:
  session = await session_service.create_session(
         app_name=app_name,
         user_id=user_id)

  await call_agent("Hello!", session.id, user_id)

asyncio.run(main())

세션 만들기 및 에이전트 쿼리

다음 코드를 사용하여 세션을 만들고 에이전트에 쿼리를 보냅니다. 자체 명명 규칙을 사용하여 세션을 정리하거나 다른 서비스에서 빠르게 전환하려면 맞춤 세션 ID를 지정하세요. 세션 ID를 지정하지 않으면 Agent Platform에서 자동으로 생성합니다.

# file: my_agent/runner.py
# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id,
       session_id=session_id)

await call_agent("Hello!", session.id, user_id)
# Agent response: "Hello, world"

await call_agent("Thanks!", session.id, user_id)
# Agent response: "Goodbye, world"

커스텀 세션 ID의 경우 시스템 생성 ID와의 충돌을 방지하기 위해 다음 제한사항을 고려하세요.

  • 첫 번째 문자가 영문자인 경우 ID의 길이는 최대 63자일 수 있습니다. 유효한 문자는 소문자, 숫자, 하이픈([a-z0-9-])입니다. 마지막 문자는 영문자 또는 숫자여야 합니다.
  • 첫 번째 문자가 숫자인 경우 ID는 최대 9자까지 가능합니다. 유효한 문자는 앞에 0이 없는 숫자 ([0-9])입니다.

세션이 생성되고 러너에게 전달되면 ADK는 세션을 사용하여 현재 상호작용의 이벤트를 저장합니다. 해당 세션의 ID를 제공하여 이전 세션을 재개할 수도 있습니다.

세션 TTL (수명) 구성

모든 세션에는 만료 시간이 있어야 합니다. 세션을 만들거나 업데이트할 때 이 만료 시간을 정의할 수 있습니다. 세션과 하위 이벤트는 만료 시간이 지나면 자동으로 삭제됩니다. 만료 시간 (expire_time)을 직접 설정하거나 수명 (ttl)을 초 단위로 설정할 수 있습니다. 둘 다 지정되지 않은 경우 시스템은 기본 TTL인 365일을 적용합니다.

TTL(수명)

TTL을 설정하면 서버에서 새로 생성된 세션의 만료 시간을 create_time + ttl로, 업데이트된 세션의 만료 시간을 update_time + ttl로 계산합니다.

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted 10 days after creation time.
        ttl=f"{24 * 60 * 60 * 10}s"
  )
  ```

만료 시간 

  import datetime

  expire_time = datetime.datetime.now(
        tz=datetime.timezone.utc) + datetime.timedelta(seconds=24 * 60 * 60 * 10)

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted at the provided time (10 days after current time).
        expire_time=expire_time.isoformat()
  )

기존 세션 나열

지정된 사용자 ID와 연결된 모든 기존 세션을 표시합니다.

# List sessions
sessions = await session_service.list_sessions(app_name=app_name,user_id=user_id)
print(sessions)
# ListSessionsResponse(session_ids=['1122334455', '9988776655'])

세션 상태 관리

상태는 에이전트가 대화하는 데 필요한 정보를 보유합니다. 세션을 만들 때 초기 상태를 사전으로 제공할 수 있습니다.

# Create a session with state
session = await session_service.create_session(
    app_name=app_name,
    user_id=user_id,
    state={'key': 'value'})

print(session.state['key'])
# value

실행기 외부에서 세션 상태를 업데이트하려면 state_delta를 사용하여 새 이벤트를 세션에 추가합니다.

# file: my_agent/runner.py
from google.adk.events import Event, EventActions
import time

# Define state changes
state_changes = {'key': 'new_value'}

# Create event with actions
actions_with_update = EventActions(state_delta=state_changes)
system_event = Event(
invocation_id="invocation_id",
author="system", # Or 'agent', 'tool' etc.
actions=actions_with_update,
timestamp=time.time()
)

# Append the event
await session_service.append_event(session, system_event)

# Check updated state
updated_session = await session_service.get_session(
    app_name=app_name,
    user_id=user_id,
    session_id=session.id)
# State is updated to new value
print(updated_session.state['key'])
# new_value

이벤트의 맞춤 입력란

ADK는 유연한 이벤트 스키마를 지원하므로 세션 이벤트에 임의의 데이터를 포함할 수 있습니다. 이는 다른 에이전트 프레임워크와의 상호 운용성이나 맞춤 이벤트 데이터를 저장하는 데 유용합니다.

맞춤 필드가 있는 이벤트를 추가하면 ADK가 이벤트 데이터를 저장된 세션 이벤트의 raw_event 필드로 직렬화합니다.

# Create an event with custom fields
custom_event = Event(
  invocation_id="invocation_id",
  author="user",
  timestamp=time.time(),
  custom_field="custom_value",
  another_field={"nested_key": "nested_value"}
)

# Append the event
await session_service.append_event(session, custom_event)

세션을 가져오면 가져온 이벤트는 이러한 맞춤 필드를 유지합니다.

세션 삭제

사용자 ID와 연결된 특정 세션을 삭제하려면 다음 안내를 따르세요.

await session_service.delete_session(app_name=app_name, user_id=user_id, session_id=session.id)

에이전트를 Agent Platform에 배포

로컬에서 에이전트를 테스트한 후 Agent Platform 인스턴스를 파라미터로 업데이트하여 에이전트를 프로덕션에 배포할 수 있습니다.

Google Cloud 프로젝트

client.agent_engines.update(
    resource_name=agent_engine.api_resource.name,
    agent=AGENT,
    config={
      "display_name": DISPLAY_NAME,      # Optional.
      "requirements": REQUIREMENTS,      # Optional.
      "staging_bucket": STAGING_BUCKET,  # Required.
    },
)

다음을 바꿉니다.

  • AGENT: query / stream_query 메서드를 구현하는 애플리케이션입니다(예: ADK 에이전트의 경우 AdkApp). 자세한 내용은 배포 고려사항을 참조하세요.

  • DISPLAY_NAME: 에이전트의 사용자 친화적인 이름입니다.

  • REQUIREMENTS: 에이전트에 필요한 pip 패키지 목록입니다. 예를 들면 ["google-cloud-storage", "google-cloud-aiplatform[agent_engines,adk]"]입니다.

  • STAGING_BUCKET: gs://로 시작하는 Cloud Storage 버킷입니다.

삭제

이 프로젝트에 사용된 모든 리소스를 삭제하려면 Vertex AI Agent Engine 인스턴스를 하위 리소스와 함께 삭제하면 됩니다.

agent_engine.delete(force=True)

다음 단계