PGVectorStore 集成

使用 LangChain Python 集成 PGVectorStore 向量存储。

PGVectorStore 是基于 postgres 后端实现的 LangChain 向量存储实现。

本笔记本将介绍如何使用 PGVectorStore API。

代码位于名为 langchain-postgres 的集成包中。

设置

该包需要 PostgreSQL 数据库并启用 pgvector 扩展。

您可以通过以下命令启动一个支持 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
# This notebook also requires the following dependencies
pip install -qU  langchain-core langchain-cohere sqlalchemy

设置 Postgres 值

设置 Postgres 值以测试本笔记本功能。

# @title Set your values or use the defaults to connect to 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 用于配置与 PostgreSQL 数据库共享连接池，这是行业最佳实践，用于管理连接数并减少延迟通过缓存数据库连接。

PGVectorStore 支持 asyncpg 和 psycopg3 驱动。

使用 PGEngine.from_connection_string() 创建 PGEngine 时需要提供：

1. url : 使用 postgresql+asyncpg 驱动的连接字符串。

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

# See docker command above to launch a Postgres instance with pgvector enabled.
CONNECTION_STRING = (
    f"postgresql+asyncpg://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}"
    f":{POSTGRES_PORT}/{POSTGRES_DB}"
)
# To use psycopg3 driver, set your connection string to `postgresql+psycopg://`

from langchain_postgres import PGEngine

pg_engine = PGEngine.from_connection_string(url=CONNECTION_STRING)

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

1. engine : AsyncEngine 对象

from sqlalchemy.ext.asyncio import create_async_engine

# Create an SQLAlchemy Async Engine
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 来指定模式名称（ wherever you pass table_name）。例如：

SCHEMA_NAME="my_schema"

await pg_engine.ainit_vectorstore_table(
    table_name=TABLE_NAME,
    vector_size=768,
    schema_name=SCHEMA_NAME,    # Default: "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()  # Re-index using the default index name
await store.areindex("my-hnsw-index")  # Re-index using the index name

删除索引

移除向量索引。

await store.adrop_vector_index()  # Delete index using the default name
await store.adrop_vector_index("my-hnsw-index")  # Delete index using the index name

创建自定义向量存储

通过特殊列名或自定义元数据列自定义向量存储。

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 可忽略不应用于文档元数据的列。这对使用现有表时很有帮助，因为现有数据列可能并非全部必要。
• 使用不同的 distance_strategy 进行向量搜索时的相似度计算。
• 使用 index_query_options 调整向量搜索时的本地索引参数。

from langchain_postgres import Column

# Set table name
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")],
)


# Initialize 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)

# Use a dictionary filter on search
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生成向量化数据，存入embed列以表示描述的语义含义。
• embedding_column="embed": embed列存储从产品描述生成的向量。这些向量用于查找具有相似描述的产品。
• metadata_columns=["name", "category", "price_usd", "quantity", "sku", "image_url"]: 这些列为产品元数据。元数据包含产品名称、类别、价格、库存量、SKU（库存单位）和图片URL等信息，用于搜索结果展示或过滤分类。
• metadata_json_column="metadata": metadata列采用灵活的JSON格式存储产品额外信息。支持存储非标准化的复杂数据。

# Set an existing table name
TABLE_NAME = "products"
# SCHEMA_NAME = "my_schema"

# Initialize PGVectorStore
custom_store = await PGVectorStore.create(
    engine=pg_engine,
    table_name=TABLE_NAME,
    # schema_name=SCHEMA_NAME,
    embedding_service=embedding,
    # Connect to existing VectorStore by customizing below column names
    id_column="product_id",
    content_column="description",
    embedding_column="embed",
    metadata_columns=["name", "category", "price_usd", "quantity", "sku", "image_url"],
    metadata_json_column="metadata",
)

注意：

1. 若embed列是新增列或维度与嵌入模型不兼容，需一次性为旧记录生成嵌入： ALTER TABLE products ADD COLUMN embed vector(768) DEFAULT NULL
2. 新增记录通过@[VectorStore]自动生成嵌入。

清理

⚠️ 注意：此操作不可逆

删除向量存储表。

await pg_engine.adrop_table(TABLE_NAME)

检索增强生成使用示例

关于如何使用此向量存储进行检索增强生成（RAG）的指南，请参考以下章节：

• 检索文档
• 使用LangChain构建RAG应用
• 智能体式RAG

---

API 参考

关于所有PGVectorStore功能和配置的详细文档，请访问@[API reference][PGVectorStore]

---

官方原文：src/oss/python/integrations/vectorstores/pgvectorstore.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。