Google은 AI 기술을 사용하여 콘텐츠를 사용자의 기본 언어로 번역합니다. AI 번역에는 오류가 있을 수 있습니다.

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

시작하기 전에

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

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

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

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

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

Agent Platform 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

Agent Platform 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: 메모리를 가져옵니다.

지원되는 모든 작업을 나열하려면 다음 안내를 따르세요.

Agent Platform SDK

다음 코드를 실행합니다.

adk_app.operation_schemas()

Python 요청 라이브러리

다음 코드를 실행합니다.

import json

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

REST API

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

세션 관리

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

세션 만들기

사용자의 세션을 만들려면 AdkApp.async_create_session 메서드를 사용합니다.

Agent Platform 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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 세션 객체의 딕셔너리 표현으로 생성됩니다.

세션 나열

사용자의 세션을 나열하려면 AdkApp.async_list_sessions 메서드를 사용합니다.

Agent Platform 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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 세션 객체의 딕셔너리 형식을 사용합니다.

세션 가져오기

특정 세션을 가져오려면 AdkApp.async_get_session 메서드를 사용합니다.

Agent Platform 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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 세션 객체의 딕셔너리 표현입니다.

세션 삭제

세션을 삭제하려면 AdkApp.async_delete_session 메서드를 사용합니다.

Agent Platform 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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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"},}'

쿼리에 대한 응답 스트리밍

세션의 에이전트에서 응답을 스트리밍하려면 AdkApp.async_stream_query 메서드를 사용합니다.

Agent Platform 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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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?",
  }
}'

Agent Platform SDK를 사용하는 경우 다음과 같은 딕셔너리 시퀀스와 같이 대화가 계속됩니다.

{'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',
 # ...
}

장기 실행 쿼리 작업

완료하는 데 오랜 시간이 걸릴 수 있는 쿼리 (최대 7일)의 경우 장기 실행 작업으로 실행할 수 있습니다. 이러한 작업은 비동기식으로 실행됩니다. 나중에 작업 상태를 확인하고 결과를 가져올 수 있습니다.

비동기 쿼리용 에이전트 배포

에이전트를 배포하려면 에이전트 배포의 일반 안내를 따르세요. 소스 기반 배포의 경우 deploymentSpec.agentFramework 필드를 google-adk로 설정합니다.

자체 컨테이너 이미지를 빌드하여 커스텀 API 엔드포인트를 사용하는 경우 SDK를 사용하여 에이전트를 만들 때 다음 환경 변수를 추가해야 합니다.

"env_vars" = {
    "API_ENDPOINT_PREFIX": "/api/myendpoint"
}

장기 실행 쿼리 작업 시작

전제 조건으로 서비스 에이전트 service-PROJECT_NUMBER@gcp-sa-aiplatform-re.iam.gserviceaccount.com에 출력 파일의 스토리지 버킷에 대한 roles/storage.objectCreator 역할을 부여해야 합니다.

장기 실행 쿼리 작업을 시작하려면 다음 안내를 따르세요.

Agent Platform SDK

import vertexai

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

response = client.agent_engines.run_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    config={
        "query": '{"input":{"user_id":"USER_ID", "message":"What is the exchange rate from US dollars to SEK today?"}}',
        "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE",
    },
)
print(response)

SDK를 사용하면 output_gcs_uri가 디렉터리 또는 파일 이름일 수 있습니다. 파일 이름인 경우 시스템은 이 파일을 사용하여 응답을 저장합니다. 디렉터리인 경우 시스템은 응답 파일을 자동으로 생성합니다. 두 경우 모두 입력 쿼리는 출력 파일과 동일한 파일 이름 접두사가 있는 동일한 디렉터리에 저장됩니다.

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1beta1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:asyncQuery -d \
'{
  "input_gcs_uri": "gs://GCS_BUCKET_NAME/INPUT_FILE",
  "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE"
}'

REST API 호출의 경우 input_gcs_uri 필드는 쿼리가 포함된 파일을 가리켜야 합니다. 파일의 콘텐츠는 QueryReasoningEngineRequest의 input 필드와 일치하는 input 필드가 있는 JSON 객체여야 합니다 (예: { "input": { "user_id": "hello", "message":"$QUERY"} }). 이 입력 파일이 출력 위치와 다른 버킷에 있는 경우 서비스 에이전트 service-PROJECT_NUMBER@gcp-sa-aiplatform-re.iam.gserviceaccount.com에 입력 파일이 있는 스토리지 버킷에 대한 roles/storage.objectReader 역할도 부여해야 합니다.

output_gcs_uri는 파일 이름이어야 합니다.

장기 실행 쿼리 작업의 상태 확인

장기 실행 쿼리 작업의 상태를 확인하고 결과를 가져오려면 다음 안내를 따르세요.

Agent Platform SDK

response = client.agent_engines.check_query_job(
    name="JOB_NAME",
    config={
        "retrieve_result": True,
    },
)
print(response)

장기 실행 쿼리 작업 취소

장기 실행 쿼리 작업을 취소하려면 장기 실행 쿼리 작업에서 반환된 LRO 리소스 이름이 있어야 합니다.

Agent Platform SDK

response = client.agent_engines.cancel_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    operation_name="projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
)

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1beta1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:cancelAsyncQuery -d \
'{
  "name": "projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
  "operation_name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID"
}'

메모리 관리

AdkApp은(는) 에이전트 정의에 PreloadMemoryTool을(를) 포함하고 에이전트를 Agent Platform에 배포하면 메모리 뱅크를 사용합니다. 이 섹션 에서는 ADK 메모리 서비스의 기본 구현을 통해 에이전트에서 메모리를 생성하고 가져오는 방법을 설명합니다.

메모리에 세션 추가

세션에서 의미 있는 정보의 메모리를 보관하려면 (향후 세션에서 사용할 수 있음) async_add_session_to_memory 메서드를 사용합니다.

Agent Platform SDK

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

여기서 SESSION_DICT는 ADK 세션 객체의 딕셔너리 형식입니다.

메모리 검색

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

Agent Platform SDK

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

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

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

에이전트 개발 키트 에이전트 사용 컬렉션을 사용해 정리하기 내 환경설정을 기준으로 콘텐츠를 저장하고 분류하세요.

시작하기 전에

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

Agent Platform SDK

Python 요청 라이브러리

REST API

지원되는 작업

Agent Platform SDK

Python 요청 라이브러리

REST API

세션 관리

세션 만들기

Agent Platform SDK

Python 요청 라이브러리

REST API

세션 나열

Agent Platform SDK

Python 요청 라이브러리

REST API

세션 가져오기

Agent Platform SDK

Python 요청 라이브러리

REST API

세션 삭제

Agent Platform SDK

Python 요청 라이브러리

REST API

쿼리에 대한 응답 스트리밍

Agent Platform SDK

Python 요청 라이브러리

REST API

장기 실행 쿼리 작업

비동기 쿼리용 에이전트 배포

장기 실행 쿼리 작업 시작

Agent Platform SDK

REST

장기 실행 쿼리 작업의 상태 확인

Agent Platform SDK

장기 실행 쿼리 작업 취소

Agent Platform SDK

REST

메모리 관리

메모리에 세션 추가

Agent Platform SDK

메모리 검색

Agent Platform SDK

다음 단계

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