PGVectorStore 集成 - Docs by LangChain

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,
)

可选提示：💡

您还可以通过在传递 table_name 的任何地方传递 schema_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 或元数据列。Column 由名称和数据类型定义。可以使用任何 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，它存储电子商务企业的商品详情。

​设置

​安装

​设置您的 Postgres 值

​初始化

​PGEngine 连接池

​初始化表

​可选提示：💡

​创建嵌入类实例

​初始化默认 PGVectorStore

​管理向量存储

​添加文档

​添加文本

​删除文档

​查询向量存储

​搜索文档

​按向量搜索文档

​添加索引

​重新索引

​删除索引

​创建自定义向量存储

​可选提示：💡

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

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

设置

安装

设置您的 Postgres 值

初始化

PGEngine 连接池

初始化表

可选提示：💡

创建嵌入类实例

初始化默认 PGVectorStore

管理向量存储

添加文档

添加文本

删除文档

查询向量存储

搜索文档

按向量搜索文档

添加索引

重新索引

删除索引

创建自定义向量存储

可选提示：💡

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

使用现有表创建向量存储