將 LangChain 與適用於 PostgreSQL 的 Azure 資料庫一起使用

適用於 PostgreSQL 的 Azure 資料庫可順暢地與主要大型語言模型（LLM）協調流程套件整合，例如 LangChain。這項整合可讓開發人員在其應用程式中使用進階 AI 功能。 LangChain 可以簡化 LLM、內嵌模型和資料庫的管理和使用，讓產生 AI 應用程式更容易開發。

本文說明如何使用適用於 PostgreSQL 的 Azure 資料庫中的整合向量資料庫，使用 LangChain 在集合中儲存和管理檔。它也會示範如何使用最接近的鄰近演算法，例如餘弦距離、L2 距離（Euclidean distance）和內部產品，建立索引和執行向量搜尋查詢，以找出接近查詢向量的檔。

向量支援

您可以使用適用於 PostgreSQL 的 Azure 資料庫，有效率地儲存和查詢 PostgreSQL 中的數百萬個向量內嵌。此服務可協助您將 AI 使用案例從概念證明擴展至生產階段。其提供下列優點：

提供熟悉的 SQL 介面來查詢向量內嵌和關係型數據。
透過使用 pgvector，在超過 1 億個向量中進行更快速且更精確的相似性搜尋，提升了的效能。
藉由將關係型元數據、向量內嵌和時間序列數據整合到單一資料庫，來簡化作業。
利用強大功能的 PostgreSQL 生態系統和 Azure 雲端平台，提供企業級功能，包括複製和高可用性。

認證

適用於 PostgreSQL 的 Azure 資料庫支持密碼型和 Microsoft Entra （先前稱為 Azure Active Directory）驗證。

Microsoft Entra 驗證可讓您使用 Microsoft Entra 識別碼向 PostgreSQL 伺服器進行驗證。 Microsoft Entra ID 不需要管理資料庫使用者的不同使用者名稱和密碼。它可讓您使用其他 Azure 服務所使用的相同安全性機制。

在本文中，您可以使用任一驗證方法。

設定

適用於 PostgreSQL 的 Azure DB 會使用開放原始碼 LangChain Postgres 支援來連線到適用於 PostgreSQL 的 Azure DB。首先，下載合作夥伴套件：

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

在適用於 PostgreSQL 的 Azure 資料庫上啟用 pgvector

請參閱在適用於 PostgreSQL 的 Azure 資料庫中啟用和使用 pgvector。

設定認證

您需要取得適用於 PostgreSQL 的 Azure 資料庫連線詳細數據，並將其新增為環境變數。

如果您要使用 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 會使用 DefaultAzureCredential 程式庫中的 azure.identity 來擷取 Azure Database for PostgreSQL 服務的權杖。

連線可以傳遞至 connection LangChain 向量存放區的 AzurePGVectorStore 參數。

若要登入 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

collection_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(
    embeddings=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(docs, ids=[uuids[-1]], on_conflict_update=True)

查看來自向量存放區的項目

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

從向量存放區刪除項目

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

向量存放區的查詢

建立向量存放區並新增相關文件之後，您可以在鏈結或代理程序中查詢向量存放區。

篩選支援

向量存放區支援一組篩選，可透過 FilterCondition 套件中的 OrFilter、AndFilter 和針對文件的中繼資料欄位套用：

操作員	意義/類別
`=`	平等（==）
`!=`	不等式 (!=)
`<`	小於 (<)
`<=`	小於或等於（<=）
`>`	大於 (>)
`>=`	大於或等於（>=）
`in`	特殊案例 (內)
`not in`	特殊案例（不在內）
`is null`	特殊情況 (是 null)
`is not null`	特殊情況 (不是 null)
`between`	特殊案例 (介於)
`not between`	特殊情況 (不介於)
`like`	文字 (例如)
`ilike`	文字 (不區分大小寫，例如)
`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(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'}]

如果您想要在向量存放區上使用最大邊際相關性搜尋：

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'}]

如需您可以在向量存放區上 PGVector 執行的完整搜尋清單，請參閱 API 參考。

轉換成檢索器

您也可以將向量存放區轉換成擷取器，以便更輕鬆地在鏈結中使用：

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')]

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-10-30