Azure Database for PostgreSQL - Flexible Server 集成 - LangChain LangGraph DeepAgent中文文档

Azure Database for PostgreSQL - Flexible Server 是基于开源 Postgres 数据库引擎的关系型数据库服务。它是一种全托管的数据库即服务，能够以可预测的性能、安全性、高可用性和动态可扩展性处理关键业务工作负载。本指南展示如何利用此集成的向量数据库将文档存储到集合中、创建索引并使用近似最近邻算法（如余弦距离、L2（欧氏距离）和 IP（内积））执行向量搜索查询，以找到与查询向量相近的文档。

向量支持

Azure Database for PostgreSQL - Flexible Server 使您能够在 PostgreSQL 中高效存储和查询数百万个向量嵌入，并将您的 AI 应用从 POC 扩展到生产环境：

提供熟悉的 SQL 接口用于查询向量嵌入和关系数据。
通过 DiskANN 索引算法在 1 亿个以上的向量上以更快、更精确的相似度搜索增强 pgvector。
通过将关系型元数据、向量嵌入和时序数据集成到单一数据库中简化操作。
充分利用强大的 PostgreSQL 生态系统和 Azure Cloud 的企业级功能，包括复制和高可用性。

身份验证

Azure Database for PostgreSQL - Flexible Server 支持基于密码以及 Microsoft Entra（前身为 Azure Active Directory）身份验证。 Entra 身份验证允许您使用 Entra 身份对 PostgreSQL 服务器进行身份验证。这消除了为数据库用户管理单独用户名和密码的需要，并允许您利用与其他 Azure 服务相同的安全机制。本指南设置为支持任一身份验证方式。您可以稍后在笔记本中配置是否使用 Entra 身份验证。

设置

Azure Database for PostgreSQL 基于开源 Postgres。本集成使用专用的 langchain-azure-postgresql 包，提供包括 DiskANN 索引和 Microsoft Entra 身份验证在内的优化支持。首先下载合作伙伴包：

pip install -qU langchain-azure-postgresql langchain-openai azure-identity

启用 pgvector

请参阅 Azure Database for PostgreSQL 的启用说明。

设置凭据

您需要 Azure Database for PostgreSQL 的连接详情，并将其添加为环境变量来运行本笔记本。如果您想使用 Microsoft Entra 身份验证，请将 USE_ENTRA_AUTH 标志设置为 True。如果使用 Entra 身份验证，则只需提供主机和数据库名称。如果使用密码身份验证，还需要设置用户名和密码。

import getpass
import os

USE_ENTRA_AUTH = True

# Supply the connection details for the database
os.environ["DBHOST"] = "REPLACE_WITH_SERVER_NAME"
os.environ["DBNAME"] = "REPLACE_WITH_DATABASE_NAME"
os.environ["SSLMODE"] = "require"

if not USE_ENTRA_AUTH:
    # If using a username and password, supply them here
    os.environ["DBUSER"] = "REPLACE_WITH_USERNAME"
    os.environ["DBPASSWORD"] = getpass.getpass("Database Password:")

设置 `AzureOpenAIEmbeddings`

os.environ["AZURE_OPENAI_ENDPOINT"] = "REPLACE_WITH_AZURE_OPENAI_ENDPOINT"
os.environ["AZURE_OPENAI_API_KEY"] = getpass.getpass("Azure OpenAI API Key:")

AZURE_OPENAI_ENDPOINT = os.environ["AZURE_OPENAI_ENDPOINT"]
AZURE_OPENAI_API_KEY = os.environ["AZURE_OPENAI_API_KEY"]

from langchain_openai import AzureOpenAIEmbeddings

embeddings = AzureOpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key=AZURE_OPENAI_API_KEY,
    azure_endpoint=AZURE_OPENAI_ENDPOINT,
    azure_deployment="text-embedding-3-small",
)

初始化

使用 Microsoft Entra 身份验证

以下各节演示如何设置 LangChain 以使用 Microsoft Entra 身份验证。LangChain Azure Postgres 包中的 AzurePGConnectionPool 类通过使用 azure.identity 库中的 DefaultAzureCredential 来获取 Azure Database for PostgreSQL 服务的令牌。连接可以传入 AzurePGVectorStore LangChain 向量存储的 connection 参数中。

登录 Azure

要登录 Azure，请确保已安装 Azure CLI。您需要在终端中运行以下命令：

az login

登录后，以下代码将能够获取令牌。

from langchain_azure_postgresql.common import (
    BasicAuth,
    AzurePGConnectionPool,
    ConnectionInfo,
)
from langchain_azure_postgresql.langchain import AzurePGVectorStore

entra_connection_pool = AzurePGConnectionPool(
    azure_conn_info=ConnectionInfo(
        host=os.environ["DBHOST"], dbname=os.environ["DBNAME"]
    )
)

密码身份验证

如果您不使用 Microsoft Entra 身份验证，BasicAuth 类允许使用用户名和密码：

basic_auth_connection_pool = AzurePGConnectionPool(
    azure_conn_info=ConnectionInfo(
        host=os.environ["DBHOST"],
        dbname=os.environ["DBNAME"],
        credentials=BasicAuth(
            username=os.environ["DBUSER"],
            password=os.environ["DBPASSWORD"],
        ),
    )
)

创建向量存储

from langchain_core.documents import Document
from langchain_azure_postgresql.langchain import AzurePGVectorStore

table_name = "my_docs"

# The connection is either using Entra ID or Basic Auth
connection = entra_connection_pool if USE_ENTRA_AUTH else basic_auth_connection_pool

vector_store = AzurePGVectorStore(
    embedding=embeddings,
    table_name=table_name,
    connection=connection,
)

Metadata columns are specified as a string, defaulting to 'jsonb' type.
Embedding type is not specified, defaulting to 'vector'.
Embedding dimension is not specified, defaulting to 1536.
Embedding index is not specified, defaulting to 'DiskANN' with 'vector_cosine_ops' opclass.

配置向量存储参数

在初始化 AzurePGVectorStore 时，您可以覆盖元数据类型、嵌入维度、索引类型等默认参数，以便根据您的具体用例和数据定制向量存储。 关键配置选项：

metadata_column_type：元数据列的类型（默认值：'jsonb'）。可设置为 'jsonb'、'text' 等。
embedding_column_type：嵌入列的类型（默认值：'vector'）。
embedding_dimension：嵌入向量的维度（默认值：1536）。
embedding_index_type：向量搜索的索引类型（默认值：'DiskANN'）。其他选项可能包括 'ivfflat'、'hnsw' 等。
embedding_index_opclass：索引的操作符类（默认值：'vector_cosine_ops'）。

示例：

vector_store = AzurePGVectorStore(
    embedding=embeddings,
    table_name=table_name,
    connection=connection,
    metadata_column_type="jsonb",           # or "text"
    embedding_column_type="vector",
    embedding_dimension=768,                # set to match your model's output
    embedding_index_type="DiskANN",         # or "ivfflat", "hnsw", etc.
    embedding_index_opclass="vector_cosine_ops",  # or "vector_l2_ops", etc.
)

初始化 DiskANN 向量索引以实现更高效的向量搜索

DiskANN 是一种可扩展的近似最近邻搜索算法，用于在任意规模下实现高效的向量搜索。它提供高召回率、高每秒查询数和低查询延迟，即使对于十亿级数据集也是如此。这些特性使其成为处理大量数据的强大工具。

vector_store.create_index()

True

管理向量存储

添加条目

请注意，通过 ID 添加文档将覆盖所有与该 ID 匹配的现有文档。

docs = [
    Document(
        page_content="there are cats in the pond",
        metadata={"doc_id": 1, "location": "pond", "topic": "animals"},
    ),
    Document(
        page_content="ducks are also found in the pond",
        metadata={"doc_id": 2, "location": "pond", "topic": "animals"},
    ),
    Document(
        page_content="fresh apples are available at the market",
        metadata={"doc_id": 3, "location": "market", "topic": "food"},
    ),
    Document(
        page_content="the market also sells fresh oranges",
        metadata={"doc_id": 4, "location": "market", "topic": "food"},
    ),
    Document(
        page_content="the new art exhibit is fascinating",
        metadata={"doc_id": 5, "location": "museum", "topic": "art"},
    ),
    Document(
        page_content="a sculpture exhibit is also at the museum",
        metadata={"doc_id": 6, "location": "museum", "topic": "art"},
    ),
    Document(
        page_content="a new coffee shop opened on Main Street",
        metadata={"doc_id": 7, "location": "Main Street", "topic": "food"},
    ),
    Document(
        page_content="the book club meets at the library",
        metadata={"doc_id": 8, "location": "library", "topic": "reading"},
    ),
    Document(
        page_content="the library hosts a weekly story time for kids",
        metadata={"doc_id": 9, "location": "library", "topic": "reading"},
    ),
    Document(
        page_content="a cooking class for beginners is offered at the community center",
        metadata={"doc_id": 10, "location": "community center", "topic": "classes"},
    ),
]

uuids = vector_store.add_documents(docs)
uuids

['00e2cfe6-6e58-4733-9ebf-a708fd16488a',
 '224a22a8-567f-4e12-ac0f-5cfe4f0a4480',
 '62058e25-8f5e-4388-81c2-a5b7348ffef0',
 '1d37d282-504d-4d28-855a-8d39694b0265',
 '1fffcd2e-6fce-423f-bac3-ee5dc9084673',
 'b99efbab-2247-418f-b80d-d865f01d3c9e',
 'd2a86d1b-5d81-4c53-b3d2-a6b1e5189a3f',
 'a9577242-823e-42bc-9b0f-01670dbec190',
 'eaa45ae8-a84b-46eb-8a27-bf8652148d17',
 '7d7f04fd-6fb8-4a29-8708-b9f835ef270a']

更新条目

updated_docs = [
    Document(
        page_content="Updated - cooking class for beginners is offered at the community center",
        metadata={"doc_id": 10, "location": "community center", "topic": "classes"},
        id=uuids[-1],
    )
]
vector_store.add_documents(updated_docs, ids=[uuids[-1]], on_conflict_update=True)

['7d7f04fd-6fb8-4a29-8708-b9f835ef270a']

检索条目

vector_store.get_by_ids([str(uuids[3])])

[Document(id='1d37d282-504d-4d28-855a-8d39694b0265', metadata={'topic': 'food', 'doc_id': 4, 'location': 'market'}, page_content='the market also sells fresh oranges')]

删除条目

vector_store.delete(ids=[uuids[3]])

True

查询向量存储

创建向量存储并添加相关文档后，您可以在链或智能体中查询向量存储。

过滤

向量存储支持通过 LangChain Azure PostgreSQL 包中的 FilterCondition、OrFilter 和 AndFilter 对文档的元数据字段应用一组过滤条件：

操作符	含义/类别
`=`	等于
`!=`	不等于
`<`	小于
`<=`	小于或等于
`>`	大于
`>=`	大于或等于
`in`	特殊用法（包含）
`not in`	特殊用法（不包含）
`is null`	特殊用法（为空）
`is not null`	特殊用法（不为空）
`between`	特殊用法（介于之间）
`not between`	特殊用法（不介于之间）
`like`	文本（like）
`ilike`	文本（不区分大小写的 like）
`AND`	逻辑（与）
`OR`	逻辑（或）

直接查询

执行简单的相似度搜索如下：

from langchain_azure_postgresql import FilterCondition, AndFilter

results = vector_store.similarity_search(
    "kitty",
    k=10,
    filter=FilterCondition(
        column="(metadata->>'doc_id')::int",
        operator="in",
        value=[1, 5, 2, 9],
    ),
)

for doc in results:
    print("* " + doc.page_content + " [" + str(doc.metadata) + "]")

* there are cats in the pond [{'topic': 'animals', 'doc_id': 1, 'location': 'pond'}]
* ducks are also found in the pond [{'topic': 'animals', 'doc_id': 2, 'location': 'pond'}]
* the new art exhibit is fascinating [{'topic': 'art', 'doc_id': 5, 'location': 'museum'}]
* the library hosts a weekly story time for kids [{'topic': 'reading', 'doc_id': 9, 'location': 'library'}]

如果您想使用逻辑 AND 过滤，以下是一个示例：

results = vector_store.similarity_search(
    "ducks",
    k=10,
    filter=AndFilter(
        AND=[
            FilterCondition(
                column="(metadata->>'doc_id')::int",
                operator="in",
                value=[1, 5, 2, 9],
            ),
            FilterCondition(
                column="metadata->>'location'",
                operator="in",
                value=["pond", "market"],
            ),
        ]
    ),
)

for doc in results:
    print("* " + doc.page_content + " [" + str(doc.metadata) + "]")

* ducks are also found in the pond [{'topic': 'animals', 'doc_id': 2, 'location': 'pond'}]
* there are cats in the pond [{'topic': 'animals', 'doc_id': 1, 'location': 'pond'}]

如果您想执行相似度搜索并获取对应分数，可以运行：

results = vector_store.similarity_search_with_score(query="cats", k=1)
for doc, score in results:
    print(f"* [SIM={score:3f}] {doc.page_content} [{doc.metadata}]")

* [SIM=0.528167] there are cats in the pond [{'topic': 'animals', 'doc_id': 1, 'location': 'pond'}]

转换为检索器进行查询

您还可以将向量存储转换为检索器，以便在链中更方便地使用。

retriever = vector_store.as_retriever(search_kwargs={"k": 1})
retriever.invoke("kitty")

[Document(id='00e2cfe6-6e58-4733-9ebf-a708fd16488a', metadata={'topic': 'animals', 'doc_id': 1, 'location': 'pond'}, page_content='there are cats in the pond')]

如果您想在向量存储上使用最大边际相关性搜索：

results = vector_store.max_marginal_relevance_search(
    "query about cats",
    k=10,
    lambda_mult=0.5,
    filter=FilterCondition(
        column="(metadata->>'doc_id')::int",
        operator="in",
        value=[1, 2, 5, 9],
    ),
)

for doc in results:
    print("* " + doc.page_content + " [" + str(doc.metadata) + "]")

* there are cats in the pond [{'topic': 'animals', 'doc_id': 1, 'location': 'pond'}]
* the new art exhibit is fascinating [{'topic': 'art', 'doc_id': 5, 'location': 'museum'}]
* the library hosts a weekly story time for kids [{'topic': 'reading', 'doc_id': 9, 'location': 'library'}]
* ducks are also found in the pond [{'topic': 'animals', 'doc_id': 2, 'location': 'pond'}]

有关可以在 AzurePGVectorStore 向量存储上执行的不同搜索的完整列表，请参阅文档。

用于检索增强生成

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

API 参考

有关所有 AzurePGVectorStore 功能和配置的详细文档，请前往 API 参考：https://github.com/langchain-ai/langchain-azure/tree/main/libs/azure-postgresql/src/langchain_azure_postgresql/langchain

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

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

Popular Providers

Integrations by component

Azure Database for PostgreSQL - Flexible Server 集成

向量支持

身份验证

设置

启用 pgvector

设置凭据

设置 `AzureOpenAIEmbeddings`

初始化

使用 Microsoft Entra 身份验证

登录 Azure

密码身份验证

创建向量存储

配置向量存储参数

初始化 DiskANN 向量索引以实现更高效的向量搜索

管理向量存储

添加条目

更新条目

检索条目

删除条目

查询向量存储

过滤

直接查询

转换为检索器进行查询

用于检索增强生成

API 参考

Popular Providers

Integrations by component

​向量支持

​身份验证

​设置

​启用 pgvector

​设置凭据

​设置 AzureOpenAIEmbeddings

​初始化

​使用 Microsoft Entra 身份验证

​登录 Azure

​密码身份验证

​创建向量存储

​配置向量存储参数

​初始化 DiskANN 向量索引以实现更高效的向量搜索

​管理向量存储

​添加条目

​更新条目

​检索条目

​删除条目

​查询向量存储

​过滤

​直接查询

​转换为检索器进行查询

​用于检索增强生成

​API 参考

向量支持

身份验证

设置

启用 pgvector

设置凭据

设置 `AzureOpenAIEmbeddings`

初始化

使用 Microsoft Entra 身份验证

登录 Azure

密码身份验证

创建向量存储

配置向量存储参数

初始化 DiskANN 向量索引以实现更高效的向量搜索

管理向量存储

添加条目

更新条目

检索条目

删除条目

查询向量存储

过滤

直接查询

转换为检索器进行查询

用于检索增强生成

API 参考