Anwendungsvorlage anpassen

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

Eine Anwendungsvorlage in der Reasoning Engine wird als Python-Klasse definiert. Der folgende Python-Code ist beispielsweise eine LangChain-Anwendung, die auf Vertex AI bereitgestellt werden kann (Sie können der Variable CLASS_NAME einen Wert wie MyAgent zuweisen):

from typing import Any, Callable, Iterable, 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):
        """All unpickle-able logic should go here.

        The .set_up() method should not be called for an object that is being
        prepared for deployment.
        """
        import vertexai
        from langchain_google_vertexai import ChatVertexAI
        from langchain.agents import AgentExecutor
        from langchain.agents.format_scratchpad.tools import format_to_tool_messages
        from langchain.agents.output_parsers.tools import ToolsAgentOutputParser
        from langchain.tools.base import StructuredTool
        from langchain_core import prompts

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

        prompt = {
            "input": lambda x: x["input"],
            "agent_scratchpad": (
                lambda x: format_to_tool_messages(x["intermediate_steps"])
            ),
        } | prompts.ChatPromptTemplate.from_messages([
            ("user", "{input}"),
            prompts.MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])

        llm = ChatVertexAI(model_name=self.model_name)
        if self.tools:
            llm = llm.bind_tools(tools=self.tools)

        self.agent_executor = AgentExecutor(
            agent=prompt | llm | ToolsAgentOutputParser(),
            tools=[StructuredTool.from_function(tool) for tool in self.tools],
        )

    def query(self, input: str):
        """Query the application.

        Args:
            input: The user prompt.

        Returns:
            The output of querying the application with the given input.
        """
        return self.agent_executor.invoke(input={"input": input})

    def stream_query(self, input: str) -> Iterable[Any]:
        """Query the application and stream the output.

        Args:
            input: The user prompt.

        Yields:
            Chunks of the response as they become available.
        """
        for chunk in self.agent_executor.stream(input={"input": input}):
            yield chunk

Beim Schreiben Ihrer Python-Klasse sind die folgenden drei Methoden für die Reasoning-Engine wichtig:

1. __init__():
   • Verwenden Sie diese Methode nur für Parameter der Anwendungskonfiguration. Sie können mit dieser Methode beispielsweise die Modellparameter und Sicherheitsattribute als Eingabeargumente von Ihren Nutzern erfassen. Sie können diese Methode auch verwenden, um Parameter wie die Projekt-ID, die Region, die Anmeldedaten für die Anwendung und die API-Schlüssel zu erfassen.
   • Der Konstruktor gibt ein Objekt zurück, das "pickle-fähig" sein muss, damit es für die Reasoning-Engine bereitgestellt werden kann. Daher sollten Sie Dienstclients initialisieren und Verbindungen zu Datenbanken in der Methode .set_up und nicht in der Methode __init__ herstellen.
   • Diese Methode ist optional. Wenn er nicht angegeben ist, verwendet Vertex AI den Standard-Python-Konstruktor für die Klasse.
2. set_up():
   • Sie müssen diese Methode verwenden, um die Logik für die Anwendungsinitialisierung 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 Vertex AI davon aus, dass die Anwendung keine .set_up-Methode aufrufen muss, bevor Nutzerabfragen bereitgestellt werden.
3. query()/stream_query():
   • Mit query() wird die vollständige Antwort als einzelnes Ergebnis zurückgegeben.
   • Mit stream_query() können Sie die Antwort in Teilen zurückgeben, sobald sie verfügbar ist. Die stream_query-Methode muss ein iterierbares Objekt (z. B. einen Generator) zurückgeben, um Streaming zu ermöglichen.
   • Sie können beide Methoden implementieren, wenn Sie sowohl Interaktionen mit einer einzelnen Antwort als auch Streaminginteraktionen mit Ihrer Anwendung 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 den Methoden query und stream_query.

Anwendung lokal testen

Instanziieren Sie die Anwendung im lokalen Speicher mit dem folgenden Code:

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

query-Methode testen

Sie können die Anwendung testen, indem Sie Testabfragen an die lokale Instanz senden:

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

Die Antwort ist ein Wörterbuch, das in 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."}

stream_query-Methode testen

Sie können die Streamingabfrage lokal testen, indem Sie die Methode stream_query aufrufen und die Ergebnisse durchgehen. 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)

Mit diesem Code wird jeder Teil der Antwort ausgegeben, sobald er generiert wird. Die Ausgabe sieht in 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 verschiedene Informationen zur Antwort, z. B. die vom Kundenservicemitarbeiter ausgeführten Aktionen, die ausgetauschten Nachrichten und die endgültige Ausgabe.

Streaming API

Bei der Verwendung der Streaming API ist Folgendes zu beachten:

• Maximale Zeitüberschreitung: Die maximale Zeitüberschreitung für Streamingantworten beträgt 10 Minuten. Wenn Ihre Anwendung längere Verarbeitungszeiten erfordert, sollten Sie die Aufgabe in kleinere Teile aufteilen.
• Streaming von Modellen und Ketten: Die Runnable-Schnittstelle von LangChain unterstützt Streaming. So können Sie nicht nur Antworten von Agents, sondern auch von Modellen und Ketten streamen.
• Kompatibilität mit LangChain: Die astream_event-Methode von LangChain wird nicht unterstützt.
• Inhaltsgenerierung drosseln: Wenn Backpressure-Probleme auftreten (d. h. der Produzent generiert Daten schneller, als der Verbraucher sie verarbeiten kann), drosseln Sie die Inhaltsgenerierungsrate. So lassen sich Pufferüberläufe vermeiden und ein reibungsloser Streamingvorgang ermöglichen.

Methodennamen anpassen

Standardmäßig sind die Methoden query und stream_query als Vorgänge in der bereitgestellten Anwendung registriert. Sie können das Standardverhalten überschreiben und die zu registrierenden Vorgänge mit der Methode register_operations definieren. Vorgänge können als Standardaufrufmodus (ein leerer String, "") oder Streamingaufrufmodus ("stream") registriert werden.

Im folgenden Beispielcode führt die register_operations-Methode dazu, dass die bereitgestellte Anwendung custom_method_1 und custom_method_2 als Vorgänge für Standardaufrufe und custom_stream_method_1 und custom_stream_method_2 als Vorgänge für Streamingaufrufe anbietet. Diese Vorgänge ersetzen die Standardvorgänge query und stream_query.

from typing import Dict, List, Any, Iterable

class CLASS_NAME:
    # ... other methods ...

    def custom_method_1(...):
        # ...

    def custom_method_2(...):
        # ...

    def custom_stream_method_1(...) -> Iterable[Any]:
        # ...

    def custom_stream_method_2(...) -> Iterable[Any]:
        # ...

    def register_operations(self) -> Dict[str, List[str]]:
        return {
            "": [
                "custom_method_1", "custom_method_2",
            ],
            "stream": [
                "custom_stream_method_1", "custom_stream_method_2",
            ],
        }

Sie können die Anwendung testen, indem Sie Testabfragen an die Instanz senden:

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

for chunk in agent.custom_stream_method_1(
    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)

Sie müssen keine Methoden für beide Aufruftypen registrieren. Wenn Sie beispielsweise nur Standardanrufe unterstützen möchten, gehen Sie so vor:

from typing import Dict, List, Any

class CLASS_NAME:
    # ... other methods ...

    def custom_method_1(...):
        # ...

    def custom_method_2(...):
        # ...

    def custom_stream_method_1(...) -> Iterable[Any]:
        # ...

    def custom_stream_method_2(...) -> Iterable[Any]:
        # ...

    def register_operations(self) -> Dict[str, List[str]]:
        return {
            # The list of synchronous methods to be registered as operations.
            "": [
                "custom_method_1", "custom_method_2",
            ],
        }

In diesem Beispiel werden nur custom_method_1 und custom_method_2 als Vorgänge in bereitgestellten Anwendungen freigegeben.

Nächste Schritte

• Stellen Sie die Anwendung bereit.