Verwenden des Postgres Vector Store-Connectors (Vorschau)

Überblick

Der Connector für den Postgres Vector Store kann verwendet werden, um auf Daten in Postgres zuzugreifen und diese zu verwalten und auch Neon Serverless Postgres zu unterstützen.

Der Verbinder weist die folgenden Merkmale auf.

Einschränkungen

Hier ist ein Beispiel für das Aufrufen von UseVector.

NpgsqlDataSourceBuilder dataSourceBuilder = new("Host=localhost;Port=5432;Username=postgres;Password=example;Database=postgres;");
dataSourceBuilder.UseVector();
NpgsqlDataSource dataSource = dataSourceBuilder.Build();

Bei Verwendung der Registrierungsmethode AddPostgresVectorStore Dependency Injection mit einer Verbindungszeichenfolge wird die Datenquelle von dieser Methode erstellt und UseVector wird automatisch angewendet.

Erste Schritte

Fügen Sie ihrem Projekt das NuGet-Paket des Postgres Vector Store-Connectors hinzu.

dotnet add package Microsoft.SemanticKernel.Connectors.PgVector --prerelease

Sie können den Vektorspeicher mithilfe von Erweiterungsmethoden, die vom Semantischer Kernel bereitgestellt werden, zum Abhängigkeitsinjektionscontainer IServiceCollection hinzufügen.

using Microsoft.Extensions.DependencyInjection;

var kernelBuilder = Kernel.CreateBuilder();
kernelBuilder.Services.AddPostgresVectorStore("<Connection String>");

Dabei ist eine Verbindungszeichenfolge für die Postgres-Instanz, im von Npgsql erwarteten Format, z. B. .

Erweiterungsmethoden, die keine Parameter verwenden, werden ebenfalls bereitgestellt. Dazu muss eine Instanz von NpgsqlDataSource separat beim Abhängigkeitseinfügungscontainer registriert werden. Beachten Sie, dass UseVector auf dem Builder aufgerufen werden muss, um die Vektorunterstützung über pgvector-dotnet zu aktivieren:

using Microsoft.Extensions.DependencyInjection;
using Microsoft.SemanticKernel;

// Using IServiceCollection with ASP.NET Core.
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddPostgresVectorStore("<Connection String>");

using Microsoft.Extensions.DependencyInjection;
using Microsoft.SemanticKernel;
using Npgsql;

// Using IServiceCollection with ASP.NET Core.
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddSingleton<NpgsqlDataSource>(sp => 
{
    NpgsqlDataSourceBuilder dataSourceBuilder = new("<Connection String>");
    dataSourceBuilder.UseVector();
    return dataSourceBuilder.Build();
});
builder.Services.AddPostgresVectorStore();

Sie können eine Postgres Vector Store-Instanz direkt mit einer benutzerdefinierten Datenquelle oder mit einem Verbindungszeichenfolge erstellen.

using Microsoft.SemanticKernel.Connectors.PgVector;
using Npgsql;

NpgsqlDataSourceBuilder dataSourceBuilder = new("<Connection String>");
dataSourceBuilder.UseVector();
NpgsqlDataSource dataSource = dataSourceBuilder.Build();
var vectorStore = new PostgresVectorStore(dataSource, ownsDataSource: true);

using Microsoft.SemanticKernel.Connectors.PgVector;

var connection = new PostgresVectorStore("<Connection String>");

Es ist möglich, einen direkten Verweis auf eine benannte Auflistung mit einer benutzerdefinierten Datenquelle oder mit einem Verbindungszeichenfolge zu erstellen.

using Microsoft.SemanticKernel.Connectors.PgVector;
using Npgsql;

NpgsqlDataSourceBuilder dataSourceBuilder = new("<Connection String>");
dataSourceBuilder.UseVector();
var dataSource = dataSourceBuilder.Build();

var collection = new PostgresCollection<string, Hotel>(dataSource, "skhotels", ownsDataSource: true);

using Microsoft.SemanticKernel.Connectors.PgVector;

var collection = new PostgresCollection<string, Hotel>("<Connection String>", "skhotels");

Datenzuordnung

Der Postgres-Connector stellt beim Zuordnen von Daten aus dem Datenmodell zum Speicher einen Standardzuordnungs-Mapper bereit. Die Standardzuordnung verwendet die Modellanmerkungen oder die Datensatzdefinition, um den Typ jeder Eigenschaft zu bestimmen und das Modell in einem Dictionary abzubilden, das für Postgres serialisiert werden kann.

• Die als Schlüssel gekennzeichnete Datenmodelleigenschaft wird dem PRIMÄRSCHLÜSSEL in der Tabelle Postgres zugeordnet.
• Die Datenmodelleigenschaften, die als Daten gekennzeichnet sind, werden einer Tabellenspalte in Postgres zugeordnet.
• Die Als Vektoren gekennzeichneten Datenmodelleigenschaften werden einer Tabellenspalte zugeordnet, die den pgvector-Typ VECTOR in Postgres aufweist.

Außerkraftsetzung des Eigenschaftsnamens

Sie können Feldnamen überschreiben, um sie im Speicher zu verwenden, die sich von den Eigenschaftsnamen im Datenmodell unterscheiden. Auf diese Weise können Sie Tabellenspaltennamen auch dann abgleichen, wenn sie nicht mit den Eigenschaftennamen im Datenmodell übereinstimmen.

Die Außerkraftsetzung des Eigenschaftsnamens erfolgt durch Festlegen der StorageName Option über die Datenmodellattribute oder datensatzdefinition.

Hier ist ein Beispiel für ein Datenmodell, bei dem StorageName für seine Attribute festgelegt wurde und wie es in Postgres als Tabelle dargestellt wird, vorausgesetzt, der Collection-Name ist Hotels.

using System;
using Microsoft.Extensions.VectorData;

public class Hotel
{
    [VectorStoreKey(StorageName = "hotel_id")]
    public int HotelId { get; set; }

    [VectorStoreData(StorageName = "hotel_name")]
    public string HotelName { get; set; }

    [VectorStoreData(StorageName = "hotel_description")]
    public string Description { get; set; }

    [VectorStoreVector(Dimensions: 4, DistanceFunction = DistanceFunction.CosineDistance, IndexKind = IndexKind.Hnsw, StorageName = "hotel_description_embedding")]
    public ReadOnlyMemory<float>? DescriptionEmbedding { get; set; }
}

CREATE TABLE IF NOT EXISTS public."Hotels" (
    "hotel_id" INTEGER PRIMARY KEY NOT NULL,
    hotel_name TEXT,
    hotel_description TEXT,
    hotel_description_embedding VECTOR(4)
);

Vektorindizierung

Das hotel_description_embedding im oben erwähnten Hotel-Modell ist eine Vektoreigenschaft mit IndexKind.HNSW-Indexierung. Dieser Index wird automatisch erstellt, wenn die Auflistung erstellt wird. HNSW ist der einzige Indextyp, der für die Indexerstellung unterstützt wird. DAS IVFFlat Index Building erfordert, dass daten bereits zur Indexerstellungszeit in der Tabelle vorhanden sind und daher nicht für die Erstellung einer leeren Tabelle geeignet ist. Sie können Indizes für Tabellen außerhalb des Connectors erstellen und ändern, die vom Connector beim Ausführen von Abfragen verwendet werden.

Verwenden mit Entra-Authentifizierung

Azure Database for PostgreSQL bietet die Möglichkeit, mithilfe der Entra-Authentifizierung eine Verbindung mit Ihrer Datenbank herzustellen. Dadurch wird die Notwendigkeit entfernt, einen Benutzernamen und ein Kennwort in Ihrem Verbindungszeichenfolge zu speichern. Um die Entra-Authentifizierung für eine Azure-DB für PostgreSQL-Datenbanken zu verwenden, können Sie die folgende Npgsql-Erweiterungsmethode nutzen und eine Verbindungszeichenfolge festlegen, die weder einen Benutzernamen noch ein Kennwort enthält.

using System.Text;
using System.Text.Json;
using Azure.Core;
using Azure.Identity;
using Npgsql;

namespace Program;

public static class NpgsqlDataSourceBuilderExtensions
{
    private static readonly TokenRequestContext s_azureDBForPostgresTokenRequestContext = new(["https://ossrdbms-aad.database.windows.net/.default"]);

    public static NpgsqlDataSourceBuilder UseEntraAuthentication(this NpgsqlDataSourceBuilder dataSourceBuilder, TokenCredential? credential = default)
    {
        credential ??= new DefaultAzureCredential();

        if (dataSourceBuilder.ConnectionStringBuilder.Username == null)
        {
            var token = credential.GetToken(s_azureDBForPostgresTokenRequestContext, default);
            SetUsernameFromToken(dataSourceBuilder, token.Token);
        }

        SetPasswordProvider(dataSourceBuilder, credential, s_azureDBForPostgresTokenRequestContext);

        return dataSourceBuilder;
    }

    public static async Task<NpgsqlDataSourceBuilder> UseEntraAuthenticationAsync(this NpgsqlDataSourceBuilder dataSourceBuilder, TokenCredential? credential = default, CancellationToken cancellationToken = default)
    {
        credential ??= new DefaultAzureCredential();

        if (dataSourceBuilder.ConnectionStringBuilder.Username == null)
        {
            var token = await credential.GetTokenAsync(s_azureDBForPostgresTokenRequestContext, cancellationToken).ConfigureAwait(false);
            SetUsernameFromToken(dataSourceBuilder, token.Token);
        }

        SetPasswordProvider(dataSourceBuilder, credential, s_azureDBForPostgresTokenRequestContext);

        return dataSourceBuilder;
    }

    private static void SetPasswordProvider(NpgsqlDataSourceBuilder dataSourceBuilder, TokenCredential credential, TokenRequestContext tokenRequestContext)
    {
        dataSourceBuilder.UsePasswordProvider(_ =>
        {
            var token = credential.GetToken(tokenRequestContext, default);
            return token.Token;
        }, async (_, ct) =>
        {
            var token = await credential.GetTokenAsync(tokenRequestContext, ct).ConfigureAwait(false);
            return token.Token;
        });
    }

    private static void SetUsernameFromToken(NpgsqlDataSourceBuilder dataSourceBuilder, string token)
    {
        var username = TryGetUsernameFromToken(token);

        if (username != null)
        {
            dataSourceBuilder.ConnectionStringBuilder.Username = username;
        }
        else
        {
            throw new Exception("Could not determine username from token claims");
        }
    }

    private static string? TryGetUsernameFromToken(string jwtToken)
    {
        // Split the token into its parts (Header, Payload, Signature)
        var tokenParts = jwtToken.Split('.');
        if (tokenParts.Length != 3)
        {
            return null;
        }

        // The payload is the second part, Base64Url encoded
        var payload = tokenParts[1];

        // Add padding if necessary
        payload = AddBase64Padding(payload);

        // Decode the payload from Base64Url
        var decodedBytes = Convert.FromBase64String(payload);
        var decodedPayload = Encoding.UTF8.GetString(decodedBytes);

        // Parse the decoded payload as JSON
        var payloadJson = JsonSerializer.Deserialize<JsonElement>(decodedPayload);

        // Try to get the username from 'upn', 'preferred_username', or 'unique_name' claims
        if (payloadJson.TryGetProperty("upn", out var upn))
        {
            return upn.GetString();
        }
        else if (payloadJson.TryGetProperty("preferred_username", out var preferredUsername))
        {
            return preferredUsername.GetString();
        }
        else if (payloadJson.TryGetProperty("unique_name", out var uniqueName))
        {
            return uniqueName.GetString();
        }

        return null;
    }

    private static string AddBase64Padding(string base64) => (base64.Length % 4) switch
    {
        2 => base64 + "==",
        3 => base64 + "=",
        _ => base64,
    };
}

Jetzt können Sie die methode UseEntraAuthentication verwenden, um die Verbindungszeichenfolge für den Postgres Connector einzurichten:

using Microsoft.SemanticKernel.Connectors.Postgres;

var connectionString = "Host=mydb.postgres.database.azure.com;Port=5432;Database=postgres;SSL Mode=Require;";  // No Username or Password
var dataSourceBuilder = new NpgsqlDataSourceBuilder(connectionString);
dataSourceBuilder.UseEntraAuthentication();
dataSourceBuilder.UseVector();
var dataSource = dataSourceBuilder.Build();

var vectorStore = new PostgresVectorStore(dataSource, ownsDataSource: true);

Standardmäßig verwendet die UseEntraAuthentication-Methode die DefaultAzureCredential zum Authentifizieren mit Azure AD. Sie können bei Bedarf auch eine benutzerdefinierte TokenCredential Implementierung bereitstellen.

Erste Schritte

Installieren Sie den semantischen Kernel mit den Extras der Postgres, die den Postgres-Client enthalten.

pip install semantic-kernel[postgres]

Anschließend können Sie mithilfe der PostgresStore Klasse eine Vektorspeicherinstanz erstellen. Sie können einen psycopg_poolAsyncConnectionPool direkt übergeben oder einen PostgresSettings Verbindungspool aus Umgebungsvariablen erstellen.

from semantic_kernel.connectors.postgres import PostgresStore, PostgresSettings

settings = PostgresSettings()

pool = await settings.create_connection_pool()
async with pool:
    vector_store = PostgresStore(connection_pool=pool)
    ...

Sie können POSTGRES_CONNECTION_STRING in Ihrer Umgebung festlegen oder die Umgebungsvariablen PGHOST, PGPORT, PGUSER, PGPASSWORD, PGDATABASE und optional PGSSLMODE wie für libpq definiert verwenden. Diese Werte werden von der PostgresSettings Klasse zum Erstellen eines Verbindungspools verwendet.

Sie können auch eine Sammlung direkt erstellen. Die Sammlung selbst ist ein Kontext-Manager, sodass Sie sie in einem with Block verwenden können. Wenn Sie keinen Verbindungspool übergeben, erstellt die Sammlung eine mithilfe der PostgresSettings Klasse.

from semantic_kernel.connectors.postgres import PostgresCollection

collection = PostgresCollection(collection_name="skhotels", record_type=Hotel)
async with collection:  # This will create a connection pool using PostgresSettings
    ...

Datenzuordnung

Der Postgres-Connector stellt beim Zuordnen von Daten aus dem Datenmodell zum Speicher einen Standardzuordnungs-Mapper bereit. Die Standardzuordnung verwendet die Modellanmerkungen oder die Datensatzdefinition, um den Typ jeder Eigenschaft zu bestimmen und das Modell in ein dict zuzuordnen, das in Postgres-Zeilen serialisiert werden kann.

• Die als Schlüssel gekennzeichnete Datenmodelleigenschaft wird dem PRIMÄRSCHLÜSSEL in der Tabelle Postgres zugeordnet.
• Die Datenmodelleigenschaften, die als Daten gekennzeichnet sind, werden einer Tabellenspalte in Postgres zugeordnet.
• Die Als Vektoren gekennzeichneten Datenmodelleigenschaften werden einer Tabellenspalte zugeordnet, die den pgvector-Typ VECTOR in Postgres aufweist.

from typing import Annotated
from pydantic import BaseModel

from semantic_kernel.connectors.postgres import PostgresCollection
from semantic_kernel.data.vector import (
    DistanceFunction,
    IndexKind,
    VectorStoreField,
    vectorstoremodel,
)

@vectorstoremodel
class Hotel(BaseModel):
    hotel_id: Annotated[int, VectorStoreField("key")]
    hotel_name: Annotated[str, VectorStoreField("data")]
    hotel_description: Annotated[str, VectorStoreField("data")]
    hotel_description_embedding: Annotated[
        list[float] | None,
        VectorStoreField(
            "vector",
            dimensions=4,
            index_kind=IndexKind.HNSW,
            distance_function=DistanceFunction.COSINE_SIMILARITY,
        ),
    ] = None

collection = PostgresCollection(collection_name="Hotels", record_type=Hotel)

async with collection:
    await collection.ensure_collection_exists()

CREATE TABLE IF NOT EXISTS public."Hotels" (
    "hotel_id" INTEGER NOT NULL,
    "hotel_name" TEXT,
    "hotel_description" TEXT,
    "hotel_description_embedding" VECTOR(4)
    PRIMARY KEY ("hotel_id")
);

Vektorindizierung

Das hotel_description_embedding im oben erwähnten Hotel-Modell ist eine Vektoreigenschaft mit IndexKind.HNSW-Indexierung. Dieser Index wird automatisch erstellt, wenn die Auflistung erstellt wird. HNSW ist der einzige Indextyp, der für die Indexerstellung unterstützt wird. DAS IVFFlat Index Building erfordert, dass daten bereits zur Indexerstellungszeit in der Tabelle vorhanden sind und daher nicht für die Erstellung einer leeren Tabelle geeignet ist. Sie können Indizes für Tabellen außerhalb des Connectors erstellen und ändern, die vom Connector beim Ausführen von Abfragen verwendet werden.

Verwenden mit Entra-Authentifizierung

Azure Database for PostgreSQL bietet die Möglichkeit, mithilfe der Entra-Authentifizierung eine Verbindung mit Ihrer Datenbank herzustellen. Dadurch wird die Notwendigkeit entfernt, einen Benutzernamen und ein Kennwort in Ihrem Verbindungszeichenfolge zu speichern. Um Entra-Authentifizierung für eine Azure PostgreSQL-Datenbank zu verwenden, können Sie die folgende benutzerdefinierte AsyncConnection-Klasse verwenden.

import base64
import json
import logging
from functools import lru_cache

from azure.core.credentials import TokenCredential
from azure.core.credentials_async import AsyncTokenCredential
from azure.identity import DefaultAzureCredential
from psycopg import AsyncConnection

AZURE_DB_FOR_POSTGRES_SCOPE = "https://ossrdbms-aad.database.windows.net/.default"

logger = logging.getLogger(__name__)

async def get_entra_token_async(credential: AsyncTokenCredential) -> str:
    """Get the password from Entra using the provided credential."""
    logger.info("Acquiring Entra token for postgres password")

    async with credential:
        cred = await credential.get_token(AZURE_DB_FOR_POSTGRES_SCOPE)
        return cred.token

def get_entra_token(credential: TokenCredential | None) -> str:
    """Get the password from Entra using the provided credential."""
    logger.info("Acquiring Entra token for postgres password")
    credential = credential or get_default_azure_credentials()

    return credential.get_token(AZURE_DB_FOR_POSTGRES_SCOPE).token

@lru_cache(maxsize=1)
def get_default_azure_credentials() -> DefaultAzureCredential:
    """Get the default Azure credentials.

    This method caches the credentials to avoid creating new instances.
    """
    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)

async def get_entra_conninfo(credential: TokenCredential | AsyncTokenCredential | None) -> dict[str, str]:
    """Fetches a token returns the username and token."""
    # Fetch a new token and extract the username
    if isinstance(credential, AsyncTokenCredential):
        token = await get_entra_token_async(credential)
    else:
        token = get_entra_token(credential)
    claims = decode_jwt(token)
    username = claims.get("upn") or claims.get("preferred_username") or claims.get("unique_name")
    if not username:
        raise ValueError("Could not extract username from token. Have you logged in?")

    return {"user": username, "password": token}

class AsyncEntraConnection(AsyncConnection):
    """Asynchronous connection class for using Entra auth with Azure DB for PostgreSQL."""

    @classmethod
    async def connect(cls, *args, **kwargs):
        """Establish an asynchronous connection using Entra auth with Azure DB for PostgreSQL."""
        credential = kwargs.pop("credential", None)
        if credential and not isinstance(credential, (TokenCredential, AsyncTokenCredential)):
            raise ValueError("credential must be a TokenCredential or AsyncTokenCredential")
        if not kwargs.get("user") or not kwargs.get("password"):
            credential = credential or get_default_azure_credentials()
            entra_conninfo = await get_entra_conninfo(credential)
            if kwargs.get("user"):
                entra_conninfo.pop("user", None)
            kwargs.update(entra_conninfo)
        return await super().connect(*args, **kwargs)

Sie können die benutzerdefinierte Verbindungsklasse mit der PostgresSettings.get_connection_pool Methode verwenden, um einen Verbindungspool zu erstellen.

from semantic_kernel.connectors.postgres import PostgresSettings, PostgresStore


pool = await PostgresSettings().create_connection_pool(connection_class=AsyncEntraConnection)
async with pool:
    vector_store = PostgresStore(connection_pool=pool)
    ...

Standardmäßig verwendet die klasse AsyncEntraConnection die DefaultAzureCredential zum Authentifizieren mit Azure AD. Sie können bei Bedarf auch ein weiteres TokenCredential in den kwargs bereitstellen.

from azure.identity import ManagedIdentityCredential


pool = await PostgresSettings().create_connection_pool(
    connection_class=AsyncEntraConnection, credential=ManagedIdentityCredential()
)
async with pool:
    vector_store = PostgresStore(connection_pool=pool)
    ...

JDBC

Der JDBC-Connector kann verwendet werden, um eine Verbindung mit Postgres herzustellen.