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

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

시작하기 전에

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

Vertex AI Agent Engine 인스턴스 만들기

Vertex AI Agent Engine 세션에 액세스하려면 먼저 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: 리전 세션에서 지원되는 리전을 참고하세요.

Vertex AI Express 모드

익스프레스 모드 API 키와 함께 Vertex AI SDK for Python을 사용할 수 있습니다.

import vertexai

client = vertexai.Client(  # For service interactions via client.agent_engines
    api_key="API_KEY",
)

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

ADK 에이전트 개발

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

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 런타임은 에이전트, 도구, 콜백의 실행을 조정하고 세션 읽기 및 쓰기 호출을 조정합니다. Vertex AI Agent Engine 세션과 연결되는 VertexAiSessionService로 실행기를 초기화합니다.

Google Cloud 프로젝트 

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_ID",
      "LOCATION",
      "AGENT_ENGINE_ID"
)
runner = adk.Runner(
    agent=root_agent,
    app_name=app_name,
    session_service=session_service)

# Helper method to send query to the runner
def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run(
      user_id=user_id, session_id=session_id, new_message=content)

  for event in events:
      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를 검색할 수 있습니다.

Vertex AI Express 모드 

import os
from google.adk.sessions import VertexAiSessionService
from google.genai import types

# Set environment variables
os.environ["GOOGLE_API_KEY"] = "API_KEY"
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"

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

# Helper method to send query to the runner
def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run(
      user_id=user_id, session_id=session_id, new_message=content)

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

다음을 바꿉니다.

  • API_KEY: Vertex AI Express Mode API 키입니다.

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

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

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

에이전트와 상호작용

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

ADK UI

ADK 사용자 인터페이스로 에이전트를 테스트하고 session_service_uri 명령줄 옵션을 사용하여 Vertex AI Agent Engine 세션에 연결합니다.

agent_engine_id="AGENT_ENGINE_ID"

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 코드를 사용하여 세션과 상태를 관리합니다.

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

다음 코드를 사용하여 세션을 만들고 에이전트에 쿼리를 보냅니다.

# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id)

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

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

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

기존 세션 나열

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

# List sessions
await session_service.list_sessions(app_name=app_name,user_id=user_id)

# 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를 사용하여 새 이벤트를 세션에 추가합니다.

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

세션 삭제

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

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

Vertex AI Agent Engine에 에이전트 배포

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

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 Express 모드

이 배포 단계에 따라 소스 파일을 통해 에이전트를 배포합니다.

삭제

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

agent_engine.delete(force=True)

다음 단계