에이전트 개발 키트 에이전트 사용

시작하기 전에

이 튜토리얼에서는 사용자가 다음 안내를 읽고 따랐다고 가정합니다.

• 에이전트 개발 키트 에이전트 개발: agent를 AdkApp의 인스턴스로 개발합니다.
• 사용자 인증: 에이전트 쿼리를 위해 사용자로 인증을 수행합니다.
• SDK 가져오기 및 초기화: 필요한 경우 배포된 인스턴스를 가져올 수 있도록 클라이언트를 초기화합니다.

에이전트 인스턴스 가져오기

AdkApp를 쿼리하려면 먼저 새 인스턴스를 만들거나 기존 인스턴스를 가져와야 합니다.

특정 리소스 ID에 해당하는 AdkApp를 가져오려면 다음 안내를 따르세요.

Python용 Vertex AI SDK

다음 코드를 실행합니다.

import vertexai

client = vertexai.Client(  # For service interactions via client.agent_engines
    project="PROJECT_ID",
    location="LOCATION",
)

adk_app = client.agent_engines.get(name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

print(adk_app)

각 항목의 의미는 다음과 같습니다.

• PROJECT_ID는 에이전트를 개발하고 배포하는 데 사용되는 Google Cloud 프로젝트 ID입니다.
• LOCATION은 지원되는 리전 중 하나이며
• RESOURCE_ID는 배포된 에이전트의 ID이며 reasoningEngine 리소스로 등록되어 있습니다.

Python 요청 라이브러리

다음 코드를 실행합니다.

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

response = requests.get(
f"https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    headers={
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": f"Bearer {get_identity_token()}",
    },
)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID

Python용 Vertex AI SDK를 사용하는 경우 adk_app 객체는 다음을 포함하는 AgentEngine 클래스에 해당합니다.

• 배포된 에이전트에 관한 정보가 포함된 adk_app.api_resource adk_app.operation_schemas()를 호출하여 adk_app에서 지원하는 작업 목록을 반환할 수도 있습니다. 자세한 내용은 지원되는 작업을 참고하세요.
• 동기 서비스 상호작용을 허용하는 adk_app.api_client
• 비동기 서비스 상호작용을 허용하는 adk_app.async_api_client

이 섹션의 나머지 부분에서는 adk_app라는 이름의 AgentEngine 인스턴스가 있다고 가정합니다.

지원되는 작업

AdkApp에 지원되는 작업은 다음과 같습니다.

• async_stream_query: 쿼리에 대한 응답을 스트리밍합니다.

• async_create_session: 새 세션을 만드는 데 사용됩니다.

• async_list_sessions: 사용 가능한 세션을 나열합니다.

• async_get_session: 특정 세션을 검색합니다.

• async_delete_session: 특정 세션을 삭제합니다.

• async_add_session_to_memory: 세션의 메모리를 생성합니다.

• async_search_memory: 메모리를 가져오는 데 사용됩니다.

지원되는 모든 작업을 나열하려면 다음을 실행합니다.

Python용 Vertex AI SDK

다음 코드를 실행합니다.

adk_app.operation_schemas()

Python 요청 라이브러리

다음 코드를 실행합니다.

import json

json.loads(response.content).get("spec").get("classMethods")

REST API

curl 요청에 대한 응답에서 spec.class_methods에 표시됩니다.

세션 관리

AdkApp은 에이전트를 Vertex AI Agent Engine에 배포한 후 클라우드 기반 관리 세션을 사용합니다. 이 섹션에서는 관리 세션을 사용하는 방법을 설명합니다.

참고: 이 안내는 에이전트를 개발할 때 session_service_builder를 사용하여 데이터베이스를 맞춤설정하지 않았다고 가정합니다.

세션 만들기

사용자의 세션을 만들려면 다음 단계를 따르세요.

Python용 Vertex AI SDK

session = await adk_app.async_create_session(user_id="USER_ID")

print(session)

Python 요청 라이브러리

다음 코드를 실행합니다.

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_create_session",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_create_session", "input": {"user_id": "USER_ID"},}'

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

세션은 ADK 세션 객체의 사전 표현으로 생성됩니다.

세션 나열

사용자의 세션을 나열하려면 다음을 실행하세요.

Python용 Vertex AI SDK

response = await adk_app.async_list_sessions(user_id="USER_ID"):
for session in response.sessions:
    print(session)

Python 요청 라이브러리

다음 코드를 실행합니다.

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_list_sessions",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_list_sessions", "input": {"user_id": "USER_ID"},}'

여기서 USER_ID는 정의한 사용자 ID입니다. 예를 들면 user-123입니다.

세션이 반환되면 ADK 세션 객체의 사전 형식을 사용합니다.

세션 가져오기

특정 세션을 가져오려면 사용자 ID와 세션 ID가 모두 필요합니다.

Python용 Vertex AI SDK

session = await adk_app.async_get_session(user_id="USER_ID", session_id="SESSION_ID")

print(session)

Python 요청 라이브러리

다음 코드를 실행합니다.

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_get_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_get_session", "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},}'

session는 ADK 세션 객체의 사전 표현입니다.

세션 삭제

세션을 삭제하려면 사용자 ID와 세션 ID가 모두 필요합니다.

Python용 Vertex AI SDK

await adk_app.async_delete_session(user_id="USER_ID", session_id="SESSION_ID")

Python 요청 라이브러리

다음 코드를 실행합니다.

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_delete_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_delete_session", "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},}'

쿼리에 대한 응답 스트리밍

세션에서 에이전트의 응답을 스트리밍하려면 다음 단계를 따르세요.

Python용 Vertex AI SDK

async for event in adk_app.async_stream_query(
    user_id="USER_ID",
    session_id="SESSION_ID",  # Optional
    message="What is the exchange rate from US dollars to SEK today?",
):
  print(event)

Python 요청 라이브러리

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

requests.post(
    f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:streamQuery",
    headers={
        "Content-Type": "application/json",
        "Authorization": f"Bearer {get_identity_token()}",
    },
    data=json.dumps({
        "class_method": "async_stream_query",
        "input": {
            "user_id": "USER_ID",
            "session_id": "SESSION_ID",
            "message": "What is the exchange rate from US dollars to SEK today?",
        },
    }),
    stream=True,
)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:streamQuery?alt=sse -d '{
  "class_method": "async_stream_query",
  "input": {
    "user_id": "USER_ID",
    "session_id": "SESSION_ID",
    "message": "What is the exchange rate from US dollars to SEK today?",
  }
}'

참고: session_id= 인수는 선택사항입니다. 지정하지 않으면 새 세션이 자동으로 생성되어 해당 쿼리를 서빙하는 데 사용됩니다.

Vertex AI SDK for Python을 사용하는 경우 다음과 같은 사전 시퀀스와 같이 대화가 계속됩니다.

{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_call': {'args': {'currency_date': '2025-04-03',
                                                   'currency_from': 'USD',
                                                   'currency_to': 'SEK'},
                                          'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                          'name': 'get_exchange_rate'}}],
             'role': 'model'},
 'id': 'bOPHtzji',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_response': {'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                              'name': 'get_exchange_rate',
                                              'response': {'amount': 1.0,
                                                           'base': 'USD',
                                                           'date': '2025-04-03',
                                                           'rates': {'SEK': 9.6607}}}}],
             'role': 'user'},
 'id': '9AoDFmiL',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'text': 'The exchange rate from USD to SEK on '
                                '2025-04-03 is 1 USD to 9.6607 SEK.'}],
             'role': 'model'},
 'id': 'hmle7trT',
 # ...
}

추억 관리

AdkApp은 에이전트 정의에 PreloadMemoryTool을 포함하고 Vertex AI Agent Engine에 에이전트를 배포하는 경우 Vertex AI Agent Engine 메모리 뱅크를 사용합니다. 이 섹션에서는 ADK 메모리 서비스의 기본 구현을 통해 에이전트에서 메모리를 생성하고 검색하는 방법을 설명합니다.

세션을 메모리에 추가

의미 있는 정보의 메모리를 세션에 유지하여 향후 세션에서 사용할 수 있도록 하려면 async_add_session_to_memory 메서드를 사용하세요.

Python용 Vertex AI SDK

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

여기서 SESSION_DICT은 ADK 세션 객체의 사전 형식입니다.

추억 검색

에이전트의 메모리를 검색하려면 async_search_memory 메서드를 사용하면 됩니다.

Python용 Vertex AI SDK

response = await adk_app.async_search_memory(
    user_id="USER_ID",
    query="QUERY",
)
print(response)

각 항목의 의미는 다음과 같습니다.

• USER_ID은 관련 메모리의 범위입니다.
• QUERY은 유사 검색을 실행할 쿼리입니다.

다음 단계

• 에이전트 사용
• 에이전트 평가
• 배포된 에이전트 관리
• 지원 받기