Personaliza una plantilla de aplicación

Organiza tus páginas con colecciones Guarda y categoriza el contenido según tus preferencias.

Una plantilla de aplicación en Reasoning Engine se define como una clase de Python. Para dar un ejemplo, el siguiente código de Python es un ejemplo de una aplicación LangChain que se puede implementar en Vertex AI (puedes asignar a la variable CLASS_NAME un valor como MyAgent):

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

Cuando escribes tu clase de Python, los siguientes tres métodos son importantes para el motor de razonamiento:

1. __init__():
   • Usa este método solo para los parámetros de configuración de la aplicación. Por ejemplo, puedes usar este método para recopilar los parámetros del modelo y los atributos de seguridad como argumentos de entrada de tus usuarios. También puedes usar este método para recopilar parámetros, como el ID del proyecto, la región, las credenciales de la aplicación y las claves de API.
   • El constructor muestra un objeto que debe ser "pickle" para que se pueda implementar en el motor de razonamiento. Por lo tanto, debes inicializar los clientes de servicio y establecer conexiones con las bases de datos en el método .set_up, en lugar de hacerlo en el método __init__.
   • Este método es opcional. Si no se especifica, Vertex AI usa el constructor de Python predeterminado para la clase.
2. set_up():
   • Debes usar este método para definir la lógica de inicialización de la aplicación. Por ejemplo, usa este método para establecer conexiones con bases de datos o servicios dependientes, importar paquetes dependientes o datos de procesamiento previo que se usan para entregar consultas.
   • Este método es opcional. Si no se especifica, Vertex AI supone que la aplicación no necesita llamar a un método .set_up antes de entregar consultas de los usuarios.
3. query()/stream_query():
   • Usa query() para mostrar la respuesta completa como un solo resultado.
   • Usa stream_query() para mostrar la respuesta en fragmentos a medida que esté disponible, lo que habilita una experiencia de transmisión. El método stream_query debe mostrar un objeto iterable (por ejemplo, un generador) para habilitar la transmisión.
   • Puedes implementar ambos métodos si deseas admitir interacciones de transmisión y de respuesta única con tu aplicación.
   • Debes proporcionar a este método una docstring clara que defina lo que hace, documenta sus atributos y proporciona anotaciones de tipo para sus entradas. Evita los argumentos variables en los métodos query y stream_query.

Prueba la aplicación de forma local

Crea una instancia de la aplicación en la memoria local con el siguiente código:

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

Prueba el método query

Puedes probar la aplicación si envías consultas de prueba a la instancia local:

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

La respuesta es un diccionario similar al siguiente:

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

Prueba el método stream_query

Para probar la consulta de transmisión de forma local, llama al método stream_query y itera por los resultados. Por ejemplo:

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)

Este código imprime cada fragmento de la respuesta a medida que se genera. El resultado podría verse de la siguiente manera:

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

En este ejemplo, cada fragmento contiene información diferente sobre la respuesta, como las acciones que realizó el agente, los mensajes que se intercambiaron y el resultado final.

API de Streaming

A continuación, se incluyen algunos aspectos clave que debes tener en cuenta cuando uses la API de transmisión:

• Tiempo de espera máximo: El tiempo de espera máximo para las respuestas de transmisión es de 10 minutos. Si tu aplicación requiere tiempos de procesamiento más largos, considera dividir la tarea en partes más pequeñas.
• Modelos y cadenas de transmisión: La interfaz Runnable de LangChain admite la transmisión, por lo que puedes transmitir respuestas no solo de agentes, sino también de modelos y cadenas.
• Compatibilidad con LangChain: Ten en cuenta que el método astream_event de LangChain no es compatible.
• Limita la generación de contenido: Si tienes problemas de contrapresión (en los que el productor genera datos más rápido de lo que el consumidor puede procesarlos), limita la tasa de generación de contenido. Esto puede ayudar a evitar desbordamientos del búfer y garantizar una experiencia de transmisión fluida.

Personaliza los nombres de los métodos

De forma predeterminada, los métodos query y stream_query se registran como operaciones en la aplicación implementada. Puedes anular el comportamiento predeterminado y definir el conjunto de operaciones que se registrarán con el método register_operations. Las operaciones se pueden registrar como modos de invocación estándar (representados por una cadena vacía "") o de transmisión ("stream").

En el siguiente código de ejemplo, el método register_operations hará que la aplicación implementada ofrezca custom_method_1 y custom_method_2 como operaciones para llamadas estándar, y custom_stream_method_1 y custom_stream_method_2 como operaciones para llamadas de transmisión. Estas operaciones reemplazan las operaciones predeterminadas query y 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",
            ],
        }

Para probar la aplicación, envía consultas de prueba a la instancia de la siguiente manera:

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)

No es necesario que registres métodos para ambos tipos de invocación. Por ejemplo, para admitir solo llamadas estándar, puedes hacer lo siguiente:

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",
            ],
        }

En este ejemplo, solo custom_method_1 y custom_method_2 se exponen como operaciones en las aplicaciones implementadas.

¿Qué sigue?

• Implementa la aplicación