Agent Development Kit-Agent verwenden

Zusätzlich zu den allgemeinen Anleitungen zur Verwendung eines Agents werden auf dieser Seite Funktionen beschrieben, die speziell für AdkApp gelten.

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 von AdkApp 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).

Wenn Sie eine ADK-Anwendung abfragen möchten, müssen Sie zuerst eine neue ADK-Anwendungsinstanz erstellen oder eine vorhandene Instanz abrufen.

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

Vertex AI SDK für Python

Führen Sie den folgenden Code aus:

adk_app = client.agent_engines.get(name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_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

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

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.

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.

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

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")

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

Dabei ist USER_ID eine benutzerdefinierte ID mit einem Zeichenlimit von 128.

Sitzungen auflisten

So listen Sie die Sitzungen für einen Nutzer auf:

Vertex AI SDK für Python

await adk_app.async_list_sessions(user_id="USER_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_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 eine benutzerdefinierte ID mit einem Zeichenlimit von 128.

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")

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

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

Hinweis: Das Argument session_id= ist optional. Wenn sie nicht angegeben ist, wird automatisch eine neue Sitzung erstellt und für die Bereitstellung der Anfrage verwendet.

Nächste Schritte

• Agent verwenden.
• Agent bewerten
• Bereitgestellte Agents verwalten.
• Support anfordern