Utilizzare un agente Agent Development Kit

Prima di iniziare

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

• Crea un agente Agent Development Kit: per creare agent come istanza di AdkApp.
• Autenticazione utente per l'autenticazione come utente per interrogare l'agente.
• Importa e inizializza l'SDK per inizializzare il client per ottenere un'istanza di cui è stato eseguito il deployment (se necessario).

Recuperare un'istanza di un agente

Per eseguire query su un AdkApp, devi prima creare una nuova istanza o recuperare un'istanza esistente.

Per ottenere l'AdkApp che corrisponde a un ID risorsa specifico:

SDK Agent Platform

Esegui questo codice:

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)

dove

• PROJECT_ID è l'ID progetto Google Cloud in cui crei e implementi gli agenti,
• LOCATION è una delle regioni supportate e
• RESOURCE_ID è l'ID dell'agente di cui è stato eseguito il deployment come risorsa reasoningEngine.

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

Quando utilizzi l'SDK Agent Platform, l'oggetto adk_app corrisponde a una classe AgentEngine che contiene quanto segue:

• adk_app.api_resource con informazioni sull'agente di cui è stato eseguito il deployment. Puoi anche chiamare adk_app.operation_schemas() per restituire l'elenco delle operazioni supportate da adk_app. Per maggiori dettagli, vedi Operazioni supportate.
• adk_app.api_client che consente interazioni di servizio sincrone
• adk_app.async_api_client che consente interazioni asincrone tra servizi

Il resto di questa sezione presuppone che tu disponga di un'istanza AgentEngine denominata adk_app.

Operazioni supportate

Per AdkApp sono supportate le seguenti operazioni:

• async_stream_query: per lo streaming di una risposta a una query.

• async_create_session: per creare una nuova sessione.

• async_list_sessions: per elencare le sessioni disponibili.

• async_get_session: per recuperare una sessione specifica.

• async_delete_session: per eliminare una sessione specifica.

• async_add_session_to_memory: per generare i ricordi di una sessione.

• async_search_memory: per recuperare i ricordi.

Per elencare tutte le operazioni supportate:

SDK Agent Platform

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 su cloud dopo il deployment dell'agente su Agent Platform. Questa sezione descrive come utilizzare le sessioni gestite.

Nota: queste istruzioni presuppongono che tu non abbia personalizzato il database utilizzando session_service_builder durante lo sviluppo dell'agente.

Creare una sessione

Per creare una sessione per un utente, utilizza il metodo AdkApp.async_create_session:

SDK Agent Platform

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

print(session)

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://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)

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

• USER_ID: scegli il tuo ID utente con un limite di 128 caratteri. Ad esempio, user-123.

La sessione viene creata come rappresentazione del dizionario di un oggetto sessione ADK.

Elenco sessioni

Per elencare le sessioni di un utente, utilizza il metodo AdkApp.async_list_sessions:

SDK Agent Platform

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

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://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)

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 è l'ID utente che hai definito. Ad esempio, user-123.

Se vengono restituite sessioni, queste utilizzano la forma di dizionario di un oggetto sessione ADK.

Recuperare una sessione

Per ottenere una sessione specifica, utilizza il metodo AdkApp.async_get_session:

SDK Agent Platform

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

print(session)

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://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)

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

session è la rappresentazione del dizionario di un oggetto sessione ADK.

Eliminare una sessione

Per eliminare una sessione, utilizza il metodo AdkApp.async_delete_session:

SDK Agent Platform

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://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)

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, utilizza il metodo AdkApp.async_stream_query:

SDK Agent Platform

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://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,
)

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

Nota: l'argomento session_id= è facoltativo. Se non viene specificata, viene creata e utilizzata automaticamente una nuova sessione per la pubblicazione della query. Se l'ID sessione fornito non esiste, viene restituita una risposta vuota. Ti consigliamo di ometterlo se stai solo verificando se un agente è raggiungibile.

Se utilizzi l'SDK Agent Platform, dovresti ricevere una continuazione della conversazione come la seguente sequenza di dizionari:

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

Job di query a lunga esecuzione

Per le query che possono richiedere molto tempo per essere completate (fino a sette giorni), puoi eseguirle come job a esecuzione prolungata. Questi job vengono eseguiti in modo asincrono. Puoi controllare lo stato del job e recuperare i risultati in un secondo momento.

Nota: puoi eseguire job a lunga esecuzione solo su risorse reasoningEngine create a partire dal 22 aprile 2026. Non puoi aggiornare le risorse reasoningEngine create in precedenza per eseguire job a lunga esecuzione.

Esegui il deployment di un agente per la query asincrona

Per eseguire il deployment di un agente, segui le istruzioni generali riportate in Eseguire il deployment di un agente. Per il deployment basato sull'origine, imposta il campo deploymentSpec.agentFramework su google-adk.

Se utilizzi un endpoint API personalizzato creando la tua immagine container, devi aggiungere le seguenti variabili di ambiente quando crei l'agente utilizzando l'SDK:

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

Avvia un job di query a lunga esecuzione

Come prerequisito, devi concedere al service agent [e-mail] il ruolo roles/storage.objectCreator per il bucket di archiviazione dei file di output.

Per avviare un job di query a lunga esecuzione:

SDK Agent Platform

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)

Con l'SDK, output_gcs_uri può essere una directory o un nome file. Se si tratta di un nome file, il sistema utilizza questo file per archiviare la risposta. Se si tratta di una directory, il sistema genera automaticamente un file per la risposta. In entrambi i casi, la query di input viene archiviata nella stessa directory con lo stesso prefisso del nome file del file di output.

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

Per la chiamata API REST, il campo input_gcs_uri deve puntare a un file che contiene la query. Il contenuto del file deve essere un oggetto JSON con un campo input che corrisponda al campo input di QueryReasoningEngineRequest (ad esempio { "input": { "user_id": "hello", "message":"$QUERY"} }). Se questo file di input si trova in un bucket diverso dalla posizione di output, devi anche concedere all'agente di servizio [e-mail] il ruolo roles/storage.objectReader al bucket di archiviazione in cui si trovano i file di input.

output_gcs_uri deve essere un nome file.

Controllare lo stato di un job di query a esecuzione prolungata

Per controllare lo stato e recuperare i risultati di un job di query a esecuzione prolungata:

SDK Agent Platform

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

Annullare un job di query a lunga esecuzione

Per annullare un job di query a lunga esecuzione, devi disporre del nome della risorsa LRO restituito dal job di query a lunga esecuzione.

SDK Agent Platform

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

Gestire i ricordi

AdkApp utilizza Memory Bank se includi un PreloadMemoryTool nella definizione dell'agente e lo implementi in Agent Platform. Questa sezione descrive come generare e recuperare ricordi dall'agente tramite l'implementazione predefinita del servizio di memoria ADK.

Aggiungere una sessione alla memoria

Per conservare la memoria di informazioni significative in una sessione (che possono essere utilizzate in sessioni future), utilizza il metodo async_add_session_to_memory:

SDK Agent Platform

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

dove SESSION_DICT è la forma di dizionario di un oggetto sessione ADK.

Cercare ricordi

Per cercare nelle memorie dell'agente, puoi utilizzare il metodo async_search_memory:

SDK Agent Platform

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

dove

• USER_ID è l'ambito dei ricordi pertinenti.
• QUERY è la query per cui eseguire la ricerca di similarità.

Passaggi successivi

• Utilizzare un agente.
• Valuta un agente.
• Gestisci gli agenti di cui è stato eseguito il deployment.
• Richiedere assistenza.