ADK-Agenten verwenden

Hinweis

In dieser Anleitung wird davon ausgegangen, dass Sie die Anleitungen in den folgenden Dokumenten gelesen und befolgt haben:

• Agenten mit dem Agent Development Kit (ADK) erstellen: agent als Instanz von AdkApp erstellen.
• Nutzerauthentifizierung, um sich als Nutzer für die Abfrage des KI-Agenten zu authentifizieren.
• Importieren und initialisieren Sie das SDK, um den Client für das Abrufen einer bereitgestellten Instanz zu initialisieren (falls erforderlich).

Instanz eines KI-Agenten 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:

• PROJECT_ID ist die Google Cloud Projekt-ID, unter der Sie Agenten erstellen und bereitstellen.
• LOCATION ist eine der unterstützten Regionen und
• RESOURCE_ID ist die ID des bereitgestellten KI-Agenten als reasoningEngine-Ressource.

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:

• adk_app.api_resource mit Informationen zum bereitgestellten Agenten. Sie können auch adk_app.operation_schemas() aufrufen, um die Liste der Vorgänge zurückzugeben, die von adk_app unterstützt werden. Weitere Informationen finden Sie unter Unterstützte Vorgänge.
• adk_app.api_client, der synchrone Dienstinteraktionen ermöglicht
• adk_app.async_api_client, der asynchrone Dienstinteraktionen ermöglicht

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:

• async_stream_query: Zum Streamen einer Antwort auf eine Abfrage.

• async_create_session: Zum Erstellen einer neuen Sitzung.

• async_list_sessions: Zum Auflisten der verfügbaren Sitzungen.

• async_get_session: Zum Abrufen einer bestimmten Sitzung.

• async_delete_session: Zum Löschen einer bestimmten Sitzung.

• async_add_session_to_memory: Zum Generieren von Erinnerungen an eine Sitzung.

• async_search_memory: zum Abrufen von gemerkten Informationen.

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 Agenten auf der Agent Platform bereitgestellt haben. In diesem Abschnitt wird beschrieben, wie Sie verwaltete Sitzungen verwenden.

Hinweis: In dieser Anleitung wird davon ausgegangen, dass Sie Ihre Datenbank bei der Entwicklung Ihres KI-Agenten nicht mit session_service_builder angepasst haben.

Sitzung erstellen

Verwenden Sie die Methode AdkApp.async_create_session, um eine Sitzung für einen Nutzer zu erstellen:

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 einer Zeichenbeschränkung von 128 Zeichen aus. Beispiel: user-123.

Die Sitzung wird als Dictionary-Darstellung eines ADK-Sitzungsobjekts erstellt.

Sitzungen auflisten

Verwenden Sie die Methode AdkApp.async_list_sessions, um die Sitzungen für einen Nutzer aufzulisten:

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, wird die Wörterbuchform eines ADK-Sitzungsobjekts verwendet.

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

Verwenden Sie zum Löschen einer Sitzung 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 Abfrage streamen

Hinweis: Die maximale Dauer für eine streamQuery-Anfrage beträgt 10 Minuten. Verwenden Sie für Workflows, die längere Ausführungszeiten erfordern, stattdessen Abfragejobs mit langer Ausführungszeit.

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

Hinweis: Das Argument session_id= ist optional. Wenn sie nicht angegeben ist, wird automatisch eine neue Sitzung erstellt und für die Bereitstellung dieser Abfrage verwendet. Wenn die angegebene Sitzungs-ID nicht vorhanden ist, wird eine leere Antwort zurückgegeben. Wir empfehlen, dies wegzulassen, wenn Sie nur prüfen, ob ein Agent erreichbar ist.

Wenn Sie das Agent Platform SDK verwenden, sollten Sie eine Fortsetzung der Unterhaltung wie die folgende Sequenz von Wörterbüchern 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',
 # ...
}

Abfragejobs mit langer Ausführungszeit

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

Hinweis: Sie können Jobs mit langer Ausführungszeit nur auf reasoningEngine-Ressourcen ausführen, die am oder nach dem 22. April 2026 erstellt wurden. Sie können zuvor erstellte reasoningEngine-Ressourcen nicht aktualisieren, um Jobs mit langer Ausführungszeit auszuführen.

Agent für asynchrone Abfragen bereitstellen

Folgen Sie der allgemeinen Anleitung unter Agent bereitstellen, um einen Agent bereitzustellen. Legen Sie für ein quellbasiertes Deployment 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 KI-Agenten mit dem SDK die folgenden Umgebungsvariablen hinzufügen:

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

Abfragejob mit langer Ausführungszeit starten

Als Voraussetzung müssen Sie dem Dienst-Agenten [e-mail] die Rolle roles/storage.objectCreator für den Storage-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, wird die Antwort in dieser Datei gespeichert. Wenn es sich um ein Verzeichnis handelt, generiert das System automatisch eine Datei für die Antwort. In beiden Fällen wird die Eingabeabfrage im selben Verzeichnis mit demselben Dateinamensprä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"
}'

Beim REST API-Aufruf muss das Feld input_gcs_uri auf eine Datei verweisen, die die Abfrage enthält. Der Inhalt der Datei muss ein JSON-Objekt mit einem input-Feld sein, das mit dem input-Feld von QueryReasoningEngineRequest (z. B. { "input": { "user_id": "hello", "message":"$QUERY"} }) übereinstimmt. Wenn sich diese Eingabedatei in einem anderen Bucket als dem Ausgabespeicherort befindet, müssen Sie dem Dienst-Agent [e-mail] auch die Rolle roles/storage.objectReader für den Storage-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 Abfragejobs mit langer Ausführungszeit ab:

Agent Platform SDK

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

Abfragejob mit langer Ausführungszeit 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 Stornierung erfolgt asynchron. Die Abbruchanfrage wird zurückgegeben, sobald sie akzeptiert wurde. Der Job kann jedoch weiterhin den Status RUNNING von check_query_job melden, bis bereits laufende Arbeiten, z. B. ein blockierender Toolaufruf, abgeschlossen sind.

Sobald der Job abgebrochen wurde, wird der Vorgang mit dem Fehlercode 1 (CANCELLED) und der Meldung Cancelled by user. abgeschlossen. Beachten Sie, dass check_query_job jeden Vorgang, der mit einem Fehler abgeschlossen wurde, mit dem Status FAILED meldet. Ein abgebrochener Job wird also als FAILED und nicht mit einem separaten Abbruchstatus gemeldet. Sehen Sie sich den Fehlercode an, um einen Abbruch von einem echten Fehler zu unterscheiden.

Gemerkte Informationen verwalten

AdkApp verwendet die Memory Bank, wenn Sie eine PreloadMemoryTool in die Agentendefinition aufnehmen und den Agenten auf der Agent Platform bereitstellen. In diesem Abschnitt wird beschrieben, wie Sie mit der Standardimplementierung des ADK-Memory-Dienstes gemerkte Informationen vom Agenten generieren und abrufen.

Sitzung zu gemerkten Informationen hinzufügen

Wenn Sie wichtige Informationen aus einer Sitzung speichern möchten, um sie in zukünftigen Sitzungen zu verwenden, nutzen 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 gemerkten Informationen suchen

Wenn Sie die gemerkten Informationen des KI-Agenten durchsuchen möchten, können Sie die Methode async_search_memory verwenden:

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 gemerkte Informationen.
• QUERY ist die Abfrage, für die die Ähnlichkeitssuche ausgeführt werden soll.

Nächste Schritte

• KI-Agenten verwenden
• KI-Agenten bewerten
• Bereitgestellte Agenten verwalten
• Support anfordern