Benutzerdefinierten KI-Agenten erstellen

Wenn Sie mit der Agent Runtime einen benutzerdefinierten Agenten entwickeln, definieren Sie die Logik und die Funktionen des Agenten als Python-Klasse. Dieser klassenbasierte Ansatz bietet Ihnen die Flexibilität, eine beliebige Python-kompatible Bibliothek oder einen beliebigen Dienst einzubinden. Gleichzeitig übernimmt die Agent Platform die Bereitstellung, Skalierung und Verwaltung Ihrer Agenteninstanzen.

In den folgenden Schritten wird beschrieben, wie Sie eine benutzerdefinierte Vorlage zum Instanziieren von Agenten erstellen, die auf der Agent Platform bereitgestellt werden können:

1. Einfaches Beispiel
2. Optional: Streamantworten
3. Optional: Benutzerdefinierte Methoden registrieren
4. Optional: Typanmerkungen angeben
5. Optional: Traces an Cloud Trace senden
6. Optional: Mit Umgebungsvariablen arbeiten
7. Optional: In Secret Manager einbinden
8. Optional: Anmeldedaten verarbeiten
9. Optional: Fehlerbehebung

Einfaches Beispiel

Der folgende Python-Code ist eine Vorlage zum Instanziieren von Agenten, die auf der Agent Platform bereitgestellt werden können (Sie können der Variablen CLASS_NAME einen Wert wie MyAgent zuweisen):

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
        location: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.location = location

    def set_up(self):
        import vertexai
        from langchain_google_vertexai import ChatVertexAI
        from langgraph.prebuilt import create_react_agent

        vertexai.init(project=self.project, location=self.location)

        model = ChatVertexAI(model_name=self.model_name)
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

Überlegungen zur Bereitstellung

Beim Schreiben Ihrer Python-Klasse sind die folgenden drei Methoden wichtig:

1. __init__():
   • Verwenden Sie diese Methode nur für Parameter der Agentenkonfiguration. Sie können mit dieser Methode beispielsweise die Modellparameter und Sicherheitsattribute als Eingabeargumente von Ihren Nutzern erfassen. Sie können mit dieser Methode auch Parameter wie die Projekt-ID, die Region, die Anmeldedaten für die Anwendung und die API-Schlüssel erfassen.
   • Der Konstruktor gibt ein Objekt zurück, das „pickle-fähig“ sein muss, damit es für die Agent Runtime bereitgestellt werden kann. Daher sollten Sie Dienstclients initialisieren und Verbindungen zu Datenbanken in der Methode .set_up anstelle der Methode __init__ herstellen.
   • Diese Methode ist optional. Wenn sie nicht angegeben ist, verwendet die Agent Runtime den Standard-Python-Konstruktor für die Klasse.
2. set_up():
   • Sie müssen diese Methode verwenden, um die Logik für die Agenteninitialisierung zu definieren. Sie können mit dieser Methode beispielsweise Verbindungen zu Datenbanken oder abhängigen Diensten herstellen, abhängige Pakete importieren oder Daten vorausberechnen, die zur Verarbeitung von Abfragen verwendet werden.
   • Diese Methode ist optional. Wenn sie nicht angegeben ist, geht die Agent Runtime davon aus, dass der Agent keine .set_up-Methode aufrufen muss, bevor Nutzerabfragen bereitgestellt werden.
3. query() / stream_query():
   • Verwenden Sie query(), um die vollständige Antwort als einzelnes Ergebnis zurückzugeben.
   • Verwenden Sie stream_query(), um die Antwort in Teilen zurückzugeben, sobald sie verfügbar ist, und so ein Streaming zu ermöglichen. Die Methode stream_query muss ein iterierbares Objekt (z. B. einen Generator) zurückgeben, um das Streaming zu ermöglichen.
   • Sie können beide Methoden implementieren, wenn Sie sowohl Einzelantworten als auch Streaming-Interaktionen mit Ihrem Agenten unterstützen möchten.
   • Sie sollten dieser Methode einen eindeutigen Docstring zuweisen, der ihre Funktion definiert, ihre Attribute dokumentiert und Typenannotationen für ihre Eingaben bereitstellt. Vermeiden Sie Variablenargumente in der Methode query und stream_query.

Agenten lokal instanziieren

Sie können mit dem folgenden Code eine lokale Instanz Ihres Agenten erstellen:

agent = CLASS_NAME(
    model=model,  # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

Methode query testen

Sie können den Agenten testen, indem Sie Abfragen an die lokale Instanz senden:

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

print(response)

Die Antwort ist ein Wörterbuch, das etwa so aussieht:

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

Asynchrone Abfragen

Wenn Sie asynchron auf Abfragen antworten möchten, können Sie eine Methode (z. B. async_query) definieren, die eine Python-Coroutine zurückgibt. Die folgende Vorlage erweitert das einfache Beispiel, um asynchron zu antworten, und kann auf der Agent Platform bereitgestellt werden:

class AsyncAgent(CLASS_NAME):

    async def async_query(self, **kwargs):
        from langchain.load.dump import dumpd

        for chunk in self.graph.ainvoke(**kwargs):
            yield dumpd(chunk)

agent = AsyncAgent(
    model=model,                # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

Methode async_query testen

Sie können den Agenten lokal testen, indem Sie die Methode async_query aufrufen. Beispiel:

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

Die Antwort ist ein Wörterbuch, das etwa so aussieht:

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

Streamingantworten

Wenn Sie Antworten auf Abfragen streamen möchten, können Sie eine Methode namens stream_query definieren, die Antworten liefert. Die folgende Vorlage erweitert das einfache Beispiel, um Antworten zu streamen, und kann auf der Agent Platform bereitgestellt werden:

from typing import Iterable

class StreamingAgent(CLASS_NAME):

    def stream_query(self, **kwargs) -> Iterable:
        from langchain.load.dump import dumpd

        for chunk in self.graph.stream(**kwargs):
            yield dumpd(chunk)

agent = StreamingAgent(
    model=model,                # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

Bei der Verwendung der Streaming API sind einige wichtige Punkte zu beachten:

• Maximale Zeitüberschreitung: Die maximale Zeitüberschreitung für Streamingantworten beträgt 10 Minuten. Wenn Ihr Agent längere Verarbeitungszeiten benötigt, sollten Sie die Aufgabe in kleinere Teile aufteilen.
• Streamingmodelle und -ketten: Die Runnable-Schnittstelle von LangChain unterstützt das Streaming, sodass Sie Antworten nicht nur von Agenten, sondern auch von Modellen und Ketten streamen können.
• LangChain-Kompatibilität: Asynchrone Methoden wie die Methode LangChain's astream_event werden derzeit nicht unterstützt.
• Inhaltsgenerierung drosseln: Wenn Sie Probleme mit dem Backpressure haben (der Producer generiert Daten schneller als der Consumer sie verarbeiten kann), sollten Sie die Rate der Inhaltsgenerierung drosseln. So lassen sich Pufferüberläufe vermeiden und ein reibungsloses Streaming gewährleisten.

Methode stream_query testen

Sie können die Streamingabfrage lokal testen, indem Sie die Methode stream_query aufrufen und die Ergebnisse durchlaufen. Beispiel:

import pprint

for chunk in agent.stream_query(
    input="What is the exchange rate from US dollars to Swedish currency?"
):
    # Use pprint with depth=1 for a more concise, high-level view of the
    # streamed output.
    # To see the full content of the chunk, use:
    # print(chunk)
    pprint.pprint(chunk, depth=1)

Dieser Code gibt jeden Teil der Antwort aus, sobald er generiert wird. Die Ausgabe sieht etwa so aus:

{'actions': [...], 'messages': [...]}
{'messages': [...], 'steps': [...]}
{'messages': [...],
 'output': 'The exchange rate from US dollars to Swedish currency is 1 USD to '
           '10.5751 SEK. \n'}

In diesem Beispiel enthält jeder Teil unterschiedliche Informationen zur Antwort, z. B. die vom Agenten ausgeführten Aktionen, die ausgetauschten Nachrichten und die endgültige Ausgabe.

Streamingantworten asynchron

Wenn Sie Antworten asynchron streamen möchten, können Sie eine Methode (z. B. async_stream_query) definieren, die einen asynchronen Generator zurückgibt. Die folgende Vorlage erweitert das einfache Beispiel, um Antworten asynchron zu streamen, und kann auf der Agent Platform bereitgestellt werden:

class AsyncStreamingAgent(CLASS_NAME):

    async def async_stream_query(self, **kwargs):
        from langchain.load.dump import dumpd

        for chunk in self.graph.astream(**kwargs):
            yield dumpd(chunk)

agent = AsyncStreamingAgent(
    model=model,                # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

Methode async_stream_query testen

Ähnlich wie beim Testen von Streamingabfragen können Sie den Agenten lokal testen, indem Sie die Methode async_stream_query aufrufen und die Ergebnisse durchlaufen. Beispiel:

import pprint

async for chunk in agent.async_stream_query(
    input="What is the exchange rate from US dollars to Swedish currency?"
):
    # Use pprint with depth=1 for a more concise, high-level view of the
    # streamed output.
    # To see the full content of the chunk, use:
    # print(chunk)
    pprint.pprint(chunk, depth=1)

Dieser Code gibt jeden Teil der Antwort aus, sobald er generiert wird. Die Ausgabe sieht etwa so aus:

{'actions': [...], 'messages': [...]}
{'messages': [...], 'steps': [...]}
{'messages': [...],
 'output': 'The exchange rate from US dollars to Swedish currency is 1 USD to '
           '10.5751 SEK. \n'}

Benutzerdefinierte Methoden registrieren

Standardmäßig werden die Methoden query und stream_query als Vorgänge im bereitgestellten Agenten registriert. Sie können das Standardverhalten überschreiben und die Menge der zu registrierenden Vorgänge mit der Methode register_operations definieren. Vorgänge können entweder als Standardausführungsmodus (durch einen leeren String "") oder als Streamingausführungsmodus ("stream") registriert werden.

Wenn Sie mehrere Vorgänge registrieren möchten, können Sie eine Methode namens register_operations definieren, in der die Methoden aufgeführt sind, die Nutzern zur Verfügung gestellt werden sollen, wenn der Agent bereitgestellt wird. Im folgenden Beispielcode führt die register_operations Methode dazu, dass der bereitgestellte Agent query und get_state als Vorgänge registriert, die synchron ausgeführt werden, und stream_query und get_state_history als Vorgänge, die die Antworten streamen:

from typing import Iterable

class CustomAgent(StreamingAgent):

    def get_state(self) -> dict: # new synchronous method
        return self.graph.get_state(**kwargs)._asdict()

    def get_state_history(self) -> Iterable: # new streaming operation
        for state_snapshot in self.graph.get_state_history(**kwargs):
            yield state_snapshot._asdict()

    def register_operations(self):
        return {
            # The list of synchronous operations to be registered
            "": ["query", "get_state"],
            # The list of streaming operations to be registered
            "stream": ["stream_query", "get_state_history"],
        }

Hinweis: Sie müssen Methoden nicht sowohl für Standard- als auch für Streaming vorgänge registrieren. Wenn Sie beispielsweise nur Standardvorgänge unterstützen möchten, können Sie in der register_operations Methode {"": ["query", "get_state"]} zurückgeben, ohne eine entsprechende Liste von Methoden für "stream" anzugeben.

Sie können die benutzerdefinierten Methoden testen, indem Sie sie direkt in der lokalen Instanz des Agenten aufrufen, ähnlich wie beim Testen der query und stream_query Methoden.

Typanmerkungen angeben

Sie können Typanmerkungen verwenden, um die erwarteten Eingabe- und Ausgabetypen Ihrer Agentenmethoden anzugeben. Wenn der Agent bereitgestellt wird, werden in der Ein- und Ausgabe der vom Agenten unterstützten Vorgänge nur JSON-serialisierbare Typen unterstützt. Die Schemas der Ein- und Ausgaben können mit TypedDict oder Pydantic-Modellen annotiert werden.

Im folgenden Beispiel annotieren wir die Eingabe als TypedDict und konvertieren die Rohausgabe von .get_state (ein NamedTuple) mit der Methode ._asdict() in ein serialisierbares Wörterbuch:

from typing import Any, Dict, TypedDict

# schemas.py
class RunnableConfig(TypedDict, total=False):
    metadata: Dict[str, Any]
    configurable: Dict[str, Any]

# agents.py
class AnnotatedAgent(CLASS_NAME):

    def get_state(self, config: RunnableConfig) -> dict:
        return self.graph.get_state(config=config)._asdict()

    def register_operations(self):
        return {"": ["query", "get_state"]}

Traces an Cloud Trace senden

Wenn Sie Traces mit Instrumentierungsbibliotheken, die OpenTelemetry unterstützen, an Cloud Trace senden möchten, können Sie sie in der Methode .set_up importieren und initialisieren. Für gängige Agenten-Frameworks können Sie die OpenTelemetry Google Cloud Integration in Kombination mit einem Instrumentierungs Framework wie OpenInference oder OpenLLMetry verwenden.

Die folgende Vorlage ist beispielsweise eine Änderung des einfachen Beispiels, um Traces nach Cloud Trace zu exportieren:

OpenInference

Installieren Sie zuerst das erforderliche Paket mit pip:

pip install openinference-instrumentation-langchain==0.1.34

Importieren und initialisieren Sie dann das Instrument:

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
        location: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.location = location

    def set_up(self):
        # The additional code required for tracing instrumentation.
        from opentelemetry import trace
        from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
        from opentelemetry.sdk.trace import TracerProvider
        from opentelemetry.sdk.trace.export import SimpleSpanProcessor
        from openinference.instrumentation.langchain import LangChainInstrumentor
        import google.cloud.trace_v2 as cloud_trace_v2
        import google.auth

        credentials, _ = google.auth.default()

        trace.set_tracer_provider(TracerProvider())
        cloud_trace_exporter = CloudTraceSpanExporter(
            project_id=self.project,
            client=cloud_trace_v2.TraceServiceClient(
                credentials=credentials.with_quota_project(self.project),
            ),
        )
        trace.get_tracer_provider().add_span_processor(
            SimpleSpanProcessor(cloud_trace_exporter)
        )
        LangChainInstrumentor().instrument()
        # end of additional code required

        import vertexai
        from langchain_google_vertexai import ChatVertexAI
        from langgraph.prebuilt import create_react_agent

        vertexai.init(project=self.project, location=self.location)

        model = ChatVertexAI(model_name=self.model_name)
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

OpenLLMetry

Installieren Sie zuerst das erforderliche Paket mit pip:

pip install opentelemetry-instrumentation-langchain==0.38.10

Importieren und initialisieren Sie dann das Instrument:

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
        location: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.location = location

    def set_up(self):
        # The additional code required for tracing instrumentation.
        from opentelemetry import trace
        from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
        from opentelemetry.sdk.trace import TracerProvider
        from opentelemetry.sdk.trace.export import SimpleSpanProcessor
        from opentelemetry.instrumentation.langchain import LangchainInstrumentor
        import google.cloud.trace_v2 as cloud_trace_v2
        import google.auth

        credentials, _ = google.auth.default()

        trace.set_tracer_provider(TracerProvider())
        cloud_trace_exporter = CloudTraceSpanExporter(
            project_id=self.project,
            client=cloud_trace_v2.TraceServiceClient(
                credentials=credentials.with_quota_project(self.project),
            ),
        )
        trace.get_tracer_provider().add_span_processor(
            SimpleSpanProcessor(cloud_trace_exporter)
        )
        LangchainInstrumentor().instrument()
        # end of additional code required

        import vertexai
        from langchain_google_vertexai import ChatVertexAI
        from langgraph.prebuilt import create_react_agent

        vertexai.init(project=self.project, location=self.location)

        model = ChatVertexAI(model_name=self.model_name)
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

Mit Umgebungsvariablen arbeiten

Wenn Sie Umgebungsvariablen festlegen möchten, müssen sie während der Entwicklung über os.environ verfügbar sein. Folgen Sie beim Bereitstellen des Agenten der Anleitung unter Umgebungsvariablen definieren.

In Secret Manager einbinden

So binden Sie Secret Manager ein:

1. Installieren Sie die Clientbibliothek:

   pip install google-cloud-secret-manager

2. Folgen Sie der Anleitung unter Rollen für einen bereitgestellten Agenten gewähren , um dem Dienstkonto über die Google Cloud Konsole die Rolle „Secret Manager Secret Accessor“ (roles/secretmanager.secretAccessor) zu gewähren.

3. Importieren und initialisieren Sie den Client in der Methode .set_up und rufen Sie das entsprechende Secret bei Bedarf ab. Die folgende Vorlage ist beispielsweise eine Änderung des einfachen Beispiels, um einen API-Schlüssel für ChatAnthropic zu verwenden, der in Secret Manager gespeichert wurde:

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.secret_id = secret_id # <- new

    def set_up(self):
        from google.cloud import secretmanager
        from langchain_anthropic import ChatAnthropic
        from langgraph.prebuilt import create_react_agent

        # Get the API Key from Secret Manager here.
        self.secret_manager_client = secretmanager.SecretManagerServiceClient()
        secret_version = self.secret_manager_client.access_secret_version(request={
            "name": "projects/PROJECT_ID/secrets/SECRET_ID/versions/SECRET_VERSION",
        })
        # Use the API Key from Secret Manager here.
        model = ChatAnthropic(
            model_name=self.model_name,
            model_kwargs={"api_key": secret_version.payload.data.decode()},  # <- new
        )
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

Anmeldedaten verarbeiten

Wenn der Agent bereitgestellt wird, muss er möglicherweise verschiedene Arten von Anmeldedaten verarbeiten:

1. Standardanmeldedaten für Anwendungen (Application Default Credentials, ADC) , die häufig von Dienstkonten stammen,
2. OAuth, das häufig von Nutzerkonten stammt, und
3. Identitätsanbieter für Anmeldedaten von externen Konten (Identitätsföderation von Arbeitslasten).

Standardanmeldedaten für Anwendungen

import google.auth

credentials, project = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)

Sie können im Code so verwendet werden:

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str = "meta/llama3-405b-instruct-maas",
        tools: Sequence[Callable],
        location: str,
        project: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.endpoint = f"https://{location}-aiplatform.googleapis.com"
        self.base_url = f'{self.endpoint}/v1beta1/projects/{project}/locations/{location}/endpoints/openapi'

    def query(self, **kwargs):
        import google.auth
        from langchain_openai import ChatOpenAI
        from langgraph.prebuilt import create_react_agent

        # Note: the credential lives for 1 hour by default.
        # After expiration, it must be refreshed.
        creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
        creds.refresh(google.auth.transport.requests.Request())

        model = ChatOpenAI(
            model=self.model_name,
            base_url=self.base_url,
            api_key=creds.token,  # Use the token from the credentials here.
        )
        graph = create_react_agent(model, tools=self.tools)
        return graph.invoke(**kwargs)

Weitere Informationen finden Sie unter Funktionsweise von Standardanmeldedaten für Anwendungen works.

OAuth

Nutzeranmeldedaten werden in der Regel mit OAuth 2.0 abgerufen.

Wenn Sie ein Zugriffstoken haben (z. B. von oauthlib), können Sie eine google.oauth2.credentials.Credentials Instanz erstellen. Wenn Sie außerdem ein Aktualisierungstoken erhalten, können Sie auch das Aktualisierungstoken und den Token-URI angeben, damit die Anmeldedaten automatisch aktualisiert werden können:

credentials = google.oauth2.credentials.Credentials(
    token="ACCESS_TOKEN",
    refresh_token="REFRESH_TOKEN",  # Optional
    token_uri="TOKEN_URI",          # E.g. "https://oauth2.googleapis.com/token"
    client_id="CLIENT_ID",          # Optional
    client_secret="CLIENT_SECRET"   # Optional
)

Hier basieren TOKEN_URI, CLIENT_ID und CLIENT_SECRET auf OAuth-Client anmeldedaten erstellen.

Wenn Sie kein Zugriffstoken haben, können Sie mit google_auth_oauthlib.flow den OAuth 2.0-Autorisierungsgewährungs Ablauf ausführen, um eine entsprechende google.oauth2.credentials.Credentials Instanz abzurufen:

from google.cloud import secretmanager
from google_auth_oauthlib.flow import InstalledAppFlow
import json

# Get the client config from Secret Manager here.
secret_manager_client = secretmanager.SecretManagerServiceClient()
secret_version = client.access_secret_version(request={
    "name": "projects/PROJECT_ID/secrets/SECRET_ID/versions/SECRET_VERSION",
})
client_config = json.loads(secret_version.payload.data.decode())

# Create flow instance to manage the OAuth 2.0 Authorization Grant Flow steps.
flow = InstalledAppFlow.from_client_config(
    client_config,
    scopes=['https://www.googleapis.com/auth/cloud-platform'],
    state="OAUTH_FLOW_STATE"  # from flow.authorization_url(...)
)

# You can get the credentials from the flow object.
credentials: google.oauth2.credentials.Credentials = flow.credentials

# After obtaining the credentials, you can then authorize API requests on behalf
# of the given user or service account. For example, to authorize API requests
# to vertexai services, you'll specify it in vertexai.init(credentials=)
import vertexai

vertexai.init(
    project="PROJECT_ID",
    location="LOCATION",
    credentials=credentials, # specify the credentials here
)

Weitere Informationen finden Sie in der Dokumentation zum google_auth_oauthlib.flow Modul.

Identitätsanbieter

Wenn Sie Nutzer über E-Mail/Passwort, Telefonnummer, soziale Netzwerke wie Google, Facebook oder GitHub oder einen benutzerdefinierten Authentifizierungsmechanismus authentifizieren möchten, können Sie Identity Platform oder Firebase Authentication oder einen beliebigen Identitätsanbieter verwenden, der OpenID Connect (OIDC) unterstützt.

Weitere Informationen finden Sie unter Über einen OIDC-Identitätsanbieter auf Ressourcen zugreifen.

Fehlerbehebung

Damit API-Fehler in einem strukturierten JSON-Format zurückgegeben werden, empfehlen wir, die Fehlerbehandlung in Ihrem Agentencode mit einem try...except-Block zu implementieren, der in einen Decorator abstrahiert werden kann.

Die Agent Platform kann zwar verschiedene Statuscodes intern verarbeiten, aber Python bietet keine standardisierte Möglichkeit, Fehler mit zugehörigen HTTP-Statuscodes für alle Ausnahmetypen darzustellen. Der Versuch, alle möglichen Python-Ausnahmen den HTTP-Statuscodes im zugrunde liegenden Dienst zuzuordnen, wäre komplex und schwer zu verwalten.

Ein skalierbarerer Ansatz besteht darin, relevante Ausnahmen explizit in Ihren Agentenmethoden abzufangen oder einen wiederverwendbaren Decorator wie error_wrapper zu verwenden. Anschließend können Sie entsprechende Statuscodes zuordnen (z. B. durch Hinzufügen von code und error Attributen zu benutzerdefinierten Ausnahmen oder durch die spezifische Behandlung von Standardausnahmen ) und den Fehler als JSON-Wörterbuch für den Rückgabewert formatieren. Dazu sind nur minimale Codeänderungen in den Agentenmethoden selbst erforderlich, oft müssen Sie nur den Decorator hinzufügen.

Im Folgenden sehen Sie ein Beispiel dafür, wie Sie die Fehlerbehandlung in Ihrem Agenten implementieren können:

from functools import wraps
import json

def error_wrapper(func):
    @wraps(func)  # Preserve original function metadata
    def wrapper(*args, **kwargs):
        try:
            # Execute the original function with its arguments
            return func(*args, **kwargs)
        except Exception as err:
            error_code = getattr(err, 'code')
            error_message = getattr(err, 'error')

            # Construct the error response dictionary
            error_response = {
                "error": {
                    "code": error_code,
                    "message": f"'{func.__name__}': {error_message}"
                }
            }
            # Return the Python dictionary directly.
            return error_response

    return wrapper

# Example exception
class SessionNotFoundError(Exception):
    def __init__(self, session_id, message="Session not found"):
        self.code = 404
        self.error = f"{message}: {session_id}"
        super().__init__(self.error)

# Example Agent Class
class MyAgent:
    @error_wrapper
    def get_session(self, session_id: str):
        # Simulate the condition where the session isn't found
        raise SessionNotFoundError(session_id=session_id)


# Example Usage: Session Not Found
agent = MyAgent()
error_result = agent.get_session(session_id="nonexistent_session_123")
print(json.dumps(error_result, indent=2))

Der vorherige Code führt zu folgender Ausgabe: json { "error": { "code": 404, "message": "Invocation error in 'get_session': Session not found: nonexistent_session_123" } }

Nächste Schritte

Anleitung

Agenten bereitstellen

Informationen zu den fünf Möglichkeiten, einen Agenten in der Agent Platform Runtime bereitzustellen, je nach Ihren Entwicklungsanforderungen.

Anleitung

Agenten verwenden

Agenten mit der Agent Platform Runtime verwenden.

Anleitung

Agenten bewerten

Einen einfachen Agenten erstellen und bereitstellen und den Gen AI Evaluation Service verwenden, um den Agenten zu bewerten.

Fehlerbehebung

Fehlerbehebung bei der Agentenerstellung

Informationen zum Beheben häufiger Fehler beim Erstellen benutzerdefinierter Agenten.

Ressource

Support anfordern

Ressourcen und Support für die Google Agent Platform.