跳到主要内容
Cloud SQL 是一项完全托管的关系型数据库服务，提供高性能、无缝集成和出色的可伸缩性。它提供 MySQL、PostgreSQL 和 SQL Server 数据库引擎。利用 Cloud SQL 的 LangChain 集成，扩展您的数据库应用程序以构建 AI 驱动的体验。
本笔记本介绍了如何使用 Cloud SQL for MySQL 通过 MySQLLoader 和 MySQLDocumentSaver 来保存、加载和删除 LangChain 文档。 在 GitHub 上了解有关此包的更多信息。 在 Colab 中打开

​
开始之前

要运行此 notebook，您需要执行以下操作 在本 notebook 的运行时环境中确认数据库访问权限后，填写以下值并运行该单元格，然后再运行示例脚本。 
# @markdown Please fill in the both the Google Cloud region and name of your Cloud SQL instance.
REGION = "us-central1"  # @param {type:"string"}
INSTANCE = "test-instance"  # @param {type:"string"}

# @markdown Please specify a database and a table for demo purpose.
DATABASE = "test"  # @param {type:"string"}
TABLE_NAME = "test-default"  # @param {type:"string"}

​
🦜🔗 库安装

该集成存在于其自己的 langchain-google-cloud-sql-mysql 包中，因此我们需要安装它。 
pip install -upgrade --quiet langchain-google-cloud-sql-mysql
仅限 Colab：取消注释以下单元格以重新启动内核，或使用按钮重新启动内核。对于 Vertex AI Workbench，您可以使用顶部的按钮重新启动终端。 
# # Automatically restart kernel after installs so that your environment can access the new packages
# import IPython

# app = IPython.Application.instance()
# app.kernel.do_shutdown(True)

​
☁ 设置您的 Google Cloud 项目

设置您的 Google Cloud 项目，以便您可以在此 notebook 中利用 Google Cloud 资源。 如果您不知道您的项目 ID，请尝试以下操作：
  • 运行 gcloud config list。
  • 运行 gcloud projects list。
  • 查看支持页面：查找项目 ID。
# @markdown Please fill in the value below with your Google Cloud project ID and then run the cell.

PROJECT_ID = "my-project-id"  # @param {type:"string"}

# Set the project id
!gcloud config set project {PROJECT_ID}

​
🔐 身份验证

以登录到此 notebook 的 IAM 用户身份向 Google Cloud 进行身份验证，以访问您的 Google Cloud 项目。
  • 如果您正在使用 Colab 运行此 notebook，请使用下面的单元格并继续。
  • 如果您正在使用 Vertex AI Workbench，请查看此处的设置说明。
from google.colab import auth

auth.authenticate_user()

​
基本用法

​
MySQL 引擎连接池

在从 MySQL 表保存或加载文档之前，我们首先需要配置一个到 Cloud SQL 数据库的连接池。MySQLEngine 配置了一个到您的 Cloud SQL 数据库的连接池，实现了从您的应用程序成功连接并遵循行业最佳实践。 要使用 MySQLEngine.from_instance() 创建一个 MySQLEngine，您只需提供 4 个信息：
  1. project_id：Cloud SQL 实例所在的 Google Cloud 项目的项目 ID。
  2. region：Cloud SQL 实例所在的区域。
  3. instance：Cloud SQL 实例的名称。
  4. database：要连接到 Cloud SQL 实例上的数据库名称。
默认情况下，将使用 IAM 数据库身份验证作为数据库身份验证方法。此库使用从环境中获取的应用程序默认凭据 (ADC) 所属的 IAM 主体。 有关 IAM 数据库身份验证的更多信息，请参阅： 或者，也可以使用内置数据库身份验证，通过用户名和密码访问 Cloud SQL 数据库。只需向 MySQLEngine.from_instance() 提供可选的 user 和 password 参数。
  • user：用于内置数据库身份验证和登录的数据库用户
  • password：用于内置数据库身份验证和登录的数据库密码。
from langchain_google_cloud_sql_mysql import MySQLEngine

engine = MySQLEngine.from_instance(
    project_id=PROJECT_ID, region=REGION, instance=INSTANCE, database=DATABASE
)

​
初始化表

通过 MySQLEngine.init_document_table(<table_name>) 初始化一个默认架构的表。表列：
  • page_content (类型: text)
  • langchain_metadata (类型: JSON)
overwrite_existing=True 标志表示新初始化的表将替换任何同名的现有表。 
engine.init_document_table(TABLE_NAME, overwrite_existing=True)

​
保存文档

使用 MySQLDocumentSaver.add_documents(<documents>) 保存 LangChain 文档。要初始化 MySQLDocumentSaver 类，您需要提供两项内容：
  1. engine - MySQLEngine 实例。
  2. table_name - 在 Cloud SQL 数据库中存储 LangChain 文档的表的名称。
from langchain_core.documents import Document
from langchain_google_cloud_sql_mysql import MySQLDocumentSaver

test_docs = [
    Document(
        page_content="Apple Granny Smith 150 0.99 1",
        metadata={"fruit_id": 1},
    ),
    Document(
        page_content="Banana Cavendish 200 0.59 0",
        metadata={"fruit_id": 2},
    ),
    Document(
        page_content="Orange Navel 80 1.29 1",
        metadata={"fruit_id": 3},
    ),
]
saver = MySQLDocumentSaver(engine=engine, table_name=TABLE_NAME)
saver.add_documents(test_docs)

​
加载文档

使用 MySQLLoader.load() 或 MySQLLoader.lazy_load() 加载 langchain 文档。lazy_load 返回一个生成器，只在迭代期间查询数据库。要初始化 MySQLLoader 类，您需要提供
  1. engine - MySQLEngine 实例。
  2. table_name - 在 Cloud SQL 数据库中存储 LangChain 文档的表的名称。
from langchain_google_cloud_sql_mysql import MySQLLoader

loader = MySQLLoader(engine=engine, table_name=TABLE_NAME)
docs = loader.lazy_load()
for doc in docs:
    print("Loaded documents:", doc)

​
通过查询加载文档

除了从表中加载文档，我们还可以选择从 SQL 查询生成的视图中加载文档。例如： 
from langchain_google_cloud_sql_mysql import MySQLLoader

loader = MySQLLoader(
    engine=engine,
    query=f"select * from `{TABLE_NAME}` where JSON_EXTRACT(langchain_metadata, '$.fruit_id') = 1;",
)
onedoc = loader.load()
onedoc
从 SQL 查询生成的视图可以具有与默认表不同的模式。在这种情况下，MySQLLoader 的行为与从具有非默认模式的表加载相同。请参阅 加载具有自定义文档页面内容和元数据的文档 部分。

​
删除文档

使用 MySQLDocumentSaver.delete(<documents>) 从 MySQL 表中删除 LangChain 文档列表。 对于具有默认模式（page_content，langchain_metadata）的表，删除条件是： 如果列表中存在一个 document，使得以下条件成立，则应删除 row：
  • document.page_content 等于 row[page_content]
  • document.metadata 等于 row[langchain_metadata]
from langchain_google_cloud_sql_mysql import MySQLLoader

loader = MySQLLoader(engine=engine, table_name=TABLE_NAME)
docs = loader.load()
print("Documents before delete:", docs)
saver.delete(onedoc)
print("Documents after delete:", loader.load())

​
高级用法

​
加载具有自定义文档页面内容和元数据的文档

首先我们准备一个非默认模式的示例表，并用一些任意数据填充它。 
import sqlalchemy

with engine.connect() as conn:
    conn.execute(sqlalchemy.text(f"DROP TABLE IF EXISTS `{TABLE_NAME}`"))
    conn.commit()
    conn.execute(
        sqlalchemy.text(
            f"""
            CREATE TABLE IF NOT EXISTS `{TABLE_NAME}`(
                fruit_id INT AUTO_INCREMENT PRIMARY KEY,
                fruit_name VARCHAR(100) NOT NULL,
                variety VARCHAR(50),
                quantity_in_stock INT NOT NULL,
                price_per_unit DECIMAL(6,2) NOT NULL,
                organic TINYINT(1) NOT NULL
            )
            """
        )
    )
    conn.execute(
        sqlalchemy.text(
            f"""
            INSERT INTO `{TABLE_NAME}` (fruit_name, variety, quantity_in_stock, price_per_unit, organic)
            VALUES
                ('Apple', 'Granny Smith', 150, 0.99, 1),
                ('Banana', 'Cavendish', 200, 0.59, 0),
                ('Orange', 'Navel', 80, 1.29, 1);
            """
        )
    )
    conn.commit()
如果我们仍然使用 MySQLLoader 的默认参数从这个示例表中加载 LangChain 文档，那么加载文档的 page_content 将是表的第一个列，而 metadata 将由所有其他列的键值对组成。 
loader = MySQLLoader(
    engine=engine,
    table_name=TABLE_NAME,
)
loader.load()
我们可以在初始化 MySQLLoader 时通过设置 content_columns 和 metadata_columns 来指定要加载的内容和元数据。
  1. content_columns：要写入文档 page_content 的列。
  2. metadata_columns：要写入文档 metadata 的列。
例如，这里，content_columns 中的列值将连接成一个以空格分隔的字符串，作为加载文档的 page_content，而加载文档的 metadata 将仅包含 metadata_columns 中指定的列的键值对。 
loader = MySQLLoader(
    engine=engine,
    table_name=TABLE_NAME,
    content_columns=[
        "variety",
        "quantity_in_stock",
        "price_per_unit",
        "organic",
    ],
    metadata_columns=["fruit_id", "fruit_name"],
)
loader.load()

​
保存具有自定义页面内容和元数据的文档

为了将 LangChain 文档保存到具有自定义元数据字段的表中，我们首先需要通过 MySQLEngine.init_document_table() 创建这样一个表，并指定我们想要的 metadata_columns 列表。在此示例中，创建的表将具有以下表列：
  • description (类型: text): 用于存储水果描述。
  • fruit_name (类型: text): 用于存储水果名称。
  • organic (类型: tinyint(1)): 用于指示水果是否为有机。
  • other_metadata (类型: JSON): 用于存储水果的其他元数据信息。
我们可以使用 MySQLEngine.init_document_table() 的以下参数来创建表：
  1. table_name：Cloud SQL 数据库中用于存储 langchain 文档的表的名称。
  2. metadata_columns：一个 sqlalchemy.Column 列表，指示所需的元数据列。
  3. content_column：用于存储 langchain 文档 page_content 的列名。默认值：page_content。
  4. metadata_json_column：用于存储 langchain 文档额外 metadata 的 JSON 列名。默认值：langchain_metadata。
engine.init_document_table(
    TABLE_NAME,
    metadata_columns=[
        sqlalchemy.Column(
            "fruit_name",
            sqlalchemy.UnicodeText,
            primary_key=False,
            nullable=True,
        ),
        sqlalchemy.Column(
            "organic",
            sqlalchemy.Boolean,
            primary_key=False,
            nullable=True,
        ),
    ],
    content_column="description",
    metadata_json_column="other_metadata",
    overwrite_existing=True,
)
使用 MySQLDocumentSaver.add_documents(<documents>) 保存文档。如本例所示：
  • document.page_content 将保存到 description 列。
  • document.metadata.fruit_name 将保存到 fruit_name 列。
  • document.metadata.organic 将保存到 organic 列。
  • document.metadata.fruit_id 将以 JSON 格式保存到 other_metadata 列。
test_docs = [
    Document(
        page_content="Granny Smith 150 0.99",
        metadata={"fruit_id": 1, "fruit_name": "Apple", "organic": 1},
    ),
]
saver = MySQLDocumentSaver(
    engine=engine,
    table_name=TABLE_NAME,
    content_column="description",
    metadata_json_column="other_metadata",
)
saver.add_documents(test_docs)

with engine.connect() as conn:
    result = conn.execute(sqlalchemy.text(f"select * from `{TABLE_NAME}`;"))
    print(result.keys())
    print(result.fetchall())

​
删除具有自定义页面内容和元数据的文档

我们还可以通过 MySQLDocumentSaver.delete(<documents>) 从具有自定义元数据列的表中删除文档。删除标准是： 如果列表中存在一个 document，使得
  • document.page_content 等于 row[page_content]
  • 对于 document.metadata 中的每个元数据字段 k
    • document.metadata[k] 等于 row[k] 或 document.metadata[k] 等于 row[langchain_metadata][k]
  • row 中不存在 document.metadata 中没有的额外元数据字段。
loader = MySQLLoader(engine=engine, table_name=TABLE_NAME)
docs = loader.load()
print("Documents before delete:", docs)
saver.delete(docs)
print("Documents after delete:", loader.load())
在 GitHub 上编辑此页面源文件。
以编程方式连接这些文档到 Claude、VSCode 等，通过 MCP 获取实时答案。
© . This site is unofficial and not affiliated with LangChain, Inc.