將 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 雲端平台，提供企業級功能，包括複製和高可用性。

Authentication

適用於 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 來擷取適用於 PostgreSQL 的 Azure 資料庫服務的權杖。

連線可以傳遞至 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 和針對文件的中繼資料欄位套用：

Operator	意義/類別
`=`	平等（==）
`!=`	不等式 (!=)
`<`	小於 (<)
`<=`	小於或等於（<=）
`>`	大於 (>)
`>=`	大於或等於（>=）
`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 2026-01-20