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

Benutzerdefinierten KI-Agenten verwenden

Hinweis

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

Benutzerdefinierten Agent erstellen: Zum Erstellen eines benutzerdefinierten agent.
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

Zum Abfragen eines KI-Agenten benötigen Sie zuerst eine Instanz eines KI-Agenten. Sie können entweder eine neue Instanz erstellen oder eine vorhandene Instanz eines Agenten abrufen.

So rufen Sie den Agenten ab, der 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",
)

agent = client.runtimes.get(name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

print(agent)

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 KI-Agenten als reasoningEngine-Ressource.

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

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 für Python verwenden, entspricht das agent-Objekt einer AgentRuntime-Klasse, die Folgendes enthält:

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

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

Unterstützte Vorgänge auflisten

Wenn Sie den KI-Agenten lokal entwickeln, haben Sie Zugriff auf die unterstützten Vorgänge und kennen diese. Wenn Sie einen bereitgestellten Agent verwenden möchten, können Sie die unterstützten Vorgänge auflisten:

Agent Platform SDK

Führen Sie den folgenden Code aus:

print(agent.operation_schemas())

Anfragen

Führen Sie den folgenden Code aus:

import json

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

REST

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

Das Schema für jeden Vorgang ist ein Wörterbuch, das die Informationen einer Methode für den Agenten dokumentiert, die Sie aufrufen können. Die unterstützten Vorgänge hängen vom Framework ab, das Sie zur Entwicklung Ihres KI-Agenten verwendet haben:

Das folgende Beispiel zeigt das Schema für den Vorgang query einer LangchainAgent:

{'api_mode': '',
 'name': 'query',
 'description': """Queries the Agent with the given input and config.
    Args:
        input (Union[str, Mapping[str, Any]]):
            Required. The input to be passed to the Agent.
        config (langchain_core.runnables.RunnableConfig):
            Optional. The config (if any) to be used for invoking the Agent.
    Returns:
        The output of querying the Agent with the given input and config.
""",            '        ',
 'parameters': {'$defs': {'RunnableConfig': {'description': 'Configuration for a Runnable.',
                                             'properties': {'configurable': {...},
                                                            'run_id': {...},
                                                            'run_name': {...},
                                                            ...},
                                             'type': 'object'}},
                'properties': {'config': {'nullable': True},
                               'input': {'anyOf': [{'type': 'string'}, {'type': 'object'}]}},
                'required': ['input'],
                'type': 'object'}}

Dabei gilt:

name ist der Name des Vorgangs (z.B. agent.query für einen Vorgang mit dem Namen query).
api_mode ist der API-Modus des Vorgangs ("" für synchron, "stream" für Streaming).
description ist eine Beschreibung des Vorgangs basierend auf dem Docstring der Methode.
parameters ist das Schema der Eingabeargumente im OpenAPI-Schemaformat.

KI-Agenten mit unterstützten Vorgängen abfragen

Für benutzerdefinierte Agenten können Sie die folgenden Abfrage- oder Streamingvorgänge verwenden, die Sie bei der Entwicklung Ihres Agenten definiert haben:

query
stream_query
async_query
async_stream_query

Beachten Sie, dass bestimmte Frameworks nur bestimmte Abfrage- oder Streamingvorgänge unterstützen:

Framework	Unterstützte Abfragevorgänge
Agent Development Kit (ADK)	`async_stream_query`
LangChain	`query`, `stream_query`
LangGraph	`query`, `stream_query`
AG2	`query`
LlamaIndex	`query`

KI-Agenten abfragen

Senden Sie mit dem Vorgang query eine Anfrage an den KI-Agenten:

Agent Platform SDK

agent.query(input="What is the exchange rate from US dollars to Swedish Krona today?")

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:query",
    headers={
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": f"Bearer {get_identity_token()}",
    },
    data=json.dumps({
        "class_method": "query",
        "input": {
            "input": "What is the exchange rate from US dollars to Swedish Krona today?"
        }
    })
)

REST

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": "query",
  "input": {
    "input": "What is the exchange rate from US dollars to Swedish Krona today?"
  }
}'

Die Abfrageantwort ist ein String, der der Ausgabe eines lokalen Anwendungstests ähnelt:

{"input": "What is the exchange rate from US dollars to Swedish Krona today?",
 # ...
 "output": "For 1 US dollar you will get 10.7345 Swedish Krona."}

Antworten des KI-Agenten streamen

Streamen Sie eine Antwort des KI-Agenten mit dem Vorgang stream_query:

Agent Platform SDK

agent = client.runtimes.get("projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

for response in agent.stream_query(
    input="What is the exchange rate from US dollars to Swedish Krona today?"
):
    print(response)

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": "stream_query",
        "input": {
            "input": "What is the exchange rate from US dollars to Swedish Krona today?"
        },
    }),
    stream=True,
)

REST

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": "stream_query",
  "input": {
    "input": "What is the exchange rate from US dollars to Swedish Krona today?"
  }
}'

Die Agent Platform streamt Antworten als Sequenz iterativ generierter Objekte. Eine Gruppe von drei Antworten könnte beispielsweise so aussehen:

{'actions': [{'tool': 'get_exchange_rate', ...}]}  # first response
{'steps': [{'action': {'tool': 'get_exchange_rate', ...}}]}  # second response
{'output': 'The exchange rate is 11.0117 SEK per USD as of 2024-12-03.'}  # final response

KI-Agenten asynchron abfragen

Wenn Sie beim Erstellen des KI-Agenten einen async_query-Vorgang definiert haben, wird im Agent Platform SDK für Python die clientseitige asynchrone Abfrage des KI-Agenten unterstützt:

Agent Platform SDK für Python

agent = client.runtimes.get("projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

response = await agent.async_query(
    input="What is the exchange rate from US dollars to Swedish Krona today?"
)
print(response)

Die Abfrageantwort ist ein Wörterbuch, das der Ausgabe eines lokalen Tests entspricht:

{"input": "What is the exchange rate from US dollars to Swedish Krona today?",
 # ...
 "output": "For 1 US dollar you will get 10.7345 Swedish Krona."}

Antworten des KI-Agenten asynchron streamen

Wenn Sie beim Erstellen des KI-Agenten einen async_stream_query-Vorgang definiert haben, können Sie eine Antwort des KI-Agenten asynchron mit einem seiner Vorgänge (z.B. async_stream_query) streamen:

Agent Platform SDK für Python

agent = client.runtimes.get("projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

async for response in agent.async_stream_query(
    input="What is the exchange rate from US dollars to Swedish Krona today?"
):
    print(response)

Der Vorgang async_stream_query ruft im Hintergrund denselben streamQuery-Endpunkt auf und streamt Antworten asynchron als Sequenz von iterativ generierten Objekten. Eine Gruppe von drei Antworten könnte beispielsweise so aussehen:

{'actions': [{'tool': 'get_exchange_rate', ...}]}  # first response
{'steps': [{'action': {'tool': 'get_exchange_rate', ...}}]}  # second response
{'output': 'The exchange rate is 11.0117 SEK per USD as of 2024-12-03.'}  # final response

Die Antworten sollten mit denen übereinstimmen, die während des lokalen Tests generiert wurden.

Abfragejobs mit langer Ausführungszeit

Abfragen, deren Ausführung sehr lange dauern kann (bis zu sieben Tage), können Sie als Jobs mit langer Ausführungszeit ausführen. Weitere Informationen finden Sie unter ADK-Agent verwenden.

Abfragejob mit langer Ausführungszeit starten

So starten Sie einen Abfragejob mit langer Ausführungszeit:

Agent Platform SDK für Python

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": {"product": "Google"} }',
        "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE",
    },
)
print(response)

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

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)

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

Hinweis

Instanz eines KI-Agenten abrufen

Agent Platform SDK

Anfragen

REST

Unterstützte Vorgänge auflisten

Agent Platform SDK

Anfragen

REST

KI-Agenten mit unterstützten Vorgängen abfragen

KI-Agenten abfragen

Agent Platform SDK

Anfragen

REST

Antworten des KI-Agenten streamen

Agent Platform SDK

Anfragen

REST

KI-Agenten asynchron abfragen

Agent Platform SDK für Python

Antworten des KI-Agenten asynchron streamen

Agent Platform SDK für Python

Abfragejobs mit langer Ausführungszeit

Abfragejob mit langer Ausführungszeit starten

Agent Platform SDK für Python

REST

Status eines Abfragejobs mit langer Ausführungszeit prüfen

Agent Platform SDK

Nächste Schritte

Benutzerdefinierten KI-Agenten verwenden