Agent Development Kit-Agenten verwenden

Hinweis

In dieser Anleitung wird davon ausgegangen, dass Sie die Anleitung unter folgenden Links gelesen und befolgt haben:

Instanz eines Agents abrufen

Wenn Sie eine AdkApp abfragen möchten, müssen Sie zuerst eine neue Instanz erstellen oder eine vorhandene Instanz abrufen.

So rufen Sie die AdkApp ab, die einer bestimmten Ressourcen-ID entspricht:

Agent Platform SDK

Führen Sie den folgenden Code aus:

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)

Dabei gilt:

Python-Bibliothek „requests“

Führen Sie den folgenden Code aus:

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

Wenn Sie das Agent Platform SDK verwenden, entspricht das adk_app-Objekt einer AgentRuntime-Klasse, die Folgendes enthält:

Im weiteren Verlauf dieses Abschnitts wird davon ausgegangen, dass Sie eine AgentRuntime-Instanz mit dem Namen adk_app haben.

Unterstützte Vorgänge

Die folgenden Vorgänge werden für AdkApp unterstützt:

So listen Sie alle unterstützten Vorgänge auf:

Agent Platform SDK

Führen Sie den folgenden Code aus:

adk_app.operation_schemas()

Python-Bibliothek „requests“

Führen Sie den folgenden Code aus:

import json

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

REST API

Wird in spec.class_methods aus der Antwort auf die curl-Anfrage dargestellt.

Sitzungen verwalten

AdkApp verwendet cloudbasierte verwaltete Sitzungen, nachdem Sie den Agent auf der Agent Platform bereitgestellt haben. In diesem Abschnitt wird beschrieben, wie Sie verwaltete Sitzungen verwenden.

Sitzung erstellen

Verwenden Sie zum Erstellen einer Sitzung für einen Nutzer die Methode AdkApp.async_create_session:

Agent Platform SDK 

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

print(session)

Python-Bibliothek „requests“

Führen Sie den folgenden Code aus:

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: Wählen Sie eine eigene Nutzer-ID mit einem Zeichenlimit von 128 Zeichen aus. Beispiel: user-123.

Die Sitzung wird als Wörterbuchdarstellung eines ADK-Sitzungsobjekts erstellt.

Sitzungen auflisten

Verwenden Sie zum Auflisten der Sitzungen für einen Nutzer die Methode 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-Bibliothek „requests“

Führen Sie den folgenden Code aus:

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

Dabei ist USER_ID die von Ihnen definierte Nutzer-ID. Beispiel: user-123.

Wenn Sitzungen zurückgegeben werden, verwenden sie die Wörterbuchform eines ADK-Sitzungsobjekts.

Sitzung abrufen

Verwenden Sie die Methode AdkApp.async_get_session, um eine bestimmte Sitzung abzurufen:

Agent Platform SDK 

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

print(session)

Python-Bibliothek „requests“

Führen Sie den folgenden Code aus:

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 ist die Dictionary-Darstellung eines ADK-Sitzungsobjekts.

Sitzung löschen

Wenn Sie eine Sitzung löschen möchten, verwenden Sie die Methode AdkApp.async_delete_session:

Agent Platform SDK 

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

Python-Bibliothek „requests“

Führen Sie den folgenden Code aus:

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

Antwort auf eine Anfrage streamen

Wenn Sie Antworten von einem Agenten in einer Sitzung streamen möchten, verwenden Sie die Methode 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-Bibliothek „requests“ 

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

Wenn Sie das Agent Platform SDK verwenden, sollten Sie eine Fortsetzung der Unterhaltung wie die folgende Sequenz von Dictionaries erhalten:

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

Jobs für Abfragen mit langer Ausführungszeit

Abfragen, deren Ausführung lange dauern kann (bis zu sieben Tage), können Sie als lang andauernde Jobs ausführen. Diese Jobs werden asynchron ausgeführt. Sie können den Jobstatus prüfen und die Ergebnisse später abrufen.

Agent für asynchrone Abfrage bereitstellen

Folgen Sie der allgemeinen Anleitung unter KI-Agent bereitstellen, um einen KI-Agent bereitzustellen. Legen Sie für die quellenbasierte Bereitstellung das Feld deploymentSpec.agentFramework auf google-adk fest.

Wenn Sie einen benutzerdefinierten API-Endpunkt verwenden, indem Sie ein eigenes Container-Image erstellen, müssen Sie beim Erstellen des Agents mit dem SDK die folgenden Umgebungsvariablen hinzufügen:

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

Lange laufenden Abfragejob starten

Als Voraussetzung müssen Sie dem Dienst-Agenten service-PROJECT_NUMBER@gcp-sa-aiplatform-re.s3ns-system.iam.gserviceaccount.com die Rolle roles/storage.objectCreator für den Speicher-Bucket für Ausgabedateien zuweisen.

So starten Sie einen Abfragejob mit langer Ausführungszeit:

Agent Platform SDK 

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)

Mit dem SDK kann output_gcs_uri ein Verzeichnis oder ein Dateiname sein. Wenn es sich um einen Dateinamen handelt, verwendet das System diese Datei zum Speichern der Antwort. Wenn es sich um ein Verzeichnis handelt, generiert das System automatisch eine Datei für die Antwort. In beiden Fällen wird die Eingabeanfrage im selben Verzeichnis mit demselben Dateinamenpräfix wie die Ausgabedatei gespeichert.

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

Für den REST API-Aufruf muss das Feld input_gcs_uri auf eine Datei verweisen, die die Anfrage enthält. Der Inhalt der Datei muss ein JSON-Objekt mit einem Feld input sein, das dem Feld input von QueryReasoningEngineRequest entspricht (z. B. { "input": { "user_id": "hello", "message":"$QUERY"} }). Wenn sich diese Eingabedatei in einem anderen Bucket als der Ausgabespeicherort befindet, müssen Sie dem Dienst-Agenten service-PROJECT_NUMBER@gcp-sa-aiplatform-re.s3ns-system.iam.gserviceaccount.com auch die Rolle roles/storage.objectReader für den Speicher-Bucket zuweisen, in dem sich die Eingabedateien befinden.

output_gcs_uri muss ein Dateiname sein.

Status eines Abfragejobs mit langer Ausführungszeit prüfen

So prüfen Sie den Status und rufen die Ergebnisse eines lang andauernden Abfragejobs ab:

Agent Platform SDK 

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

Lange laufenden Abfragejob abbrechen

Wenn Sie einen Abfragejob mit langer Ausführungszeit abbrechen möchten, benötigen Sie den LRO-Ressourcennamen, der vom Abfragejob mit langer Ausführungszeit zurückgegeben wird.

Agent Platform SDK 

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

Die Kündigung erfolgt asynchron. Die Kündigungsanfrage wird zurückgegeben, sobald sie akzeptiert wurde. Der Job kann jedoch weiterhin den Status RUNNING von check_query_job melden, bis bereits laufende Vorgänge wie ein blockierender Toolaufruf abgeschlossen sind.

Sobald der Job abgebrochen wurde, wird der zugehörige Vorgang mit dem Fehlercode 1 (CANCELLED) und der Meldung Cancelled by user. abgeschlossen. check_query_job gibt jeden Vorgang, der mit einem Fehler abgeschlossen wurde, mit dem Status FAILED zurück. Ein abgebrochener Job wird also als FAILED gemeldet und nicht mit einem separaten Status für abgebrochene Jobs. Anhand des Fehlercodes können Sie eine Kündigung von einem tatsächlichen Fehler unterscheiden.

Gemerkte Informationen verwalten

AdkApp verwendet Memory Bank, wenn Sie eine PreloadMemoryTool in der Agentendefinition einfügen und den Agenten auf der Agent Platform bereitstellen. In diesem Abschnitt wird beschrieben, wie Sie mit der Standardimplementierung des ADK-Speicherdienstes Erinnerungen für den Agenten generieren und abrufen.

Sitzung zu gemerkten Informationen hinzufügen

Wenn Sie sich in einer Sitzung an wichtige Informationen erinnern möchten, die in zukünftigen Sitzungen verwendet werden können, verwenden Sie die Methode async_add_session_to_memory:

Agent Platform SDK 

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

Dabei ist SESSION_DICT die Dictionary-Form eines ADK-Sitzungsobjekts.

Nach Erinnerungen suchen

Mit der Methode async_search_memory können Sie die Erinnerungen des Agents durchsuchen:

Agent Platform SDK 

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

Dabei gilt:

  • USER_ID ist der Bereich für relevante Erinnerungen.
  • QUERY ist die Abfrage, für die die Suche nach Ähnlichkeiten ausgeführt werden soll.

Nächste Schritte