このページは Cloud Translation API によって翻訳されました。

Cloud Run で LangChain アプリをビルドしてデプロイする方法を学習する

1. 概要

この Codelab では、Gemini を使用して Cloud Run リリースノートに関する質問を可能にする LangChain アプリをデプロイする方法を学びます。

アプリの動作の例を次に示します。たとえば、「Cloud Run で Cloud Storage バケットをボリュームとしてマウントできますか？」と質問すると、「2024 年 1 月 19 日以降は可能です」などの回答が返されます。

根拠のある回答を返すために、アプリはまず質問に類似する Cloud Run リリースノートを取得し、質問とリリースノートの両方を Gemini にプロンプトします。（これは一般に RAG と呼ばれるパターンです）。アプリのアーキテクチャを示す図を以下に示します。

2. 設定と要件

まず、開発環境が正しく設定されていることを確認しましょう。

アプリに必要なリソースをデプロイするには、Google Cloud プロジェクトが必要です。
アプリをデプロイするには、ローカルマシンに gcloud がインストールされ、認証され、プロジェクトを使用するように構成されている必要があります。
- gcloud auth login
- gcloud config set project
アプリケーションをローカルマシンで実行する場合は（推奨）、割り当てプロジェクトの設定など、アプリケーションのデフォルト認証情報が正しく設定されていることを確認する必要があります。
- gcloud auth application-default login
- gcloud auth application-default set-quota-project
また、次のソフトウェアがインストールされている必要があります。
- Python（バージョン 3.11 以降が必要）
- LangChain CLI
- poetry: 依存関係を管理
- pipx: 隔離された仮想環境に LangChain CLI と poetry をインストールして実行します。

このチュートリアルに必要なツールのインストールを開始する際に役立つブログ記事があります。

Cloud Workstations

ローカルマシンの代わりに、Google Cloud で Cloud ワークステーションを使用することもできます。2024 年 4 月時点では、Python バージョン 3.11 より前が実行されているため、開始する前に Python をアップグレードする必要があります。

Cloud APIs を有効にする

まず、次のコマンドを実行して、使用する適切な Google Cloud プロジェクトが構成されていることを確認します。

gcloud config list project

正しいプロジェクトが表示されない場合は、次のコマンドを使用してプロジェクトを設定できます。

gcloud config set project <PROJECT_ID>

次に、次の API を有効にします。

gcloud services enable \
  bigquery.googleapis.com \
  sqladmin.googleapis.com \
  aiplatform.googleapis.com \
  cloudresourcemanager.googleapis.com \
  artifactregistry.googleapis.com \
  cloudbuild.googleapis.com \
  run.googleapis.com \
  secretmanager.googleapis.com

地域を選択してください

Google Cloud は世界中のさまざまなロケーションで利用できるため、このラボで使用するリソースをデプロイするには、いずれかのロケーションを選択する必要があります。リージョンをシェルの環境変数として設定します（後続のコマンドでこの変数を使用します）。

export REGION=us-central1

3. ベクトルデータベースインスタンスを作成する

このアプリの重要な部分は、ユーザーの質問に関連するリリースノートを取得することです。具体的には、Cloud Storage に関する質問の場合は、プロンプトに次のリリースノートを追加します。

テキストエンベディングとベクトルデータベースを使用して、意味的に類似したリリースノートを検索できます。

Cloud SQL で PostgreSQL をベクトルデータベースとして使用する方法について説明します。新しい Cloud SQL インスタンスの作成には時間がかかるため、今すぐ作成しましょう。

gcloud sql instances create sql-instance \
  --database-version POSTGRES_14 \
  --tier db-f1-micro \
  --region $REGION

このコマンドを実行して、次の手順に進みます。いずれはデータベースを作成してユーザーを追加する必要がありますが、今はスピナーを眺めて時間を無駄にせずに、

PostgreSQL はリレーショナルデータベースサーバーです。Cloud SQL のすべての新しいインスタンスには、拡張機能 pgvector がデフォルトでインストールされているため、ベクトルデータベースとしても使用できます。

4. LangChain アプリをスキャフォールディングする

続行するには、LangChain CLI がインストールされていること、および依存関係を管理する poetry があることを確認してください。pipx を使用してインストールする方法は次のとおりです。

pipx install langchain-cli poetry

次のコマンドを使用して、LangChain アプリをスキャフォールドします。プロンプトが表示されたら、フォルダに run-rag という名前を付け、Enter キーを押してパッケージのインストールをスキップします。

langchain app new

run-rag ディレクトリに移動して依存関係をインストールする

poetry install

LangServe アプリが作成されました。LangServe は、LangChain チェーンに FastAPI をラップします。組み込みのプレイグラウンドにより、プロンプトを簡単に送信し、すべての中間ステップを含む結果を簡単に検査できます。エディタで run-rag フォルダを開いて、内容を確認することをおすすめします。

5. インデックス作成ジョブを作成する

ウェブアプリの作成を開始する前に、Cloud Run リリースノートに Cloud SQL データベースにインデックスが付けられていることを確認しましょう。このセクションでは、次の処理を行うインデックスジョブを作成します。

インデックス作成ジョブは、リリースノートを取得し、テキストエンベディングモデルを使用してベクトルに変換し、ベクトルデータベースに保存します。これにより、関連するリリースノートをセマンティックな意味に基づいて効率的に検索できるようになります。

run-rag/app フォルダに、次の内容の indexer.py ファイルを作成します。

import os
from google.cloud.sql.connector import Connector
import pg8000
from langchain_community.vectorstores.pgvector import PGVector
from langchain_google_vertexai import VertexAIEmbeddings
from google.cloud import bigquery


# Retrieve all Cloud Run release notes from BigQuery 
client = bigquery.Client()
query = """
SELECT
  CONCAT(FORMAT_DATE("%B %d, %Y", published_at), ": ", description) AS release_note
FROM `bigquery-public-data.google_cloud_release_notes.release_notes`
WHERE product_name= "Cloud Run"
ORDER BY published_at DESC
"""
rows = client.query(query)

print(f"Number of release notes retrieved: {rows.result().total_rows}")

# Set up a PGVector instance 
connector = Connector()

def getconn() -> pg8000.dbapi.Connection:
    conn: pg8000.dbapi.Connection = connector.connect(
        os.getenv("DB_INSTANCE_NAME", ""),
        "pg8000",
        user=os.getenv("DB_USER", ""),
        password=os.getenv("DB_PASS", ""),
        db=os.getenv("DB_NAME", ""),
    )
    return conn

store = PGVector(
    connection_string="postgresql+pg8000://",
    use_jsonb=True,
    engine_args=dict(
        creator=getconn,
    ),
    embedding_function=VertexAIEmbeddings(
        model_name="textembedding-gecko@003"
    ),
    pre_delete_collection=True  
)

# Save all release notes into the Cloud SQL database
texts = list(row["release_note"] for row in rows)
ids = store.add_texts(texts)

print(f"Done saving: {len(ids)} release notes")

必要な依存関係を追加します。

poetry add \
  "cloud-sql-python-connector[pg8000]" \
  langchain-google-vertexai==1.0.5 \
  langchain-community==0.2.5 \
  pgvector

データベースとユーザーを作成する

Cloud SQL インスタンス sql-instance にデータベース release-notes を作成します。

gcloud sql databases create release-notes --instance sql-instance

app というデータベースユーザーを作成します。

gcloud sql users create app --instance sql-instance --password "myprecious"

インデックスジョブをデプロイして実行する

ジョブをデプロイして実行します。

DB_INSTANCE_NAME=$(gcloud sql instances describe sql-instance --format="value(connectionName)")

gcloud run jobs deploy indexer \
  --source . \
  --command python \
  --args app/indexer.py \
  --set-env-vars=DB_INSTANCE_NAME=$DB_INSTANCE_NAME \
  --set-env-vars=DB_USER=app \
  --set-env-vars=DB_NAME=release-notes \
  --set-env-vars=DB_PASS=myprecious \
  --region=$REGION \
  --execute-now

長いコマンドですが、何が行われているか見てみましょう。

最初のコマンドは、接続名（project:region:instance という形式の一意の ID）を取得し、環境変数 DB_INSTANCE_NAME として設定します。

2 番目のコマンドは、Cloud Run ジョブをデプロイします。フラグの役割は次のとおりです。

--source .: ジョブのソースコードが現在の作業ディレクトリ（コマンドを実行しているディレクトリ）にあることを指定します。
--command python: コンテナ内で実行するコマンドを設定します。この場合は、Python を実行します。
--args app/indexer.py: python コマンドに引数を指定します。これにより、アプリディレクトリでスクリプト indexer.py を実行するように指示されます。
--set-env-vars: Python スクリプトが実行中にアクセスできる環境変数を設定します。
--region=$REGION: ジョブをデプロイするリージョンを指定します。
--execute-now: デプロイ後すぐにジョブを開始するよう Cloud Run に指示します。

ジョブが正常に完了したことを確認するには、次の操作を行います。

ウェブコンソールからジョブ実行のログを読み取ります。「Done の保存: xxx release notes」と表示されます（xxx は保存したリリースノートの数）。
ウェブコンソールで Cloud SQL インスタンスに移動し、Cloud SQL Studio を使用して langchain_pg_embedding テーブルのレコード数をクエリすることもできます。

6. ウェブアプリケーションを作成する

エディタで app/server.py ファイルを開きます。次のような行が表示されます。

# Edit this to add the chain you want to add

このコメントを次のスニペットに置き換えます。

# (1) Initialize VectorStore
connector = Connector()


def getconn() -> pg8000.dbapi.Connection:
    conn: pg8000.dbapi.Connection = connector.connect(
        os.getenv("DB_INSTANCE_NAME", ""),
        "pg8000",
        user=os.getenv("DB_USER", ""),
        password=os.getenv("DB_PASS", ""),
        db=os.getenv("DB_NAME", ""),
    )
    return conn


vectorstore = PGVector(
    connection_string="postgresql+pg8000://",
    use_jsonb=True,
    engine_args=dict(
        creator=getconn,
    ),
    embedding_function=VertexAIEmbeddings(
        model_name="textembedding-gecko@003"
    )
)

# (2) Build retriever


def concatenate_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)


notes_retriever = vectorstore.as_retriever() | concatenate_docs

# (3) Create prompt template
prompt_template = PromptTemplate.from_template(
    """You are a Cloud Run expert answering questions. 
Use the retrieved release notes to answer questions
Give a concise answer, and if you are unsure of the answer, just say so.

Release notes: {notes}

Here is your question: {query}
Your answer: """)

# (4) Initialize LLM
llm = VertexAI(
    model_name="gemini-1.0-pro-001",
    temperature=0.2,
    max_output_tokens=100,
    top_k=40,
    top_p=0.95
)

# (5) Chain everything together
chain = (
    RunnableParallel({
        "notes": notes_retriever,
        "query": RunnablePassthrough()
    })
    | prompt_template
    | llm
    | StrOutputParser()
)

また、次のインポートも追加する必要があります。

import pg8000
import os
from google.cloud.sql.connector import Connector
from langchain_google_vertexai import VertexAI
from langchain_google_vertexai import VertexAIEmbeddings
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.vectorstores.pgvector import PGVector

最後に、「NotImplemented」と記述されている行を次のように変更します。

# add_routes(app, NotImplemented)
add_routes(app, chain)

7. ウェブアプリケーションを Cloud Run にデプロイする

run-rag ディレクトリから、次のコマンドを使用してアプリを Cloud Run にデプロイします。

DB_INSTANCE_NAME=$(gcloud sql instances describe sql-instance --format="value(connectionName)")

gcloud run deploy run-rag \
  --source . \
  --set-env-vars=DB_INSTANCE_NAME=$DB_INSTANCE_NAME \
  --set-env-vars=DB_USER=app \
  --set-env-vars=DB_NAME=release-notes \
  --set-env-vars=DB_PASS=myprecious \
  --region=$REGION \
  --allow-unauthenticated

このコマンドは次の処理を行います。

ソースコードを Cloud Build にアップロードする
docker build を実行します。
生成されたコンテナイメージを Artifact Registry に push します。
コンテナイメージを使用して Cloud Run サービスを作成します。

コマンドが完了すると、run.app ドメインの HTTPS URL が一覧表示されます。これは、新しい Cloud Run サービスの公開 URL です。

8. 遊び場を探索する

Cloud Run サービスの URL を開き、/playground に移動します。テキストフィールドが表示されます。次のように、Cloud Run リリースノートに関する質問に使用できます。

9. 完了

Cloud Run に LangChain アプリを正常にビルドしてデプロイしました。お疲れさまでした。

主なコンセプトは次のとおりです。

LangChain フレームワークを使用して検索拡張生成（RAG）アプリケーションを構築する。
Cloud SQL にデフォルトでインストールされている pgvector を使用して、Cloud SQL 上の PostgreSQL をベクトルデータベースとして使用します。
実行時間の長いインデックスジョブを Cloud Run ジョブとして実行し、ウェブアプリケーションを Cloud Run サービスとして実行する。
LangServe を使用して LangChain チェーンを FastAPI アプリケーションにラップし、RAG アプリを操作するための便利なインターフェースを提供します。

クリーンアップ

このチュートリアルで使用したリソースについて、Google Cloud Platform アカウントに課金されないようにする手順は次のとおりです。

Cloud Console で [リソースの管理] ページに移動します。
プロジェクトリストでプロジェクトを選択し、[削除] をクリックします。
ダイアログでプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。

プロジェクトを保持する場合は、次のリソースを削除してください。

Cloud SQL インスタンス
Cloud Run サービス
Cloud Run ジョブ