了解如何在 Cloud Run 上构建和部署 LangChain 应用

1. 概览

在此代码实验中，您将学习如何部署一个使用 Gemini 的 LangChain 应用，以便您就 Cloud Run 版本说明提出问题。

以下是该应用的工作方式示例：如果您提出“我可以在 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 Workstations，而不是使用本地机器。请注意，截至 2024 年 4 月，它运行的 Python 版本低于 3.11，因此您可能需要先升级 Python，然后才能开始使用。

启用 Cloud API

首先，运行以下命令，确保您已配置正确的 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 在全球多个位置提供服务，您需要选择一个位置来部署本实验中要使用的资源。在 shell 中将区域设置为环境变量（后续命令会使用此变量）：

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 会将 FastAPI 封装在 LangChain 链周围。它附带一个内置的 Playground，可让您轻松发送提示并检查结果（包括所有中间步骤）。建议您在编辑器中打开 run-rag 文件夹，然后探索其中的内容。

5. 创建索引作业

在开始组建 Web 应用之前，请确保 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。

第二条命令用于部署 Cloud Run 作业。这些标志的作用如下：

--source .：指定作业的源代码位于当前工作目录（您运行命令的目录）中。
--command python：设置要在容器内执行的命令。在本例中，它是运行 Python。
--args app/indexer.py：为 python 命令提供实参。这会指示它在应用目录中运行脚本 indexer.py。
--set-env-vars：设置 Python 脚本在执行期间可以访问的环境变量。
--region=$REGION：指定作业应部署到的区域。
--execute-now：告知 Cloud Run 在部署作业后立即启动该作业。

如需验证作业是否已成功完成，您可以执行以下操作：

通过 Web 控制台读取作业执行的日志。它应报告“已完成保存：xxx 篇版本说明”（其中 xxx 是已保存的版本说明的数量）。
您还可以前往 Web 控制台中的 Cloud SQL 实例，并使用 Cloud SQL Studio 查询 langchain_pg_embedding 表中的记录数。

6. 编写 Web 应用

在编辑器中打开 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. 将 Web 应用部署到 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。
使用容器映像创建 Cloud Run 服务。

命令运行完毕后，系统会列出 run.app 网域上的 HTTPS 网址。这是新 Cloud Run 服务的公开网址

8. 探索游乐场

打开 Cloud Run 服务网址，然后前往 /playground。系统会显示一个文本字段。您可以使用它来询问有关 Cloud Run 版本说明的问题，例如：

9. 恭喜

您已在 Cloud Run 上成功构建并部署了 LangChain 应用。太棒了！

以下是关键概念：

使用 LangChain 框架构建检索增强生成 (RAG) 应用。
使用 Cloud SQL 上的 PostgreSQL 作为向量数据库，并使用默认安装在 Cloud SQL 上的 pgvector。
将长时间运行的索引作业作为 Cloud Run 作业运行，并将 Web 应用作为 Cloud Run 服务运行。
使用 LangServe 将 LangChain 链封装在 FastAPI 应用中，从而提供一个方便的界面来与 RAG 应用互动。

清理

为避免因本教程中使用的资源导致您的 Google Cloud Platform 账号产生费用，请执行以下操作：

在 Cloud Console 中，转到“管理资源”页面。
在项目列表中，选择您的项目，然后点击“删除”。
在对话框中输入项目 ID，然后点击“关停”以删除项目。

如果您想保留项目，请务必删除以下资源：

Cloud SQL 实例
Cloud Run 服务
Cloud Run 作业