搭配適用於 PostgreSQL 的 Azure 資料庫使用 LangChain

2025-06-23

適用於 PostgreSQL 的 Azure 資料庫與 LangChain 等領先的大型語言模型（LLM）協調流程套件緊密整合，讓開發人員能夠運用其應用程式內進階 AI 功能的強大功能。 LangChain 可以簡化 LLM、內嵌模型和資料庫的管理和使用，讓您更輕鬆地開發 Generative AI 應用程式。

本教學課程說明如何使用適用於 PostgreSQL 的 Azure 資料庫整合向量資料庫，使用 LangChain 在集合中儲存和管理檔。它也會示範如何使用近似的鄰近演算法來建立索引和執行向量搜尋查詢，例如 Cosine Distance、L2（Euclidean distance）和 IP（內部乘積），以找出接近查詢向量的檔。

向量支援

適用於 PostgreSQL 的 Azure 資料庫 - 彈性伺服器可讓您有效率地儲存和查詢 PostgreSQL 中的數百萬個向量內嵌，並將 AI 使用案例從 POC（概念證明）調整為生產環境：

提供熟悉的 SQL 介面來查詢向量內嵌和關係型數據。
使用 DiskANN 索引演算法，提升 pgvector 在 100M+ 向量之間更快速且更精確的相似度搜尋。
藉由將關係型元數據、向量內嵌和時間序列數據整合到單一資料庫，來簡化作業。
利用強大的 PostgreSQL 生態系統和 Azure Cloud 的強大功能來提供企業級的功能，包括資料複寫和高可用性。

認證

適用於 PostgreSQL 的 Azure 資料庫 - 彈性伺服器支援密碼型和 Microsoft Entra （先前稱為 Azure Active Directory）驗證。 Entra 驗證可讓您使用 Entra 身分識別向 PostgreSQL 伺服器進行驗證。 Entra ID 不需要管理資料庫使用者的不同使用者名稱和密碼，並可讓您使用相同的安全性機制來管理其他 Azure 服務。

此筆記本已設定為使用任一驗證方法。您可以設定是否稍後要在筆記本中使用 Entra 驗證。

設定

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

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

在 Azure Database for PostgreSQL - Flexible Server 上啟用 pgvector

請參閱適用於 PostgreSQL 的 Azure 資料庫啟用指示。

憑據

您需要適用於 PostgreSQL 的 Azure 資料庫連線詳細數據，並將其新增為環境變數來執行此筆記本。

如果您要使用 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"] = "<server-name>"
os.environ["DBNAME"] = "<database-name>"
os.environ["SSLMODE"] = "require"

if not USE_ENTRA_AUTH:
    # If 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 設定為使用 Entra 驗證的函式。它提供一個函式get_token_and_username使用 azure.identity 程式庫中的 DefaultAzureCredential 擷取適用於 PostgreSQL 的 Azure 資料庫服務的權杖。它可確保 sqlalchemy 引擎具有有效的令牌，可用來建立新的連線。它也會解析令牌，即 Java Web 令牌 (JWT)，以提取用來連接到資料庫的用戶名稱。

create_postgres_engine函式會建立 sqlalchemy Engine ，根據從 TokenManager 擷取的令牌動態設定使用者名稱和密碼。這 Engine 可以傳遞至 connection LangChain VectorStore 的參數 PGVector 。

登入 Azure

若要登入 Azure，請確定您已安裝 Azure CLI 。您必須在終端機中執行下列命令：

az login

登入之後，下列程式代碼會擷取令牌。

import base64
import json
from functools import lru_cache

from azure.identity import DefaultAzureCredential
from sqlalchemy import create_engine, event
from sqlalchemy.engine.url import URL


@lru_cache(maxsize=1)
def get_credential():
    """Memoized function to create the Azure credential, which caches tokens."""
    return DefaultAzureCredential()


def decode_jwt(token):
    """Decode the JWT payload to extract claims."""
    payload = token.split(".")[1]
    padding = "=" * (4 - len(payload) % 4)
    decoded_payload = base64.urlsafe_b64decode(payload + padding)
    return json.loads(decoded_payload)


def get_token_and_username():
    """Fetches a token returns the username and token."""
    # Fetch a new token and extract the username
    token = get_credential().get_token(
        "https://ossrdbms-aad.database.windows.net/.default"
    )
    claims = decode_jwt(token.token)
    username = claims.get("upn")
    if not username:
        raise ValueError("Could not extract username from token. Have you logged in?")

    return username, token.token


def create_postgres_engine():
    db_url = URL.create(
        drivername="postgresql+psycopg",
        username="",  # This will be replaced dynamically
        password="",  # This will be replaced dynamically
        host=os.environ["DBHOST"],
        port=os.environ.get("DBPORT", 5432),
        database=os.environ["DBNAME"],
    )

    # Create a sqlalchemy engine
    engine = create_engine(db_url, echo=True)

    # Listen for the connection event to inject dynamic credentials
    @event.listens_for(engine, "do_connect")
    def provide_dynamic_credentials(dialect, conn_rec, cargs, cparams):
        # Fetch the dynamic username and token
        username, token = get_token_and_username()

        # Override the connection parameters
        cparams["user"] = username
        cparams["password"] = token

    return engine

密碼驗證

如果未使用 Entra 驗證， get_connection_uri 則會提供從環境變數提取使用者名稱和密碼的連接 URI。

import urllib.parse


def get_connection_uri():
    # Read URI parameters from the environment
    dbhost = os.environ["DBHOST"]
    dbname = os.environ["DBNAME"]
    dbuser = urllib.parse.quote(os.environ["DBUSER"])
    password = os.environ["DBPASSWORD"]
    sslmode = os.environ["SSLMODE"]

    # Construct connection URI
    # Use psycopg 3!
    db_uri = (
        f"postgresql+psycopg://{dbuser}:{password}@{dbhost}/{dbname}?sslmode={sslmode}"
    )
    return db_uri

建立向量存放區

from langchain_core.documents import Document
from langchain_postgres import PGVector
from langchain_postgres.vectorstores import PGVector

collection_name = "my_docs"

# The connection is either a sqlalchemy engine or a connection URI
connection = create_postgres_engine() if USE_ENTRA_AUTH else get_connection_uri()

vector_store = PGVector(
    embeddings=embeddings,
    collection_name=collection_name,
    connection=connection,
    use_jsonb=True,
)

管理向量存放區

將項目新增至向量存儲庫

依識別碼新增文件會覆蓋任何符合該識別碼的現有文件。

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

vector_store.add_documents(docs, ids=[doc.metadata["id"] for doc in docs])

更新向量存放區中的項目

docs = [
    Document(
        page_content="Updated - cooking class for beginners is offered at the community center",
        metadata={"id": 10, "location": "community center", "topic": "classes"},
    )
]
vector_store.add_documents(docs, ids=[doc.metadata["id"] for doc in docs])

從向量儲存庫刪除項目

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

查詢向量存放區

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

篩選支援

向量存放區支援一組篩選條件，可套用至檔的元數據欄位。

操作員	意義/類別
$eq	Equality (==)
$ne	不等式 (!=)
$lt	小於 (<)
$lte	小於或等於（<=）
$gt	大於 (>)
$gte	大於或等於（>=）
$in	特殊案例 (內)
$nin	特殊案例 (不在內)
$between	特殊案例 (介於)
$like	文字 (例如)
$ilike	文字 (不區分大小寫，例如)
$and	邏輯 (和)
$or	邏輯 (或)

直接查詢

您可以執行簡單的相似度搜尋，如下所示：

results = vector_store.similarity_search(
    "kitty", k=10, filter={"id": {"$in": [1, 5, 2, 9]}}
)
for doc in results:
    print(f"* {doc.page_content} [{doc.metadata}]")

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

如果您提供具有多個字段的字典，但沒有運算符，最上層會被解釋為邏輯 AND 篩選條件。

vector_store.similarity_search(
    "ducks",
    k=10,
    filter={"id": {"$in": [1, 5, 2, 9]}, "location": {"$in": ["pond", "market"]}},
)

[Document(id='2', metadata={'id': 2, 'topic': 'animals', 'location': 'pond'}, page_content='ducks are also found in the pond'),
 Document(id='1', metadata={'id': 1, 'topic': 'animals', 'location': 'pond'}, page_content='there are cats in the pond')]

vector_store.similarity_search(
    "ducks",
    k=10,
    filter={
        "$and": [
            {"id": {"$in": [1, 5, 2, 9]}},
            {"location": {"$in": ["pond", "market"]}},
        ]
    },
)

[Document(id='2', metadata={'id': 2, 'topic': 'animals', 'location': 'pond'}, page_content='ducks are also found in the pond'),
 Document(id='1', metadata={'id': 1, 'topic': 'animals', 'location': 'pond'}, page_content='there are cats in the 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 [{'id': 1, 'topic': 'animals', 'location': 'pond'}]

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

透過轉換為檢索器來查詢

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

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

[Document(id='1', metadata={'id': 1, 'topic': 'animals', 'location': 'pond'}, page_content='there are cats in the pond')]

目前的限制

langchain_postgres僅適用於 psycopg3。將連接字串從 postgresql+psycopg2://... 更新為 postgresql+psycopg://langchain:langchain@...
內嵌存放區和集合的架構已變更，以使add_documents能夠與使用者指定的ID正確運作。
現在必須傳遞明確的連接物件。

目前 沒有任何機制 支援在架構變更上輕鬆移轉數據。因此，向量存放區中的任何架構變更都需要使用者重新建立數據表並讀取檔。

共用方式為