Utilizza un agente ADK

Prima di iniziare

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

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

Ottieni un'istanza di un agente

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

Per ottenere il AdkApp corrispondente a un ID risorsa specifico:

SDK Agent Platform

Esegui questo codice:

import agentplatform

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

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

print(adk_app)

dove

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

Libreria di 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 AgentRuntime che contiene quanto segue:

• adk_app.api_resource con le 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, consulta Operazioni supportate.
• adk_app.api_client che consente interazioni sincrone tra i servizi
• adk_app.async_api_client che consente interazioni asincrone tra i servizi

Il resto di questa sezione presuppone che tu abbia un'istanza AgentRuntime 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 di 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.

Gestisci 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.

Crea 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 di 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 caratteri di 128. Ad esempio, user-123.

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

Elenca 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 di 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.

Recupera 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 di 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.

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

Trasmetti in streaming una risposta a una query

Nota: la durata massima di una richiesta streamQuery è di 10 minuti. Per i workflow che richiedono tempi di esecuzione più lunghi, utilizza i job di query a lunga esecuzione.

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 di 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 specificato, viene creata automaticamente una nuova sessione che viene utilizzata per gestire la query. Se l'ID sessione fornito non esiste, viene restituita una risposta vuota. Ti consigliamo di omettere questo campo 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 lunga esecuzione. 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 il 22 aprile 2026 o in una data successiva. Non puoi aggiornare le risorse reasoningEngine create in precedenza per eseguire job a lunga esecuzione.

Esegui il deployment di un agente per le query asincrone

Per eseguire il deployment di un agente, segui le istruzioni generali riportate in Esegui 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 durante la creazione dell'agente utilizzando l'SDK:

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

Avvia un job di query a lunga esecuzione

Come prerequisito, devi concedere all'agente di servizio [e-mail] il ruolo roles/storage.objectCreator per il bucket di archiviazione per i file di output.

Per avviare un job di query a lunga esecuzione:

SDK Agent Platform

import agentplatform

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

response = client.runtimes.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 corrisponde 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 per il bucket di archiviazione in cui si trovano i file di input.

output_gcs_uri deve essere un nome file.

Controlla lo stato di un job di query a lunga esecuzione

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

SDK Agent Platform

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

Annulla un job di query a lunga esecuzione

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

SDK Agent Platform

response = client.runtimes.cancel_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    config={
        "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"
}'

La cancellazione è asincrona. La richiesta di annullamento viene restituita non appena viene accettata, ma il job può continuare a segnalare lo stato RUNNING da check_query_job fino al completamento del lavoro già in corso, ad esempio una chiamata allo strumento di blocco.

Una volta annullato il job, l'operazione viene completata con un codice di errore 1 (CANCELLED) e il messaggio Cancelled by user.. Tieni presente che check_query_job segnala ogni operazione completata con un errore come stato FAILED, pertanto un job annullato viene segnalato come FAILED anziché tramite uno stato annullato distinto. Controlla il codice di errore per distinguere un annullamento da un errore effettivo.

Gestisci i ricordi

AdkApp utilizza Memory Bank se includi un PreloadMemoryTool nella definizione dell'agente ed esegui il deployment dell'agente su Agent Platform. Questa sezione descrive come generare e recuperare le memorie dall'agente tramite l'implementazione predefinita del servizio di memoria ADK.

Aggiungi la 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.

Cerca nei ricordi

Per eseguire ricerche 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

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