reasoning_engines.LangchainAgent) untuk
mengembangkan aplikasi. Di bagian ini, kita akan membahas langkah-langkah untuk menyesuaikan
template aplikasi Anda sendiri. Hal ini mungkin berguna jika Anda memiliki kebutuhan yang melebihi
apa yang disediakan template bawaan.
Template aplikasi di Reasoning Engine ditentukan sebagai class Python. Untuk
memberikan contoh, kode Python berikut adalah contoh aplikasi LangChain
yang dapat di-deploy di Vertex AI (Anda dapat memberikan
nilai variabel CLASS_NAME seperti 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
Saat menulis class Python, tiga metode berikut penting untuk mesin penalaran:
__init__():- Gunakan metode ini hanya untuk parameter konfigurasi aplikasi. Misalnya, Anda dapat menggunakan metode ini untuk mengumpulkan parameter model dan atribut keamanan sebagai argumen input dari pengguna. Anda juga dapat menggunakan metode ini untuk mengumpulkan parameter seperti project ID, region, kredensial aplikasi, dan kunci API.
- Konstruktor menampilkan objek yang harus "dapat di-pickle" agar dapat di-deploy ke mesin penalaran. Oleh karena itu, Anda harus melakukan inisialisasi klien layanan
dan membuat koneksi ke database dalam metode
.set_up, bukan dalam metode__init__. - Metode ini bersifat opsional. Jika tidak ditentukan, Vertex AI akan menggunakan konstruktor Python default untuk class.
set_up():- Anda harus menggunakan metode ini untuk menentukan logika inisialisasi aplikasi. Misalnya, Anda menggunakan metode ini untuk membuat koneksi ke database atau layanan dependen, mengimpor paket dependen, atau melakukan prakomputasi data yang digunakan untuk menayangkan kueri.
- Metode ini bersifat opsional. Jika tidak ditentukan, Vertex AI akan mengasumsikan bahwa aplikasi tidak perlu memanggil metode
.set_upsebelum menayangkan kueri pengguna.
query()/stream_query():- Gunakan
query()untuk menampilkan respons lengkap sebagai satu hasil. - Gunakan
stream_query()untuk menampilkan respons dalam potongan saat tersedia, sehingga memungkinkan pengalaman streaming. Metodestream_queryharus menampilkan objek yang dapat di-iterasi (misalnya generator) untuk mengaktifkan streaming. - Anda dapat menerapkan kedua metode tersebut jika ingin mendukung interaksi streaming dan respons tunggal dengan aplikasi Anda.
- Anda harus memberikan docstring yang jelas ke metode ini yang menentukan fungsinya,
mendokumentasikan atributnya, dan memberikan anotasi jenis untuk inputnya.
Hindari argumen variabel dalam metode
querydanstream_query.
- Gunakan
Menguji aplikasi secara lokal
Buat instance aplikasi di memori lokal menggunakan kode berikut:
agent = CLASS_NAME(
model=model, # Required.
tools=[get_exchange_rate], # Optional.
project=PROJECT_ID,
location=LOCATION,
)
agent.set_up()
Menguji metode query
Anda dapat menguji aplikasi dengan mengirimkan kueri pengujian ke instance lokal:
response = agent.query(
input="What is the exchange rate from US dollars to Swedish currency?"
)
Responsnya adalah kamus yang mirip dengan berikut ini:
{"input": "What is the exchange rate from US dollars to Swedish currency?",
# ...
"output": "For 1 US dollar you will get 10.7345 Swedish Krona."}
Menguji metode stream_query
Anda dapat menguji kueri streaming secara lokal dengan memanggil metode stream_query
dan melakukan iterasi pada hasilnya. Berikut contohnya:
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)
Kode ini mencetak setiap bagian respons saat dihasilkan. Outputnya mungkin terlihat seperti ini:
{'actions': [...], 'messages': [...]}
{'messages': [...], 'steps': [...]}
{'messages': [...],
'output': 'The exchange rate from US dollars to Swedish currency is 1 USD to '
'10.5751 SEK. \n'}
Dalam contoh ini, setiap bagian berisi informasi yang berbeda tentang respons, seperti tindakan yang dilakukan oleh agen, pesan yang dipertukarkan, dan output akhir.
Streaming API
Berikut beberapa hal penting yang perlu diingat saat menggunakan streaming API:
- Waktu tunggu maksimum: Waktu tunggu maksimum untuk respons streaming adalah 10 menit. Jika aplikasi Anda memerlukan waktu pemrosesan yang lebih lama, pertimbangkan untuk membagi tugas menjadi beberapa bagian yang lebih kecil.
- Streaming model dan rantai: Antarmuka Runnable LangChain mendukung streaming, sehingga Anda dapat melakukan streaming respons tidak hanya dari agen, tetapi juga model dan rantai.
- Kompatibilitas LangChain: Perhatikan bahwa metode
astream_eventLangChain tidak didukung. - Batasi pembuatan konten: Jika Anda mengalami masalah backpressure (saat produser menghasilkan data lebih cepat daripada yang dapat diproses konsumen), batasi kecepatan pembuatan konten. Hal ini dapat membantu mencegah buffer overflow dan memastikan pengalaman streaming yang lancar.
Menyesuaikan nama metode
Secara default, metode query dan stream_query didaftarkan sebagai operasi
dalam aplikasi yang di-deploy.
Anda dapat mengganti perilaku default dan menentukan kumpulan operasi yang akan
didaftarkan menggunakan metode register_operations.
Operasi dapat didaftarkan sebagai mode pemanggilan standar (diwakili oleh string kosong
"") atau streaming ("stream").
Dalam kode contoh berikut, metode register_operations akan menghasilkan
aplikasi
yang di-deploy yang menawarkan custom_method_1 dan custom_method_2 sebagai operasi untuk
panggilan standar, dan custom_stream_method_1 dan custom_stream_method_2 sebagai
operasi untuk panggilan streaming.
Operasi ini menggantikan operasi query dan stream_query default.
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",
],
}
Anda dapat menguji aplikasi dengan mengirimkan kueri pengujian ke instance seperti berikut:
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)
Anda tidak perlu mendaftarkan metode untuk kedua jenis pemanggilan. Misalnya, untuk hanya mendukung panggilan standar, Anda dapat melakukan hal berikut:
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",
],
}
Dalam contoh ini, hanya custom_method_1 dan custom_method_2 yang diekspos sebagai
operasi dalam aplikasi yang di-deploy.