Codelab - 使用 Firestore、Vector Search、Langchain 和 Gemini 构建上下文感知型瑜伽姿势推荐应用（Python 版）

1. 简介

在此 Codelab 中，您将构建一个应用，该应用使用矢量搜索来推荐瑜伽姿势。

在本 Codelab 中，您将采用分步方法，具体步骤如下：

1. 利用现有的 Hugging Face 数据集（JSON 格式）中的瑜伽姿势。
2. 使用额外的字段说明来增强数据集，该说明使用 Gemini 为每个姿势生成说明。
3. 使用 Langchain 创建文档，使用 Firestore Langchain 集成在 Firestore 中创建集合和嵌入。
4. 在 Firestore 中创建复合索引，以便进行向量搜索。
5. 在 Flask 应用中使用矢量搜索，将所有内容整合到一起，如下所示：

实践内容

• 设计、构建和部署一个 Web 应用，该应用使用 Vector Search 推荐瑜伽姿势。

学习内容

• 如何使用 Gemini 生成文本内容，并在此 Codelab 的上下文中生成瑜伽姿势的说明
• 如何使用 适用于 Firestore 的 Langchain 文档加载器将 Hugging Face 增强型数据集中的记录以及向量嵌入加载到 Firestore 中
• 如何使用 适用于 Firestore 的 Langchain 向量存储区根据自然语言查询搜索数据
• 如何使用 Google Cloud Text to Speech API 生成音频内容

所需条件

• Chrome 网络浏览器
• Gmail 账号
• 启用了结算功能的 Cloud 项目

此 Codelab 面向各种级别（包括新手）的开发者，其示例应用中使用了 Python。不过，您无需了解 Python 即可理解所介绍的概念。

2. 准备工作

创建项目

1. 在 Google Cloud Console 的项目选择器页面上，选择或创建一个 Google Cloud 项目。
2. 确保您的 Cloud 项目已启用结算功能。了解如何检查项目是否已启用结算功能。
3. 您将使用 Cloud Shell，这是一个在 Google Cloud 中运行的命令行环境，它预加载了 bq。点击 Google Cloud 控制台顶部的“激活 Cloud Shell”。

4. 连接到 Cloud Shell 后，您可以使用以下命令检查自己是否已通过身份验证，以及项目是否已设置为您的项目 ID：

gcloud auth list

5. 在 Cloud Shell 中运行以下命令，以确认 gcloud 命令了解您的项目。

gcloud config list project

6. 如果项目未设置，请使用以下命令进行设置：

gcloud config set project <YOUR_PROJECT_ID>

7. 通过以下命令启用所需的 API。此过程可能需要几分钟的时间，请耐心等待。

gcloud services enable firestore.googleapis.com \
                       compute.googleapis.com \
                       cloudresourcemanager.googleapis.com \
                       servicenetworking.googleapis.com \
                       run.googleapis.com \
                       cloudbuild.googleapis.com \
                       cloudfunctions.googleapis.com \
                       aiplatform.googleapis.com \
                       texttospeech.googleapis.com

成功执行该命令后，您应该会看到如下所示的消息：

Operation "operations/..." finished successfully.

您可以通过控制台搜索各个产品或使用此链接，以替代 gcloud 命令。

如果缺少任何 API，您随时可以在实现过程中启用它。

如需了解 gcloud 命令和用法，请参阅文档。

克隆代码库并设置环境设置

下一步是克隆我们将在本 Codelab 的其余部分中引用的示例代码库。假设您在 Cloud Shell 中，请在主目录中输入以下命令：

git clone https://github.com/rominirani/yoga-poses-recommender-python

如需启动编辑器，请点击 Cloud Shell 窗口工具栏上的“打开编辑器”。点击左上角的菜单栏，然后依次选择“File”（文件）→“Open Folder”（打开文件夹），如下所示：

选择 yoga-poses-recommender-python 文件夹，您应该会看到该文件夹打开，其中包含以下文件，如下所示：

现在，我们需要设置要使用的环境变量。点击 config.template.yaml 文件，您应该会看到如下所示的内容：

project_id: your-project-id
location: us-central1
gemini_model_name: gemini-1.5-flash-002
embedding_model_name: text-embedding-004
image_generation_model_name: imagen-3.0-fast-generate-002
database: (default)
collection: poses
test_collection: test-poses
top_k: "3"

请根据您在创建 Google Cloud 项目和 Firestore 数据库区域时选择的值，更新 project_id 和 location 的值。理想情况下，我们希望 Google Cloud 项目和 Firestore 数据库的 location 值相同，例如 us-central1。

在本 Codelab 中，我们将使用预配置的值（当然，project_id 和 location 除外，您需要根据自己的配置进行设置）。

请将此文件另存为 config.yaml，并保存在 config.template.yaml 文件所在的文件夹中。

现在，最后一步是创建一个 Python 环境，我们将在本地使用该环境，并使用为我们设置的所有 Python 依赖项。查看包含其详细信息的 pyproject.toml 文件，其内容如下所示：

dependencies = [
    "datasets>=3.2.0",
    "flask>=3.1.0",
    "google-cloud-aiplatform>=1.78.0",
    "google-cloud-texttospeech>=2.24.0",
    "langchain-community>=0.3.15",
    "langchain-core>=0.3.31",
    "langchain-google-community>=2.0.4",
    "langchain-google-firestore>=0.5.0",
    "langchain-google-vertexai>=2.0.7",
    "pydantic-settings>=2.7.1",
    "pyyaml>=6.0.2",
    "tenacity>=9.0.0",
]

这些依赖项已在 requirements.txt 中锁定了版本.。总而言之，我们需要创建一个虚拟 Python 环境，并在其中安装 requirements.txt 中的 Python 软件包依赖项。为此，请前往 Cloud Shell IDE 中的 Command Palette（Ctrl+Shift+P），然后输入 Python: Create Environment。按照接下来的几个步骤选择 Virtual Environment(venv)、Python 3.x interpreter 和 requirements.txt 文件。

创建环境后，我们需要使用以下命令激活所创建的环境

source .venv/bin/activate

您应该会在控制台中看到 (.venv)。例如 -> (.venv) yourusername@cloudshell:

太棒了！现在，我们已经准备好继续设置 Firestore 数据库了。

3. 设置 Firestore

Cloud Firestore 是一个全代管式无服务器文档数据库，我们将其用作应用数据的后端。Cloud Firestore 中的数据以文档的集合形式进行结构化。

Firestore 数据库初始化

访问 Cloud 控制台中的 Firestore 页面。

如果您之前未在项目中初始化 Firestore 数据库，请点击 Create Database 创建 default 数据库。在创建数据库时，请使用以下值：

• Firestore 模式：Native.
• 位置信息：使用默认位置信息设置。
• 对于安全规则，请使用 Test rules。
• 创建数据库。

在下一部分中，我们将为在默认 Firestore 数据库中创建一个名为 poses 的集合奠定基础。此集合将存储示例数据（文档）或瑜伽姿势信息，以便我们在应用中使用。

请注意，如果您使用的是 Firebase，那么这与您熟悉的 Firestore 是同一个 Firestore。事实上，您可以使用 Firebase 控制台访问同一数据库实例！

至此，Firestore 数据库设置部分已完成。

4. 准备瑜伽姿势数据集

我们的第一项任务是准备要用于应用的瑜伽姿势数据集。我们将从现有的 Hugging Face 数据集开始，然后使用其他信息对其进行增强。

查看 Hugging Face 瑜伽姿势数据集。请注意，虽然此 Codelab 使用了其中一个数据集，但实际上您可以使用任何其他数据集，并按照所演示的相同方法来增强数据集。

如果我们前往 Files and versions 部分，则可以获取所有姿势的 JSON 数据文件。

我们已下载 yoga_poses.json 并将该文件提供给您。此文件名为 yoga_poses_alldata.json，位于 /data 文件夹中。

我们将演示在少量记录的较小子集上生成说明并最终生成嵌入的整个过程，以便您熟悉该过程。

前往 Cloud Shell 编辑器中的 data/yoga_poses.json 文件，查看 JSON 对象列表，其中每个 JSON 对象都代表一个瑜伽姿势。我们总共有 3 条记录，示例记录如下所示：

{
   "name": "Big Toe Pose",
   "sanskrit_name": "Padangusthasana",
   "photo_url": "https://pocketyoga.com/assets/images/full/ForwardBendBigToe.png",
   "expertise_level": "Beginner",
   "pose_type": ["Standing", "Forward Bend"]
 }

现在，我们有机会介绍一下 Gemini，以及如何使用默认模型本身为其生成 description 字段。

在 Cloud Shell Editor 中，前往 generate-descriptions.py 文件。此文件的内容如下所示：

import json
import time
import logging
import vertexai
from langchain_google_vertexai import VertexAI
from tenacity import retry, stop_after_attempt, wait_exponential
from settings import get_settings

settings = get_settings()
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
# Initialize Vertex AI SDK
vertexai.init(project=settings.project_id, location=settings.location)
logging.info("Done Initializing Vertex AI SDK")


@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=4, max=10),
)
def generate_description(pose_name, sanskrit_name, expertise_level, pose_types):
    """Generates a description for a yoga pose using the Gemini API."""

    prompt = f"""
    Generate a concise description (max 50 words) for the yoga pose: {pose_name}
    Also known as: {sanskrit_name}
    Expertise Level: {expertise_level}
    Pose Type: {", ".join(pose_types)}

    Include key benefits and any important alignment cues.
    """
    try:
        model = VertexAI(model_name=settings.gemini_model_name, verbose=True)
        response = model.invoke(prompt)
        return response
    except Exception as e:
        logging.info(f"Error generating description for {pose_name}: {e}")
        return ""


def add_descriptions_to_json(input_file, output_file):
    """Loads JSON data, adds descriptions, and saves the updated data."""

    with open(input_file, "r") as f:
        yoga_poses = json.load(f)

    total_poses = len(yoga_poses)
    processed_count = 0

    for pose in yoga_poses:
        if pose["name"] != " Pose":
            start_time = time.time()  # Record start time
            pose["description"] = generate_description(
                pose["name"],
                pose["sanskrit_name"],
                pose["expertise_level"],
                pose["pose_type"],
            )
            end_time = time.time()  # Record end time

            processed_count += 1
            end_time = time.time()  # Record end time
            time_taken = end_time - start_time
            logging.info(
                f"Processed: {processed_count}/{total_poses} - {pose['name']} ({time_taken:.2f} seconds)"
            )

        else:
            pose["description"] = ""
            processed_count += 1
            logging.info(
                f"Processed: {processed_count}/{total_poses} - {pose['name']} ({time_taken:.2f} seconds)"
            )
        # Adding a delay to avoid rate limit
        time.sleep(30)

    with open(output_file, "w") as f:
        json.dump(yoga_poses, f, indent=2)


def main():
    # File paths
    input_file = "./data/yoga_poses.json"
    output_file = "./data/yoga_poses_with_descriptions.json"

    # Add descriptions and save the updated JSON
    add_descriptions_to_json(input_file, output_file)


if __name__ == "__main__":
    main()

此应用将向每个瑜伽姿势 JSON 记录添加一个新的 description 字段。它将通过调用 Gemini 模型来获取说明，我们会在该模型中为其提供必要的提示。该字段会添加到 JSON 文件中，并且新文件会写入 data/yoga_poses_with_descriptions.json 文件。

我们来了解一下主要步骤：

1. 在 main() 函数中，您会发现它会调用 add_descriptions_to_json 函数，并提供预期的输入文件和输出文件。
2. add_descriptions_to_json 函数会对每个 JSON 记录（即瑜伽帖子信息）执行以下操作：
3. 它会提取 pose_name、sanskrit_name、expertise_level 和 pose_types。
4. 它会调用用于构建提示的 generate_description 函数，然后调用 Langchain VertexAI 模型类来获取回答文本。
5. 然后，将此响应文本添加到 JSON 对象。
6. 然后，系统会将更新后的对象 JSON 列表写入目标文件。

我们来运行此应用。启动一个新的终端窗口（Ctrl+Shift+C），然后输入以下命令：

python generate-descriptions.py

如果系统要求您提供任何授权，请直接提供。

您会发现应用开始执行。我们在记录之间添加了 30 秒的延迟时间，以避免新 Google Cloud 账号可能存在的速率限制配额，因此请耐心等待。

正在运行的示例如下所示：

使用 Gemini 调用增强完所有 3 条记录后，系统会生成一个文件 data/yoga_poses_with_description.json。您可以查看一下。

现在，我们已经准备好数据文件，下一步是了解如何使用该文件填充 Firestore 数据库以及如何生成嵌入。

5. 将数据导入 Firestore 并生成向量嵌入

我们已经有了 data/yoga_poses_with_description.json 文件，现在需要使用该文件填充 Firestore 数据库，并且重要的是，为每个记录生成矢量嵌入。在后续使用自然语言提供的用户查询对这些向量嵌入进行相似度搜索时，这些向量嵌入会很有用。

我们将使用 Langchain Firestore 组件来实现上述流程。

具体步骤如下：

1. 我们将 JSON 对象列表转换为 Langchain Document 对象列表。每个文档都将具有两个属性：page_content 和 metadata。元数据对象将包含具有 name、description、sanskrit_name 等属性的整个 JSON 对象。page_content 将是一个字符串文本，将是几个字段的串联。
2. 获得 Document 对象列表后，我们将使用 FirestoreVectorStore Langchain 类（尤其是 from_documents 方法）与此文档列表、集合名称（我们使用指向 test-poses 的 TEST_COLLECTION 变量）、Vertex AI 嵌入类和 Firestore 连接详细信息（PROJECT_ID 和 DATABASE 名称）搭配使用。这将创建集合，并为每个属性生成 embedding 字段。

import-data.py 的代码如下所示（为简洁起见，部分代码已截断）：

... 

def create_langchain_documents(poses):
   """Creates a list of Langchain Documents from a list of poses."""
   documents = []
   for pose in poses:
       # Convert the pose to a string representation for page_content
       page_content = (
           f"name: {pose.get('name', '')}\n"
           f"description: {pose.get('description', '')}\n"
           f"sanskrit_name: {pose.get('sanskrit_name', '')}\n"
           f"expertise_level: {pose.get('expertise_level', 'N/A')}\n"
           f"pose_type: {pose.get('pose_type', 'N/A')}\n"
       ).strip()
       # The metadata will be the whole pose
       metadata = pose

       document = Document(page_content=page_content, metadata=metadata)
       documents.append(document)
   logging.info(f"Created {len(documents)} Langchain documents.")
   return documents

def main():
    all_poses = load_yoga_poses_data_from_local_file(
        "./data/yoga_poses_with_descriptions.json"
    )
    documents = create_langchain_documents(all_poses)
    logging.info(
        f"Successfully created langchain documents. Total documents: {len(documents)}"
    )

    embedding = VertexAIEmbeddings(
        model_name=settings.embedding_model_name,
        project=settings.project_id,
        location=settings.location,
    )

    client = firestore.Client(project=settings.project_id, database=settings.database)

    vector_store = FirestoreVectorStore.from_documents(
        client=client,
        collection=settings.test_collection,
        documents=documents,
        embedding=embedding,
    )
    logging.info("Added documents to the vector store.")


if __name__ == "__main__":
    main()

我们来运行此应用。启动一个新的终端窗口（Ctrl+Shift+C），然后输入以下命令：

python import-data.py

如果一切顺利，您应该会看到类似于以下内容的消息：

2025-01-21 14:50:06,479 - INFO - Added documents to the vector store.

如需检查记录是否已成功插入且是否已生成嵌入，请访问 Cloud 控制台中的 Firestore 页面。

点击“（默认）”数据库，系统应会显示 test-poses 集合以及该集合下的多个文档。每个文档都是一个瑜伽姿势。

点击任意文档即可查看相关字段。除了我们导入的字段之外，您还会看到 embedding 字段，这是我们使用 Langchain VertexAIEmbeddings 类自动为您生成的矢量字段，其中包含 text-embedding-004 Vertex AI 嵌入模型。

现在，我们已将包含嵌入的记录上传到 Firestore 数据库，接下来可以继续下一步，了解如何在 Firestore 中执行向量相似度搜索。

6. 将完整的瑜伽姿势导入 Firestore 数据库集合

现在，我们将创建 poses 集合，其中包含 160 个瑜伽姿势的完整列表。我们已为此生成了一个数据库导入文件，您可以直接导入该文件。这样做是为了节省实验时间。生成包含说明和嵌入的这个数据库的过程与我们在上一部分中看到的相同。

请按以下步骤导入数据库：

1. 使用下面的 gsutil 命令在项目中创建存储分区。将以下命令中的 <PROJECT_ID> 变量替换为您的 Google Cloud 项目 ID。

gsutil mb -l us-central1 gs://<PROJECT_ID>-my-bucket

2. 现在，我们已创建存储分区，接下来需要将准备好的数据库导出内容复制到此存储分区中，然后才能将其导入 Firebase 数据库。使用以下命令：

gsutil cp -r gs://yoga-database-firestore-export-bucket/2025-01-27T05:11:02_62615  gs://<PROJECT_ID>-my-bucket

现在，我们已经有了要导入的数据，接下来可以执行最后一步，将数据导入我们创建的 Firebase 数据库 (default)。

3. 使用以下 gcloud 命令：

gcloud firestore import gs://<PROJECT_ID>-my-bucket/2025-01-27T05:11:02_62615

导入过程需要几秒钟的时间，完成后，您可以访问 https://console.cloud.google.com/firestore/databases，选择 default 数据库和 poses 集合，以验证 Firestore 数据库和集合，如下所示：

至此，我们已完成将在应用中使用的 Firestore 集合的创建。

7. 在 Firestore 中执行向量相似度搜索

为了执行矢量相似度搜索，我们将接受来自用户的查询。此查询的示例可以是 "Suggest me some exercises to relieve back pain"。

查看 search-data.py 文件。需要重点关注的关键函数是搜索函数，如下所示。概括来讲，它会创建一个嵌入类，用于为用户查询生成嵌入。然后，它使用 FirestoreVectorStore 类调用其 similarity_search 函数。

def search(query: str):
    """Executes Firestore Vector Similarity Search"""
    embedding = VertexAIEmbeddings(
        model_name=settings.embedding_model_name,
        project=settings.project_id,
        location=settings.location,
    )

    client = firestore.Client(project=settings.project_id, database=settings.database)

    vector_store = FirestoreVectorStore(
        client=client, collection=settings.collection, embedding_service=embedding
    )

    logging.info(f"Now executing query: {query}")
    results: list[Document] = vector_store.similarity_search(
        query=query, k=int(settings.top_k), include_metadata=True
    )
    for result in results:
        print(result.page_content)

在使用一些查询示例运行此脚本之前，您必须先生成 Firestore 复合索引，这是成功执行搜索查询所必需的。如果您在未创建索引的情况下运行应用，系统会显示一条错误消息，指明您需要先创建索引，并显示用于先创建索引的命令。

用于创建复合索引的 gcloud 命令如下所示：

gcloud firestore indexes composite create --project=<YOUR_PROJECT_ID> --collection-group=poses --query-scope=COLLECTION --field-config=vector-config='{"dimension":"768","flat": "{}"}',field-path=embedding

由于数据库中有 150 多条记录，因此索引需要几分钟才能完成。完成后，您可以通过以下命令查看索引：

gcloud firestore indexes composite list

您应该会在列表中看到刚刚创建的索引。

现在，请尝试以下命令：

python search-data.py --prompt "Recommend me some exercises for back pain relief"

系统应该会向您提供一些建议。示例运行如下所示：

2025-01-21 15:48:51,282 - INFO - Now executing query: Recommend me some exercises for back pain relief
name: Supine Spinal Twist Pose
description: A gentle supine twist (Supta Matsyendrasana), great for beginners.  Releases spinal tension, improves digestion, and calms the nervous system.  Keep shoulders flat on the floor and lengthen the spine.

sanskrit_name: Supta Matsyendrasana
expertise_level: Beginner
pose_type: ['Supine', 'Twist']
name: Cow Pose
description: Cow Pose (Bitilasana) is a gentle backbend, stretching the chest, shoulders, and abdomen.  Maintain a neutral spine, lengthen the tailbone, and avoid hyperextension.  Benefits include improved posture and stress relief.

sanskrit_name: Bitilasana
expertise_level: Beginner
pose_type: ['Arm Leg Support', 'Back Bend']
name: Locust I Pose
description: Locust Pose I (Shalabhasana A) strengthens the back, glutes, and shoulders.  Lie prone, lift chest and legs simultaneously, engaging back muscles.  Keep hips grounded and gaze slightly forward.

sanskrit_name: Shalabhasana A
expertise_level: Intermediate
pose_type: ['Prone', 'Back Bend']

完成以上操作后，我们现在已经了解了如何使用 Firestore 向量数据库上传记录、生成嵌入和执行向量相似搜索。现在，我们可以创建一个 Web 应用，将向量搜索集成到 Web 前端。

8. Web 应用

Python Flask Web 应用位于 main.py 文件中，前端 HTML 文件位于 templates/index.html.

建议您查看这两个文件。首先，从包含 /search 处理程序的 main.py 文件开始，该处理程序会接受从前端 HTML index.html 文件传递的提示。然后，系统会调用搜索方法，该方法会执行我们在上一部分中介绍的向量相似度搜索。

然后，系统会将响应连同建议列表一起发送回 index.html。然后，index.html 会以不同的卡片形式显示这些建议。

在本地运行应用

启动一个新的终端窗口（Ctrl+Shift+C）或任何现有终端窗口，然后输入以下命令：

python main.py

示例执行如下所示：

 * Serving Flask app 'main'
 * Debug mode: on
2025-01-21 16:02:37,473 - INFO - WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on all addresses (0.0.0.0)
 * Running on http://127.0.0.1:8080
 * Running on http://10.88.0.4:8080
2025-01-21 16:02:37,473 - INFO - Press CTRL+C to quit
2025-01-21 16:02:37,474 - INFO -  * Restarting with stat
2025-01-21 16:02:41,462 - WARNING -  * Debugger is active!
2025-01-21 16:02:41,484 - INFO -  * Debugger PIN: 440-653-555

启动应用后，点击下方的“网页预览”按钮，访问应用的主页网址：

它应显示您提供的 index.html 文件，如下所示：

提供一个示例查询（例如 Provide me some exercises for back pain relief），然后点击 Search 按钮。这应该会从数据库中检索一些推荐内容。您还会看到一个 Play Audio 按钮，该按钮会根据描述生成音频流，您可以直接收听。

9. （可选）部署到 Google Cloud Run

最后一步是将此应用部署到 Google Cloud Run。下面显示了部署命令，请确保在部署之前，将变量 (<<YOUR_PROJECT_ID>>) 的值替换为特定于您项目的值。您可以从 config.yaml 文件中检索这些值。

gcloud run deploy yogaposes --source . \
  --port=8080 \
  --allow-unauthenticated \
  --region=us-central1 \
  --platform=managed  \
  --project=<<YOUR_PROJECT_ID>> \
  --env-vars-file=config.yaml

从应用的根文件夹中执行上述命令。系统可能还会要求您启用 Google Cloud API，并确认您已了解各种权限，请按要求操作。

部署过程大约需要 5-7 分钟才能完成，因此请耐心等待。

成功部署后，部署输出将提供 Cloud Run 服务网址。其格式为：

Service URL: https://yogaposes-<<UNIQUEID>.us-central1.run.app

访问该公开网址，您应该会看到同一 Web 应用已成功部署并在运行。

您还可以访问 Google Cloud 控制台中的 Cloud Run，在 Cloud Run 中查看服务列表。yogaposes 服务应是其中列出的服务之一（如果不是唯一的服务）。

您可以点击特定服务名称（在本例中为 yogaposes），查看服务的详细信息，例如网址、配置、日志等。

至此，我们已在 Cloud Run 上开发和部署了瑜伽姿势推荐器 Web 应用。

10. 恭喜

恭喜！您已成功构建一个应用，该应用会将数据集上传到 Firestore、生成嵌入，并根据用户查询执行矢量相似度搜索。

参考文档

• Firestore
• Firestore - 使用向量嵌入进行搜索
• 在 Firestore 中管理索引
• Cloud Run
• 适用于 Firestore 的 Langchain 文档加载器
• 适用于 Firestore 的 Langchain 矢量存储区