Hinweise
In dieser Anleitung wird davon ausgegangen, dass Sie die Anleitung unter folgenden Links gelesen und befolgt haben:
- Agent Development Kit-Agent entwickeln:
agent
als Instanz vonAdkApp
entwickeln. - Nutzerauthentifizierung, um sich als Nutzer zu authentifizieren und den Agent abzufragen.
- Importieren und initialisieren Sie das SDK, um den Client zum Abrufen einer bereitgestellten Instanz zu initialisieren (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
für eine bestimmte Ressourcen-ID ab:
Vertex AI SDK für Python
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 Agents entwickeln und bereitstellen.LOCATION
ist eine der unterstützten Regionen.RESOURCE_ID
ist die ID des bereitgestellten Agents alsreasoningEngine
-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 Vertex AI SDK für Python verwenden, entspricht das adk_app
-Objekt einer AgentEngine
-Klasse, die Folgendes enthält:
- eine
adk_app.api_resource
mit Informationen zum bereitgestellten Agent. Sie können auchadk_app.operation_schemas()
aufrufen, um die Liste der Vorgänge zurückzugeben, die vonadk_app
unterstützt werden. Weitere Informationen finden Sie unter Unterstützte Vorgänge. - eine
adk_app.api_client
, die synchrone Dienstinteraktionen ermöglicht - ein
adk_app.async_api_client
, das asynchrone Dienstinteraktionen ermöglicht
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 Anfrage.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:
Vertex AI SDK für Python
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 in Vertex AI Agent Engine bereitgestellt haben. In diesem Abschnitt wird beschrieben, wie Sie verwaltete Sitzungen verwenden.
Sitzung erstellen
So erstellen Sie eine Sitzung für einen Nutzer:
Vertex AI SDK für Python
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://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}: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
So listen Sie die Sitzungen für einen Nutzer auf:
Vertex AI SDK für Python
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://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}: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
Um eine bestimmte Sitzung abzurufen, benötigen Sie sowohl die Nutzer-ID als auch die Sitzungs-ID:
Vertex AI SDK für Python
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://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}: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 Wörterbuchdarstellung eines ADK-Sitzungsobjekts.
Sitzung löschen
Wenn Sie eine Sitzung löschen möchten, benötigen Sie sowohl die Nutzer-ID als auch die Sitzungs-ID:
Vertex AI SDK für Python
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://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}: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
So streamen Sie Antworten von einem Agenten in einer Sitzung:
Vertex AI SDK für Python
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://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}: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 Vertex AI SDK für Python 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',
# ...
}
Erinnerungen verwalten
AdkApp
verwendet die Speicherbank der Vertex AI Agent Engine, wenn Sie eine PreloadMemoryTool
in der Agentdefinition angeben und den Agent in Vertex AI Agent Engine bereitstellen. In diesem Abschnitt wird beschrieben, wie Sie mit der Standardimplementierung des ADK-Speicherdienstes Erinnerungen für den Agenten generieren und abrufen.
Sitzung zu „Gemerkte 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
:
Vertex AI SDK für Python
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 gemerkten Informationen des KI-Agenten durchsuchen:
Vertex AI SDK für Python
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.