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
# This notebook also requires the following dependencies
pip install -qU  langchain-core langchain-cohere sqlalchemy

设置 Postgres 参数

设置你的 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 为你的 Postgres 数据库配置共享连接池。这是管理连接数量并通过缓存数据库连接降低延迟的行业最佳实践。 PGVectorStore 支持 asyncpg 和 psycopg3 驱动。使用 PGEngine.from_connection_string() 创建 PGEngine 时，需要提供：

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 时，需要提供：

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

可选提示：💡

你也可以在传入 table_name 的地方同时传入 schema_name 来指定 schema 名称。例如：

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 或元数据列。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

# 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	特殊用法（在列表中）
$nin	特殊用法（不在列表中）
$between	特殊用法（介于之间）
$exists	特殊用法（是否为 null）
$like	文本（like）
$ilike	文本（不区分大小写 like）
$and	逻辑（与）
$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 列，唯一标识 products 表中的每一行。
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 格式存储商品的任意附加信息，适合存储不适合放入标准列的多样化和复杂数据。

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

注意：

可选：如果 embed 列是新创建的，或其维度与嵌入模型不匹配，则需要一次性为历史记录添加嵌入，方式如下： ALTER TABLE products ADD COLUMN embed vector(768) DEFAULT NULL
对于通过 VectorStore 新增的记录，嵌入将自动生成。

清理

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

await pg_engine.adrop_table(TABLE_NAME)

用于检索增强生成

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

API 参考

有关所有 VectorStore 功能和配置的详细文档，请访问 API 参考：python.langchain.com/api_reference/postgres/v2/langchain_postgres.v2.vectorstores.PGVectorStore.html

在 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 参考