Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

Benutzerdefinierten Agent verwenden

Hinweis

In dieser Anleitung wird davon ausgegangen, dass Sie die Anleitung unter folgenden Links gelesen und befolgt haben:

Benutzerdefinierten Agenten erstellen: So erstellen Sie einen benutzerdefinierten agent.
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

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

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

agent = client.agent_engines.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 Agents erstellen und bereitstellen.
LOCATION ist eine der unterstützten Regionen.
RESOURCE_ID ist die ID des bereitgestellten Agents 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 AgentEngine-Klasse, die Folgendes enthält:

agent.api_resource mit Informationen zum bereitgestellten Agent. 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, die synchrone Dienstinteraktionen ermöglicht
agent.async_api_client, die 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 Agenten lokal entwickeln, haben Sie Zugriff auf die unterstützten Vorgänge und wissen, welche Vorgänge unterstützt werden. Wenn Sie einen bereitgestellten Agent verwenden möchten, können Sie die von ihm 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, in dem die Informationen einer Methode für den Agenten dokumentiert sind, die Sie aufrufen können. Die unterstützten Vorgänge hängen vom Framework ab, das Sie zum Entwickeln Ihres Agenten verwendet haben:

Im Folgenden finden Sie beispielsweise das Schema für den query-Vorgang eines 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, die auf dem Docstring der Methode basiert.
parameters ist das Schema der Eingabeargumente im OpenAPI-Schemaformat.

KI‑Agenten mit unterstützten Vorgängen abfragen

Für benutzerdefinierte Agenten können Sie alle 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

Fragen Sie den Agenten mit der Operation query ab:

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

So streamen Sie eine Antwort des Agenten mit dem Vorgang stream_query:

Agent Platform SDK

agent = agent_engines.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-Plattform streamt Antworten als Sequenz iterativ generierter Objekte. Ein Satz mit 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 Agent einen async_query-Vorgang definiert haben, wird die clientseitige asynchrone Abfrage des Agent im Agent Platform SDK für Python unterstützt:

Agent Platform SDK für Python

agent = agent_engines.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 asynchron vom Agenten streamen

Wenn Sie beim Erstellen des Agents einen async_stream_query-Vorgang definiert haben, können Sie eine Antwort vom Agent asynchron streamen, indem Sie einen seiner Vorgänge verwenden (z.B. async_stream_query):

Agent Platform SDK für Python

agent = agent_engines.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 Folge von iterativ generierten Objekten. Ein Satz mit 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 beim lokalen Testen generiert wurden.

Jobs für Abfragen mit langer Ausführungszeit

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

Lange laufenden Abfragejob starten

So starten Sie einen Abfragejob mit langer Ausführungszeit:

Agent Platform SDK für Python

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": {"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 lang andauernden Abfragejobs ab:

Agent Platform SDK

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

Benutzerdefinierten Agent 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

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 asynchron vom Agenten streamen

Agent Platform SDK für Python

Jobs für Abfragen mit langer Ausführungszeit

Lange laufenden Abfragejob starten

Agent Platform SDK für Python

REST

Status eines Abfragejobs mit langer Ausführungszeit prüfen

Agent Platform SDK

Nächste Schritte

Benutzerdefinierten Agent verwenden