Google verwendet KI-Technologie, um Inhalte in Ihre bevorzugte Sprache zu übersetzen. KI-Übersetzungen können Fehler enthalten.

Agent Development Kit-Agenten verwenden

Hinweis

In diesem Lernprogramm wird davon ausgegangen, dass Sie die Anleitung in den folgenden Artikeln gelesen und befolgt haben:

Agent Development Kit-Agent erstellen: zum Erstellen von agent als Instanz von AdkApp.
Nutzerauthentifizierung: zum Authentifizieren als Nutzer für die Abfrage des Agenten.
SDK importieren und initialisieren: zum Initialisieren des Clients für den Abruf einer bereitgestellten Instanz (falls erforderlich).

Instanz eines 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 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)

Dabei gilt:

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

Python-Bibliothek für Anfragen

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 AgentEngine 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 für synchrone Dienstinteraktionen
adk_app.async_api_client für asynchrone Dienstinteraktionen

Im weiteren Verlauf dieses Abschnitts wird davon ausgegangen, dass Sie eine AgentEngine-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 Erinnerungen.

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 für Anfragen

Führen Sie den folgenden Code aus:

import json

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

REST API

In spec.class_methods aus der Antwort auf die Curl-Anfrage dargestellt.

Sitzungen verwalten

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

Sitzung erstellen

Verwenden Sie die AdkApp.async_create_session Methode, 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 für Anfragen

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 Zeichenbegrenzung von 128. Beispiel: user-123.

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

Sitzungen auflisten

Verwenden Sie die AdkApp.async_list_sessions Methode, 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 für Anfragen

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 AdkApp.async_get_session Methode, 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 für Anfragen

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

Der session ist die Wörterbuchdarstellung eines ADK-Sitzungsobjekts.

Sitzung löschen

Verwenden Sie die AdkApp.async_delete_session Methode, um eine Sitzung zu löschen:

Agent Platform SDK

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

Python-Bibliothek für Anfragen

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

Verwenden Sie die AdkApp.async_stream_query Methode, um Antworten von einem Agenten in einer Sitzung zu streamen:

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 für Anfragen

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

Lang andauernde Abfragejobs

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

Agent für asynchrone Abfrage bereitstellen

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

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

Lang andauernden Abfragejob starten

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

So starten Sie einen lang andauernden Abfragejob:

Agent Platform SDK

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)

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

Für den 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 übereinstimmt (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.iam.gserviceaccount.com auch die Rolle roles/storage.objectReader für den Storage-Bucket gewähren, in dem sich die Eingabe dateien befinden.

output_gcs_uri muss ein Dateiname sein.

Status eines lang andauernden Abfragejobs prüfen

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

Agent Platform SDK

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

Lang andauernden Abfragejob abbrechen

Wenn Sie einen lang andauernden Abfragejob abbrechen möchten, benötigen Sie den LRO-Ressourcennamen, der vom lang andauernden Abfragejob zurückgegeben wird.

Agent Platform SDK

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

Erinnerungen verwalten

AdkApp verwendet Memory Bank wenn Sie ein PreloadMemoryTool in die Agentendefinition einfügen und den Agenten in der Agent Platform bereitstellen. In diesem Abschnitt wird beschrieben, wie Sie mit der Standardimplementierung des ADK-SpeicherdienstesErinnerungen vom Agenten generieren und abrufen.

Sitzung zu gemerkten Informationen hinzufügen

Verwenden Sie die Methode async_add_session_to_memory, um wichtige Informationen in einer Sitzung zu speichern, die in zukünftigen Sitzungen verwendet werden können:

Agent Platform SDK

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

Dabei ist SESSION_DICT die Wörterbuchform eines ADK-Sitzungsobjekts.

Nach Erinnerungen suchen

Verwenden Sie die async_search_memory Methode, um in den Erinnerungen des Agenten zu suchen:

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 eine Ähnlichkeitssuche durchgeführt werden soll.

Agent Development Kit-Agenten verwenden Mit Sammlungen den Überblick behalten Sie können Inhalte basierend auf Ihren Einstellungen speichern und kategorisieren.

Hinweis

Instanz eines Agenten abrufen

Agent Platform SDK

Python-Bibliothek für Anfragen

REST API

Unterstützte Vorgänge

Agent Platform SDK

Python-Bibliothek für Anfragen

REST API

Sitzungen verwalten

Sitzung erstellen

Agent Platform SDK

Python-Bibliothek für Anfragen

REST API

Sitzungen auflisten

Agent Platform SDK

Python-Bibliothek für Anfragen

REST API

Sitzung abrufen

Agent Platform SDK

Python-Bibliothek für Anfragen

REST API

Sitzung löschen

Agent Platform SDK

Python-Bibliothek für Anfragen

REST API

Antwort auf eine Abfrage streamen

Agent Platform SDK

Python-Bibliothek für Anfragen

REST API

Lang andauernde Abfragejobs

Agent für asynchrone Abfrage bereitstellen

Lang andauernden Abfragejob starten

Agent Platform SDK

REST

Status eines lang andauernden Abfragejobs prüfen

Agent Platform SDK

Lang andauernden Abfragejob abbrechen

Agent Platform SDK

REST

Erinnerungen verwalten

Sitzung zu gemerkten Informationen hinzufügen

Agent Platform SDK

Nach Erinnerungen suchen

Agent Platform SDK

Nächste Schritte

Agent Development Kit-Agenten verwenden