Diese Seite wurde von der Cloud Translation API übersetzt.

Benutzerdefinierten Agent verwenden

Hinweise

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

Benutzerdefinierten Agent entwickeln: um einen benutzerdefinierten agent zu 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

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:

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

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

eine 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 vom Agent unterstützt werden. Weitere Informationen finden Sie unter Unterstützte Vorgänge.
eine agent.api_client, die synchrone Dienstinteraktionen ermöglicht
ein agent.async_api_client, das 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 das sind. Wenn Sie einen bereitgestellten Agent verwenden möchten, können Sie die von ihm unterstützten Vorgänge auflisten:

Vertex AI SDK für Python

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 basierend auf dem Docstring der Methode.
parameters ist das Schema der Eingabeargumente im OpenAPI-Schemaformat.

Agent mit unterstützten Vorgängen abfragen

Für benutzerdefinierte Agents können Sie alle folgenden Abfrage- oder Streamingvorgänge verwenden, die Sie bei der Entwicklung Ihres Agents 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-Entwicklungs-Kit	`async_stream_query`
LangChain	`query`, `stream_query`
LangGraph	`query`, `stream_query`
AG2	`query`
LlamaIndex	`query`

Agent abfragen

Fragen Sie den Agent mit dem Vorgang query ab:

Vertex AI SDK für Python

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

Streamantworten vom Kundenservicemitarbeiter

So streamen Sie eine Antwort vom Agent mithilfe des Vorgangs stream_query:

Vertex AI SDK für Python

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

Vertex AI Agent Engine streamt Antworten als Sequenz iterativ generierter Objekte. Ein Satz 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

Agent asynchron abfragen

Wenn Sie beim Entwickeln des Agenten einen async_query-Vorgang definiert haben, wird die clientseitige asynchrone Abfrage des Agenten im Vertex AI SDK für Python unterstützt:

Vertex AI 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 Dictionary, 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 Entwickeln des Agents einen async_stream_query-Vorgang definiert haben, können Sie eine Antwort vom Agent asynchron streamen. Verwenden Sie dazu einen der Vorgänge des Agents, z. B. async_stream_query:

Vertex AI 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 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

Die Antworten sollten mit denen übereinstimmen, die beim lokalen Testen generiert wurden.

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

Hinweise

Instanz eines Agenten abrufen

Vertex AI SDK für Python

Anfragen

REST

Unterstützte Vorgänge auflisten

Vertex AI SDK für Python

Anfragen

REST

Agent mit unterstützten Vorgängen abfragen

Agent abfragen

Vertex AI SDK für Python

Anfragen

REST

Streamantworten vom Kundenservicemitarbeiter

Vertex AI SDK für Python

Anfragen

REST

Agent asynchron abfragen

Vertex AI SDK für Python

Antworten asynchron vom Agenten streamen

Vertex AI SDK für Python

Nächste Schritte

Benutzerdefinierten Agent verwenden