PGVectorStore 集成

PGVectorStore 是一个使用 postgres 作为后端的 LangChain 向量存储实现。本笔记本介绍如何使用 PGVectorStore API。代码位于名为 langchain-postgres 的集成包中。

设置

此包需要一个安装了 pgvector 扩展的 PostgreSQL 数据库。您可以运行以下命令来启动一个启用 pgvector 的 Postgres 实例容器：

docker run --name pgvector-container -e POSTGRES_USER=langchain -e POSTGRES_PASSWORD=langchain -e POSTGRES_DB=langchain -p 6024:5432 -d pgvector/pgvector:pg16

安装

安装集成库 langchain-postgres。

pip install -qU  langchain-postgres
# 本笔记本还需要以下依赖项
pip install -qU  langchain-core langchain-cohere sqlalchemy

设置您的 Postgres 值

设置您的 Postgres 值，以针对 Postgres 实例测试本笔记本中的功能。

# @title 设置您的值或使用默认值连接到 Docker { display-mode: "form" }
POSTGRES_USER = "langchain"  # @param {type: "string"}
POSTGRES_PASSWORD = "langchain"  # @param {type: "string"}
POSTGRES_HOST = "localhost"  # @param {type: "string"}
POSTGRES_PORT = "6024"  # @param {type: "string"}
POSTGRES_DB = "langchain"  # @param {type: "string"}
TABLE_NAME = "vectorstore"  # @param {type: "string"}
VECTOR_SIZE = 1024  # @param {type: "int"}

初始化

PGEngine 连接池

将 PostgreSQL 建立为向量存储的要求和参数之一是 PGEngine 对象。PGEngine 配置到您的 Postgres 数据库的共享连接池。这是管理连接数量和通过缓存数据库连接减少延迟的行业最佳实践。 PGVectorStore 可与 asyncpg 和 psycopg3 驱动程序一起使用。要使用 PGEngine.from_connection_string() 创建 PGEngine，您需要提供：

url：使用 postgresql+asyncpg 驱动程序的连接字符串。

注意： 本教程演示了异步接口。所有异步方法都有对应的同步方法。

# 请参阅上面的 docker 命令以启动启用了 pgvector 的 Postgres 实例。
CONNECTION_STRING = (
    f"postgresql+asyncpg://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}"
    f":{POSTGRES_PORT}/{POSTGRES_DB}"
)
# 要使用 psycopg3 驱动程序，请将连接字符串设置为 `postgresql+psycopg://`

from langchain_postgres import PGEngine

pg_engine = PGEngine.from_connection_string(url=CONNECTION_STRING)

要使用 PGEngine.from_engine() 创建 PGEngine，您需要提供：

engine：AsyncEngine 对象

from sqlalchemy.ext.asyncio import create_async_engine

# 创建 SQLAlchemy 异步引擎
engine = create_async_engine(
    CONNECTION_STRING,
)

pg_engine = PGEngine.from_engine(engine=engine)

初始化表

PGVectorStore 类需要一个数据库表。PGEngine 引擎有一个辅助方法 ainit_vectorstore_table()，可用于为您创建具有正确模式的表。有关自定义模式，请参阅创建自定义向量存储或使用现有表创建向量存储。

await pg_engine.ainit_vectorstore_table(
    table_name=TABLE_NAME,
    vector_size=VECTOR_SIZE,
)

可选提示：💡

您还可以通过传递 schema_name 来指定模式名称，与传递 table_name 时一样。例如：

SCHEMA_NAME="my_schema"

await pg_engine.ainit_vectorstore_table(
    table_name=TABLE_NAME,
    vector_size=768,
    schema_name=SCHEMA_NAME,    # 默认值: "public"
)

创建嵌入类实例

您可以使用任何 LangChain 嵌入模型。

from langchain_cohere import CohereEmbeddings

embedding = CohereEmbeddings(model="embed-english-v3.0")

初始化默认的 PGVectorStore

使用默认表模式连接到向量存储。有关自定义模式，请参阅创建自定义向量存储或使用现有表创建向量存储。

from langchain_postgres import PGVectorStore

store = await PGVectorStore.create(
    engine=pg_engine,
    table_name=TABLE_NAME,
    # schema_name=SCHEMA_NAME,
    embedding_service=embedding,
)

管理向量存储

添加文档

将文档添加到向量存储。元数据存储在 JSON 列中，请参阅“创建自定义向量存储”以存储用于过滤的元数据。

import uuid

from langchain_core.documents import Document

docs = [
    Document(
        id=str(uuid.uuid4()),
        page_content="Red Apple",
        metadata={"description": "red", "content": "1", "category": "fruit"},
    ),
    Document(
        id=str(uuid.uuid4()),
        page_content="Banana Cavendish",
        metadata={"description": "yellow", "content": "2", "category": "fruit"},
    ),
    Document(
        id=str(uuid.uuid4()),
        page_content="Orange Navel",
        metadata={"description": "orange", "content": "3", "category": "fruit"},
    ),
]

await store.aadd_documents(docs)

添加文本

直接将文本添加到向量存储，如果未结构化为 Document。

import uuid

all_texts = ["Apples and oranges", "Cars and airplanes", "Pineapple", "Train", "Banana"]
metadatas = [{"len": len(t)} for t in all_texts]
ids = [str(uuid.uuid4()) for _ in all_texts]

await store.aadd_texts(all_texts, metadatas=metadatas, ids=ids)

删除文档

可以使用 ID 删除文档。

await store.adelete([ids[1]])

查询向量存储

搜索文档

使用自然语言查询搜索相似文档。

query = "I'd like a fruit."
docs = await store.asimilarity_search(query)
print(docs)

按向量搜索文档

使用向量嵌入搜索相似文档。

query_vector = embedding.embed_query(query)
docs = await store.asimilarity_search_by_vector(query_vector, k=2)
print(docs)

添加索引

通过应用向量索引来加速向量搜索查询。了解有关向量索引的更多信息。如果未提供名称，索引将使用默认索引名称。要添加多个索引，需要不同的索引名称。

from langchain_postgres.v2.indexes import HNSWIndex, IVFFlatIndex

index = IVFFlatIndex()
await store.aapply_vector_index(index)

index = HNSWIndex(name="my-hnsw-index")
await store.aapply_vector_index(index)

设置索引参数以调整索引，以在召回率和 QPS 之间实现最佳平衡。

index = IVFFlatIndex(name="my-ivfflat", lists=120)
await store.aapply_vector_index(index)

重新索引

使用索引表中存储的数据重建索引，替换索引的旧副本。某些索引类型在添加大量新数据后可能需要重新索引。

await store.areindex()  # 使用默认索引名称重新索引
await store.areindex("my-hnsw-index")  # 使用索引名称重新索引

删除索引

删除向量索引。

await store.adrop_vector_index()  # 使用默认名称删除索引
await store.adrop_vector_index("my-hnsw-index")  # 使用索引名称删除索引

创建自定义向量存储

使用特殊列名或自定义元数据列自定义向量存储。 ainit_vectorstore_table

使用字段 content_column、embedding_column、metadata_columns、metadata_json_column、id_column 来重命名列。
使用 Column 类创建自定义 ID 或元数据列。列由名称和数据类型定义。可以使用任何 Postgres 数据类型。
使用 store_metadata 创建 JSON 列以存储额外元数据。

可选提示：💡

要使用非 UUID ID，必须自定义 ID 列：

await pg_engine.ainit_vectorstore_table(
    ...,
    id_column=Column(name="langchain_id", data_type="INTEGER")
)

PGVectorStore

使用字段 content_column、embedding_column、metadata_columns、metadata_json_column、id_column 来重命名列。
ignore_metadata_columns 用于忽略不应用于 Document 元数据的列。当使用预先存在的表且所有数据列并非必需时，这很有帮助。
使用不同的 distance_strategy 进行向量搜索期间的相似性计算。
使用 index_query_options 在向量搜索期间调整本地索引参数。

from langchain_postgres import Column

# 设置表名
TABLE_NAME = "vectorstore_custom"
# SCHEMA_NAME = "my_schema"

await pg_engine.ainit_vectorstore_table(
    table_name=TABLE_NAME,
    # schema_name=SCHEMA_NAME,
    vector_size=VECTOR_SIZE,
    metadata_columns=[Column("len", "INTEGER")],
)


# 初始化 PGVectorStore
custom_store = await PGVectorStore.create(
    engine=pg_engine,
    table_name=TABLE_NAME,
    # schema_name=SCHEMA_NAME,
    embedding_service=embedding,
    metadata_columns=["len"],
)

使用元数据过滤器搜索文档

向量存储可以利用关系数据来过滤相似性搜索。向量存储支持一组可以针对文档元数据字段应用的过滤器。有关如何迁移到使用元数据列的详细信息，请参阅迁移指南。 PGVectorStore 当前支持以下运算符和所有 Postgres 数据类型。

运算符	含义/类别
$eq	相等 (==)
$ne	不相等 (!=)
$lt	小于 (<)
$lte	小于或等于 (<=)
$gt	大于 (>)
$gte	大于或等于 (>=)
$in	特殊情况 (in)
$nin	特殊情况 (not in)
$between	特殊情况 (between)
$exists	特殊情况 (is null)
$like	文本 (like)
$ilike	文本 (不区分大小写的 like)
$and	逻辑 (and)
$or	逻辑 (or)

import uuid

docs = [
    Document(
        id=str(uuid.uuid4()),
        page_content="Red Apple",
        metadata={"description": "red", "content": "1", "category": "fruit"},
    ),
    Document(
        id=str(uuid.uuid4()),
        page_content="Banana Cavendish",
        metadata={"description": "yellow", "content": "2", "category": "fruit"},
    ),
    Document(
        id=str(uuid.uuid4()),
        page_content="Orange Navel",
        metadata={"description": "orange", "content": "3", "category": "fruit"},
    ),
]

await custom_store.aadd_documents(docs)

# 在搜索时使用字典过滤器
docs = await custom_store.asimilarity_search(query, filter={"content": {"$gte": 1}})

print(docs)

使用现有表创建向量存储

向量存储可以构建在现有表之上。假设 PG 数据库中有一个预先存在的表：products，用于存储电子商务企业的产品详细信息。以下是此表如何映射到 PGVectorStore：

id_column="product_id"：ID 列唯一标识产品表中的每一行。
content_column="description"：description 列包含每个产品的文本描述。此文本由 embedding_service 用于创建进入 embedding_column 的向量，表示每个描述的语义含义。
embedding_column="embed"：embed 列存储从产品描述创建的向量。这些向量用于查找具有相似描述的产品。
metadata_columns=["name", "category", "price_usd", "quantity", "sku", "image_url"]：这些列被视为每个产品的元数据。元数据提供有关产品的附加信息，例如其名称、类别、价格、可用数量、SKU（库存单位）和图像 URL。此信息对于在搜索结果中显示产品详细信息或用于过滤和分类非常有用。
metadata_json_column="metadata"：metadata 列可以以灵活的 JSON 格式存储有关产品的任何附加信息。这允许存储不适合标准列的多样化和复杂数据。

# 设置现有表名
TABLE_NAME = "products"
# SCHEMA_NAME = "my_schema"

# 初始化 PGVectorStore
custom_store = await PGVectorStore.create(
    engine=pg_engine,
    table_name=TABLE_NAME,
    # schema_name=SCHEMA_NAME,
    embedding_service=embedding,
    # 通过自定义以下列名连接到现有向量存储
    id_column="product_id",
    content_column="description",
    embedding_column="embed",
    metadata_columns=["name", "category", "price_usd", "quantity", "sku", "image_url"],
    metadata_json_column="metadata",
)

注意：

可选：如果 embed 列是新创建的或维度与嵌入模型支持的不同，则需要一次性为旧记录添加嵌入，如下所示： ALTER TABLE products ADD COLUMN embed vector(768) DEFAULT NULL
对于通过 VectorStore 添加的新记录，会自动生成嵌入。

清理

⚠️ 警告：此操作无法撤销 删除向量存储表。

await pg_engine.adrop_table(TABLE_NAME)

用于检索增强生成的用法

有关如何将此向量存储用于检索增强生成 (RAG) 的指南，请参阅以下部分：

API 参考

有关所有 PGVectorStore 功能和配置的详细文档，请前往 API 参考

在 GitHub 上编辑此页面或提交问题。

连接这些文档到 Claude、VSCode 等，通过 MCP 获取实时答案。

Popular Providers

Integrations by component

设置

安装

设置您的 Postgres 值

初始化

PGEngine 连接池

初始化表

可选提示：💡

创建嵌入类实例

初始化默认的 PGVectorStore

管理向量存储

添加文档

添加文本

删除文档

查询向量存储

搜索文档

按向量搜索文档

添加索引

重新索引

删除索引

创建自定义向量存储

可选提示：💡

使用元数据过滤器搜索文档

使用现有表创建向量存储

清理

用于检索增强生成的用法

API 参考

Popular Providers

Integrations by component

​设置

​安装

​设置您的 Postgres 值

​初始化

​PGEngine 连接池

​初始化表

​可选提示：💡

​创建嵌入类实例

​初始化默认的 PGVectorStore

​管理向量存储

​添加文档

​添加文本

​删除文档

​查询向量存储

​搜索文档

​按向量搜索文档

​添加索引

​重新索引

​删除索引

​创建自定义向量存储

​可选提示：💡

​使用元数据过滤器搜索文档

​使用现有表创建向量存储

​清理

​用于检索增强生成的用法

​API 参考

设置

安装

设置您的 Postgres 值

初始化

PGEngine 连接池

初始化表

可选提示：💡

创建嵌入类实例

初始化默认的 PGVectorStore

管理向量存储

添加文档

添加文本

删除文档

查询向量存储

搜索文档

按向量搜索文档

添加索引

重新索引

删除索引

创建自定义向量存储

可选提示：💡

使用元数据过滤器搜索文档

使用现有表创建向量存储

清理

用于检索增强生成的用法

API 参考