Personalizar um modelo de aplicativo

Em Desenvolver o aplicativo, usamos um modelo pré-criado (por exemplo, reasoning_engines.LangchainAgent) para desenvolver um aplicativo. Nesta seção, você verá as etapas para personalizar seu próprio modelo de aplicativo. Isso poderá ser útil se você precisar ir além do que o modelo pré-criado oferece.

Um modelo de aplicativo no Reasoning Engine é definido como uma classe Python. Para dar um exemplo, o código Python a seguir é um exemplo de um aplicativo LangChain que pode ser implantado na Vertex AI (é possível atribuir um valor como MyAgent à variável CLASS_NAME):

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

Ao escrever a classe Python, os três métodos a seguir são importantes para o Reasoning Engine:

  1. __init__():
    • Use esse método apenas para parâmetros de configuração do aplicativo. Por exemplo, é possível usar esse método para coletar os parâmetros do modelo e os atributos de segurança como argumentos de entrada dos usuários. Também é possível usar esse método para coletar parâmetros como o ID do projeto, a região, as credenciais de aplicativo e as chaves de API.
    • O construtor retorna um objeto que precisa ser "serializável" para que possa ser implantado no Reasoning Engine. Portanto, inicialize clientes de serviço e estabeleça conexões com bancos de dados no método .set_up no lugar do método __init__.
    • Esse método é opcional. Se não for especificado, a Vertex AI usará o construtor padrão do Python para a classe.
  2. set_up():
    • Use esse método para definir a lógica de inicialização do aplicativo. Por exemplo, esse método é usado para estabelecer conexões com bancos de dados ou serviços dependentes, importar pacotes dependentes ou pré-computar dados usados para atender a consultas.
    • Esse método é opcional. Se não for especificado, a Vertex AI vai presumir que o aplicativo não precisa chamar um método .set_up antes de atender às consultas do usuário.
  3. query()/stream_query():
    • Use query() para retornar a resposta completa como um único resultado.
    • Use stream_query() para retornar a resposta em partes à medida que ela fica disponível, ativando uma experiência de streaming. O método stream_query precisa retornar um objeto iterável (por exemplo, um gerador) para ativar o streaming.
    • Você pode implementar os dois métodos se quiser oferecer suporte a interações de resposta única e de streaming com seu aplicativo.
    • Forneça a esse método uma docstring clara que defina o que ele faz, documente os atributos e forneça anotações de tipo para as entradas. Evite argumentos variáveis nos métodos query e stream_query.

Testar o aplicativo localmente

Instancie o aplicativo na memória local usando o seguinte código:

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

Testar o método query

Teste o aplicativo enviando consultas de teste para a instância local:

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

A resposta é um dicionário semelhante a este:

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

Testar o método stream_query

É possível testar a consulta de streaming localmente chamando o método stream_query e iterando os resultados. Veja um exemplo:

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)

Esse código imprime cada bloco da resposta conforme é gerado. A saída pode parecer esta:

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

Neste exemplo, cada bloco contém informações diferentes sobre a resposta, como as ações realizadas pelo agente, as mensagens trocadas e a saída final.

API Streaming

Confira alguns pontos importantes ao usar a API de streaming:

  • Tempo limite máximo: o tempo limite máximo para respostas de streaming é de 10 minutos. Se o aplicativo exigir tempos de processamento mais longos, considere dividir a tarefa em partes menores.
  • Streaming de modelos e cadeias: a interface Runnable do LangChain oferece suporte a streaming, para que você possa fazer streaming de respostas não apenas de agentes, mas também de modelos e cadeias.
  • Compatibilidade com o LangChain: o método astream_event do LangChain não é compatível.
  • Limitar a geração de conteúdo: se você encontrar problemas de contrapressão (em que o produtor gera dados mais rápido do que o consumidor pode processá-los), limite a taxa de geração de conteúdo. Isso pode ajudar a evitar transbordamentos de buffer e garantir uma experiência de streaming tranquila.

Personalizar nomes de métodos

Por padrão, os métodos query e stream_query são registrados como operações no aplicativo implantado. É possível substituir o comportamento padrão e definir o conjunto de operações a serem registradas usando o método register_operations. As operações podem ser registradas como modos de invocação padrão (representado por uma string vazia "") ou de streaming ("stream").

No exemplo de código abaixo, o método register_operations vai resultar no aplicativo implantado oferecendo custom_method_1 e custom_method_2 como operações para chamadas padrão e custom_stream_method_1 e custom_stream_method_2 como operações para chamadas de streaming. Essas operações substituem as operações padrão query e 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",
            ],
        }

Teste o aplicativo enviando consultas de teste para a instância da seguinte maneira:

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)

Não é necessário registrar métodos para os dois tipos de invocação. Por exemplo, para oferecer suporte apenas a chamadas padrão, faça o seguinte:

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

Neste exemplo, apenas custom_method_1 e custom_method_2 são expostos como operações em aplicativos implantados.

A seguir