Kundenservicemitarbeiter verwenden

Der Code zum Abfragen eines Agents ist derselbe, unabhängig davon, ob er lokal oder remote bereitgestellt wird. Daher bezieht sich der Begriff agent auf dieser Seite abwechselnd auf local_agent oder remote_agent. Da sich die unterstützten Vorgänge je nach Framework unterscheiden, stellen wir Nutzungsanleitungen für frameworkspezifische Vorlagen bereit:

Framework Beschreibung
Agent Development Kit (Vorschau) Die Vorlagen basieren auf den internen Best Practices von Google für Entwickler, die KI-Anwendungen erstellen, oder Teams, die schnell robuste, auf Agents basierende Lösungen prototypisieren und bereitstellen müssen.
LangChain Einfacher zu verwenden für grundlegende Anwendungsfälle aufgrund der vordefinierten Konfigurationen und Abstraktionen.
LangGraph Graphbasierter Ansatz zum Definieren von Workflows mit erweiterten Human-in-the-Loop- und Rewind/Replay-Funktionen.
AG2 (früher AutoGen) AG2 bietet ein Framework für Unterhaltungen mit mehreren Agents als Abstraktion auf hoher Ebene zum Erstellen von LLM-Workflows.
LlamaIndex (Vorschau) Die Abfragepipeline von LlamaIndex bietet eine High-Level-Schnittstelle zum Erstellen von RAG-Workflows (Retrieval-Augmented Generation).

Bei benutzerdefinierten Agents, die nicht auf einer der frameworkspezifischen Vorlagen basieren, können Sie so vorgehen:

  1. Nutzerauthentifizierung:
  2. Agent-Instanz abrufen
  3. Unterstützte Vorgänge nachschlagen
  4. Agent abfragen
  5. (Falls zutreffend) Antworten vom Agent streamen.

Schritt 1: Nutzerauthentifizierung

Folgen Sie der Anleitung zum Einrichten der Umgebung.

Schritt 2: 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:

from vertexai import agent_engines

agent = agent_engines.get("RESOURCE_ID")

Alternativ können Sie den vollständigen Ressourcennamen des Agents angeben:

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

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

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

Schritt 3: Unterstützte Vorgänge

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:

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 Agent dokumentiert sind, die Sie aufrufen können. Das folgende Beispiel zeigt das Vorgangsschema für einen synchronen Vorgang:

Mit dem folgenden Befehl wird eine Liste von Schemas im JSON-Format bereitgestellt, die den Vorgängen des remote_app-Objekts entsprechen:

agent.operation_schemas()

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.

Schritt 4: Agent abfragen

So fragen Sie den Agenten mit einem der unterstützten Vorgänge ab (z.B. query):

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

Schritt 5: Antworten vom Agent streamen

Falls zutreffend, können Sie eine Antwort vom Agent über einen seiner Vorgänge streamen (z.B. 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

Schritt 6: Agent asynchron abfragen

Wenn Sie beim Entwickeln des Agents einen async_query-Vorgang definiert haben, wird die clientseitige asynchrone Abfrage des Agents 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."}

Schritt 7: 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.

Nächste Schritte