はじめに
生成AIを利用できる環境が着々と揃っていっている中で、簡単なRAG環境を高速に構築したい！
と思ったことは多々あるのではないでしょうか。各サービスベンダーさんが機能拡張を進めているので
少し待てば、LLMアプリケーション開発（ほとんどコーディングは不要で、Figma的なノリで組める）も提供されるのではと思っています（既にあるのかも）
そのような中で一つポイントになるのはベクトルDB側の管理が挙げられると思い
今回はGoogle BigQueryにベクトルを保存する構成で、RAG環境の構築を検討してみようと思います。
今回の簡易アーキテクチャイメージ
pdfをアップロードし、その内容に対して質問をすれば回答してもらえる構成を検討しました。
① chainlit上でOpenAIを用いたチャットのやり取りを整え、GCSへアップロードする分岐も用意する
② ローカル上で「PDF=>BQ上でベクトル化」のスクリプトを作成し、Cloud Functions用に換装
③ 各自処理を繋げて稼働確認する
chat処理部分
基本ローカルでチャット画面を立ち上げる前提であれば、下記のスクリプト1本で動きます（便利）
補足は過剰書きで記載します。
Google Cloud
- 認証にあたりVertex側のAPIを解放する必要
- GCSへの操作にあたりサービスアカウントの設定（環境変数やOatuhなど）が必要、そのあたりの設定は他の記事に譲ります
- GCS側にアップロードする際にメタデータの付与が可能なので、チャット上で入力した文言をそのままメタデータとして扱うように設定
- VertexAIEmbeddingsのモデルは
multilingualを使用、日本語での埋め込み検索だと利用できるのは今これのみ？
LangChain & Chainlit & OpenAI
- LangChainで他ベクトルDBと同じようにVertex（google）とのやり取りが簡易に書ける
- Langchainはライブラリ自体に日々ガリガリ更新が入っているので、他とのversion互換性は確認の上で対応
- ChainlitはStreamlitとほぼ同じような実装感で、チャット側の処理が書ける。fileアップロードも可能なので、mime形式でpdf判定
- pdfアップロード時の入力contentは取得されないようなので、アップロードされるPDFは
規約と仮定して設定 - gptはお求めやすい
3.5を設定
import chainlit as cl
from google.cloud import storage
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.prompts import PromptTemplate
from langchain.schema import HumanMessage
from langchain_google_vertexai import VertexAIEmbeddings
from langchain.vectorstores.utils import DistanceStrategy
from langchain_community.vectorstores import BigQueryVectorSearch
#Google周りの設定
PROJECT_ID = "" #プロジェクトID
#BigQuery
REGION = "" #リージョン
DATASET = "" #保存先のデータセット
TABLE = "" #保存先のテーブル
#Storage
BUCKETS_NAME = "" #アップロード先のバケット名
embedding = VertexAIEmbeddings(
model_name="textembedding-gecko-multilingual@latest", project=PROJECT_ID
)
store = BigQueryVectorSearch(
project_id=PROJECT_ID,
dataset_name=DATASET,
table_name=TABLE,
location=REGION,
metadata_field="metadata",
embedding=embedding,
distance_strategy=DistanceStrategy.EUCLIDEAN_DISTANCE,
)
# OpenAI モデル選択
chat = ChatOpenAI(
model="gpt-3.5-turbo"
)
prompt = PromptTemplate(
template = """文章を元に質問に答えてください。
文章:
{document}
質問:{query}
""",
input_variables=["document","query"]
)
@cl.on_message
async def on_message(msg: cl.Message):
if not msg.elements:
documents_string = ""
query_vector = embedding.embed_query(msg.content)
documents = store.similarity_search_by_vector(query_vector, k=2)
for document in documents:
documents_string += f"""
==========================
{document.page_content}
"""
result = chat([
HumanMessage(content = prompt.format(document=documents_string,query=msg.content))
])
await cl.Message(content=result.content).send()
return
else:
for file in msg.elements:
if file.mime in "application/pdf":
# Google Cloud Storageへのアップロード
BLOB_NAME = file.name #GCSでの命名
SOURCE_FILE_NAME = file.path #GCSにアップロードしたいファイルのパス
client = storage.Client(project=PROJECT_ID)
bucket = client.get_bucket(BUCKETS_NAME)
blob = bucket.blob(BLOB_NAME)
#fileのアップロード
blob.upload_from_filename(SOURCE_FILE_NAME)
# メタデータの追記
metageneration_match_precondition = blob.metageneration
metadata = {'category': '規約'}
blob.metadata = metadata
blob.patch(if_metageneration_match=metageneration_match_precondition)
await cl.Message(content=f"データをGCS上にアップロードし、VecDBにデータを加えています").send()
return
else:
await cl.Message(content=f"アップロードできるのはPDFのみです").send()
return
上記で構築した、rag_bq_chat.py をchainlitで実行すると localhost で立ち上げが可能で
チャット上からPDFをアップロードすると、下記のような状況まで確認することができます。
$ chainlit run rag_bq_chat.py
Functions側の処理
- ローカル上にて簡単な実装と稼働確認の上で換装
- 非常に完結に書けるので、サクッと実装が可能（50行程度）
- Functions側は第一世代のPython3.12を利用、メモリは512MB
- PDF => Documentの部分は 一般的な
PyMuPDFLoaderを使用 - ベクトル化を行うにあたり、PDFから読み取れたDocumentを一つの塊(chunk)として分割し、BQ側の1レコード単位でinsertする形
- 初回は
Spacyを使っていたが日本語モデルをダウンロードしてFunctionsに圧縮して乗せてLOADさせるのが面倒だったので、tiktokenに変更
- 初回は
Cloud Functions上で設定をポチポチし、下記のように requirements.txt と main.py を設定
langchain-google-vertexai == 0.0.3
langchain == 0.1.10
langchain-community == 0.0.28
langchain-core == 0.1.32
langchain-text-splitters == 0.0.1
PyMuPDF == 1.23.26
sympy == 1.12
tiktoken == 0.6.0
from langchain_google_vertexai import VertexAIEmbeddings
from langchain.vectorstores.utils import DistanceStrategy
from langchain_community.vectorstores import BigQueryVectorSearch
from langchain.document_loaders import PyMuPDFLoader
from langchain_text_splitters import CharacterTextSplitter
from sympy import content
from google.cloud import storage
def trigger_gcs(event, context):
#Google周りの設定
PROJECT_ID = "" #プロジェクトID
#BigQuery
REGION = "" #リージョン
DATASET = "" #保存先のデータセット
TABLE = "" #保存先のテーブル
# bucket情報の読み取り
bucket_name = event["bucket"]
file_name = event["name"]
# サービスアカウントはFunctions側から指定し直接読み込み可
# client = storage.Client.from_service_account_json("./credential.json")
client = storage.Client(project=PROJECT_ID)
bucket = client.get_bucket(bucket_name)
blob = bucket.blob(file_name)
dl_file_path = "/tmp/" + file_name
blob.download_to_filename(dl_file_path)
embedding = VertexAIEmbeddings(
model_name="textembedding-gecko-multilingual@latest", project=PROJECT_ID
)
store = BigQueryVectorSearch(
project_id=PROJECT_ID,
dataset_name=DATASET,
table_name=TABLE,
location=REGION,
metadata_field="metadata",
embedding=embedding,
distance_strategy=DistanceStrategy.EUCLIDEAN_DISTANCE,
)
loader = PyMuPDFLoader(dl_file_path)
documents = loader.load()
text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
chunk_size=300, chunk_overlap=0
)
splitted_documents = text_splitter.split_documents(documents)
# データはカラム毎に整理の上でBQに格納
metadatas = [t.metadata for t in splitted_documents]
content = [t.page_content for t in splitted_documents]
store.add_texts(content, metadatas=metadatas)
print("データベースの作成が完了しました。")
先ほどのchatでPDFをアップロードの上で、処理が発火しBQ側にデータが格納されれば完了です。
類似度検索と回答生成
こちらは先の rag_bq_chat.py 側に処理が含まれています。Functionsによって追加されたベクトル情報(BQ)に対して
類似度検索を行い、上位2件のDocを参照Documentとして、gpt3.5に回答を生成してもらっています。
最終的な動作イメージ
さいごに
今回はGoogle Cloudを用いたRAGベースのチャット環境を簡易的に構築してみました。
弊社ではLLM側のご相談や構築も開始しておりますので、お気軽にお問合せください！