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:
__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.
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.
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étodostream_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
estream_query
.
- Use
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.