Utilizzare un agente Agent Development Kit

Oltre alle istruzioni generali per l'utilizzo di un agente, questa pagina descrive le funzionalità specifiche di AdkApp.

Prima di iniziare

Questo tutorial presuppone che tu abbia letto e seguito le istruzioni riportate in:

Per eseguire query su un'applicazione ADK, devi prima creare una nuova istanza dell'applicazione ADK o ottenere un'istanza esistente.

Per ottenere l'applicazione ADK corrispondente a un ID risorsa specifico:

SDK Vertex AI per Python

Esegui questo codice:

from vertexai import agent_engines

adk_app = agent_engines.get(RESOURCE_ID)

In alternativa, puoi fornire il nome completo della risorsa dell'agente:

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

Libreria delle richieste Python

Esegui questo codice:

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()}",
    },
)

API REST 

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

Operazioni supportate

Per AdkApp sono supportate le seguenti operazioni:

Per elencare tutte le operazioni supportate:

SDK Vertex AI per Python

Esegui questo codice:

adk_app.operation_schemas()

Libreria delle richieste Python

Esegui questo codice:

import json

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

API REST

Rappresentato in spec.class_methods dalla risposta alla richiesta cURL.

Gestire le sessioni

AdkApp utilizza sessioni gestite basate sul cloud dopo il deployment dell'agente in Vertex AI Agent Engine. Questa sezione descrive come utilizzare le sessioni gestite.

Creare una sessione

Per creare una sessione per un utente:

SDK Vertex AI per Python 

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

Libreria delle richieste Python

Esegui questo codice:

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)

API REST 

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"},}'

dove USER_ID è un ID definito dall'utente con un limite di 128 caratteri.

Elenco sessioni

Per elencare le sessioni di un utente:

SDK Vertex AI per Python 

await adk_app.async_list_sessions(user_id="USER_ID")

Libreria delle richieste Python

Esegui questo codice:

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)

API REST 

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"},}'

dove USER_ID è un ID definito dall'utente con un limite di 128 caratteri.

Recuperare una sessione

Per ottenere una sessione specifica, devi disporre sia dell'ID utente sia dell'ID sessione:

SDK Vertex AI per Python 

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

Libreria delle richieste Python

Esegui questo codice:

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)

API REST 

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"},}'

Eliminare una sessione

Per eliminare una sessione, devi disporre sia dell'ID utente sia dell'ID sessione:

SDK Vertex AI per Python 

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

Libreria delle richieste Python

Esegui questo codice:

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)

API REST 

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"},}'

Trasmettere in streaming una risposta a una query

Per trasmettere in streaming le risposte di un agente in una sessione:

SDK Vertex AI per Python 

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)

Libreria delle richieste 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,
)

API REST 

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?",
  }
}'

Passaggi successivi