你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

将 LangChain 与 Azure Database for PostgreSQL 配合使用

2025-09-17

Azure Database for PostgreSQL 与领先的大型语言模型（LLM）业务流程包（如 LangChain）无缝集成。此集成使开发人员能够在其应用程序中使用高级 AI 功能。 LangChain 可以简化 LLM 的管理和使用、嵌入模型和数据库，以便生成 AI 应用程序更易于开发。

本文介绍如何使用 Azure Database for PostgreSQL 中的集成矢量数据库通过 LangChain 在集合中存储和管理文档。它还演示如何使用最近的相邻算法（如余弦距离、L2 距离和内部积）创建索引和执行矢量搜索查询，以查找靠近查询向量的文档。

矢量支持

可以使用 Azure Database for PostgreSQL 有效地存储和查询 PostgreSQL 中的数百万个矢量嵌入。该服务可帮助你将 AI 用例从概念证明扩展到生产环境。它提供以下优势：

提供一个熟悉的 SQL 接口，用于查询矢量嵌入和关系数据。
使用 pgvector通过 1 亿多个向量实现更快、更精确的相似性搜索，从而增强。
通过将关系元数据、矢量嵌入和时序数据集成到单个数据库中来简化作。
使用可靠的 PostgreSQL 生态系统和 Azure 云平台的强大功能实现企业级功能，包括复制和高可用性。

身份验证

Azure Database for PostgreSQL 支持基于密码和 Microsoft Entra （前 Azure Active Directory）身份验证。

Microsoft Entra 身份验证允许使用 Microsoft Entra ID 向 PostgreSQL 服务器进行身份验证。 Microsoft Entra ID 无需为数据库用户管理单独的用户名和密码。它允许使用用于其他 Azure 服务的相同安全机制。

在本文中，可以使用任一身份验证方法。

设置

Azure Database for PostgreSQL 使用开源 LangChain Postgres 支持连接到 Azure Database for PostgreSQL。首先，下载合作伙伴包：

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

在 Azure Database for PostgreSQL 上启用 pgvector

请参阅 Azure Database for PostgreSQL 中的“启用和使用 pgvector”。

设置凭据

需要获取 Azure Database for PostgreSQL 连接详细信息，并将其添加为环境变量。

如果要使用 Microsoft Entra 身份验证，请将 USE_ENTRA_AUTH 标志设置为 True。如果使用 Microsoft Entra 身份验证，则需要提供唯一的主机和数据库名称。如果使用密码身份验证，则还需要设置用户名和密码。

import getpass
import os

USE_ENTRA_AUTH = True

# Supply the connection details for the database
os.environ["DBHOST"] = "<server-name>"
os.environ["DBNAME"] = "<database-name>"
os.environ["SSLMODE"] = "require"

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

设置 Azure OpenAI 嵌入功能

os.environ["AZURE_OPENAI_ENDPOINT"] = "<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 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,
)

矢量存储的管理

将项添加到向量存储

按 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

更新向量存储中的项

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)

查看矢量存储中的项

vector_store.get_by_ids([str(uuids[-1])])

从向量存储中删除项

vector_store.delete(ids=[uuids[-1]])

向量存储的查询

创建矢量存储并添加相关文档后，可以在链或代理中查询向量存储。

筛选支持

矢量存储支持一组筛选器，可以通过 LangChain Azure PostgreSQL 包中的 FilterCondition、OrFilter 和 AndFilter 对文档的元数据字段应用这些筛选器：

操作员	含义/类别
`=`	相等性（==）
`!=`	不平等（！=）
`<`	小于（<）
`<=`	小于或等于（<=）
`>`	大于（>）
`>=`	大于或等于（>=）
`in`	特殊大小写 (in)
`not in`	特殊大小写 (not in)
`is null`	作为特殊情况处理（为 null 时）
`is not null`	作为特殊情况处理（不为 null 时）
`between`	特殊大小写 (between)
`not between`	作为特殊情况处理（不介于）
`like`	文本 (like)
`ilike`	文本 (case-insensitive like)
`AND`	逻辑 (and)
`OR`	逻辑 (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(f"* {doc.page_content} [{doc.metadata}]")

    * there are cats in the pond [{'doc_id': 1, 'topic': 'animals', 'location': 'pond'}]
    * ducks are also found in the pond [{'doc_id': 2, 'topic': 'animals', 'location': 'pond'}]
    * the new art exhibit is fascinating [{'doc_id': 5, 'topic': 'art', 'location': 'museum'}]
    * the library hosts a weekly story time for kids [{'doc_id': 9, 'topic': 'reading', '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(f"* {doc.page_content} [{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.528338] there are cats in the pond [{'doc_id': 1, 'topic': 'animals', 'location': 'pond'}]

转换为数据检索器

还可以将矢量存储转换为检索器，以便更轻松地在链中使用：

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

[Document(id='9fe8bc1c-9a8e-4f83-b546-9b64527aa79d', metadata={'doc_id': 1, 'topic': 'animals', '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(f"* {doc.page_content} [{doc.metadata}]")

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

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

反馈

此页面是否有帮助？