De Postgres Vector Store-connector gebruiken (preview)

Overzicht

De Postgres Vector Store-connector kan worden gebruikt voor toegang tot en beheer van gegevens in Postgres en biedt ook ondersteuning voor Neon Serverless Postgres.

De verbindingslijn heeft de volgende kenmerken.

Beperkingen

Hier volgt een voorbeeld van hoe u aanroept UseVector.

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

Wanneer u de registratiemethode AddPostgresVectorStore afhankelijkheidsinjectie gebruikt met een verbindingsreeks, wordt de gegevensbron met deze methode samengesteld en wordt automatisch UseVector toegepast.

Aan de slag

Voeg het NuGet-pakket van de Postgres Vector Store-connector toe aan uw project.

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

U kunt het vectorarchief toevoegen aan de IServiceCollection afhankelijkheidsinjectiecontainer met behulp van extensiemethoden die worden geleverd door Semantic Kernel.

using Microsoft.Extensions.DependencyInjection;

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

Waarbij <Connection String> een verbindingsreeks is voor de Postgres-instantie, in de indeling die Npgsql verwacht, bijvoorbeeld Host=localhost;Port=5432;Database=postgres;Username=postgres;Password=postgres.

Extensiemethoden waarvoor geen parameters worden gebruikt, worden ook opgegeven. Hiervoor moet een exemplaar van NpgsqlDataSource afzonderlijk worden geregistreerd bij de afhankelijkheidsinjectiecontainer. Houd er rekening mee dat UseVector moet worden aangeroepen op de builder om vectorondersteuning via pgvector-dotnet in te schakelen.

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();

U kunt een Postgres Vector Store-exemplaar rechtstreeks maken met een aangepaste gegevensbron of met een verbindingsreeks.

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>");

Het is mogelijk om een directe verwijzing te maken naar een benoemde verzameling met een aangepaste gegevensbron of met een verbindingsreeks.

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");

Datamapping

De Postgres-connector biedt een standaardtoewijzingsfunctie bij het toewijzen van gegevens uit het gegevensmodel aan opslag. De standaard mapper gebruikt de modelaantekeningen of recorddefinitie om het type van elke eigenschap te bepalen en het model toe te wijzen aan een woordenlijst die kan worden geserialiseerd naar Postgres.

• De eigenschap van het gegevensmodel die is geannoteerd als een sleutel, wordt toegewezen aan de PRIMAIRE SLEUTEL in de Postgres-tabel.
• De eigenschappen van het gegevensmodel die zijn geannoteerd als gegevens, worden toegewezen aan een tabelkolom in Postgres.
• De eigenschappen van het gegevensmodel die zijn geannoteerd als vectoren, worden toegewezen aan een tabelkolom met het pgvectortype VECTOR in Postgres.

Eigenschapsnaam vervangen

U kunt overschrijf-veldnamen opgeven voor gebruik in opslag die afwijken van de attributenamen in het gegevensmodel. Hiermee kunt u kolomnamen van tabellen vergelijken, zelfs als ze niet overeenkomen met de eigenschapsnamen in het gegevensmodel.

De naam van de eigenschap kan worden overschreven door de StorageName optie in te stellen in de kenmerken van het gegevensmodel of de definitie van de record.

Hier volgt een voorbeeld van een gegevensmodel dat StorageName is ingesteld op de kenmerken en hoe het in Postgres wordt weergegeven als een tabel, ervan uitgaande dat de naam van de verzameling is 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)
);

Vectorindexering

Het hotel_description_embedding bovenstaande Hotel model is een vectoreigenschap met IndexKind.HNSW indexering. Deze index wordt automatisch gemaakt wanneer de verzameling wordt gemaakt. HNSW is het enige indextype dat wordt ondersteund voor het maken van een index. IvFFlat-indexbouw vereist dat er al gegevens in de tabel aanwezig zijn tijdens het maken van de index, en daarom is het niet geschikt voor het maken van een lege tabel. U bent vrij om indexen te maken en te wijzigen op tabellen buiten de connector, die de connector zal gebruiken bij het uitvoeren van query's.

Gebruiken met Entra-verificatie

Azure Database for PostgreSQL biedt de mogelijkheid om verbinding te maken met uw database met behulp van Entra-verificatie. Hiermee hoeft u geen gebruikersnaam en wachtwoord op te slaan in uw verbindingsreeks. Als u Entra-verificatie wilt gebruiken voor een Azure DB for PostgreSQL-database, kunt u de volgende Npgsql-extensiemethode gebruiken en een verbindingsreeks instellen die geen gebruikersnaam of wachtwoord heeft:

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,
    };
}

U kunt nu de methode UseEntraAuthentication gebruiken om de verbindingsreeks voor de Postgres-connector in te stellen:

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);

De methode UseEntraAuthentication maakt standaard gebruik van de methode DefaultAzureCredential voor verificatie met Azure AD. U kunt indien nodig ook een aangepaste TokenCredential implementatie bieden.

Aan de slag

Installeer semantische kernel met de Postgres-extra's, waaronder de Postgres-client.

pip install semantic-kernel[postgres]

Vervolgens kunt u een vector store-exemplaar maken met behulp van de PostgresStore klasse. U kunt een psycopg_poolAsyncConnectionPool rechtstreeks doorgeven of de PostgresSettings groep gebruiken om een verbindingsgroep te maken op basis van omgevingsvariabelen.

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)
    ...

U kunt instellen POSTGRES_CONNECTION_STRING in uw omgeving, of de omgevingsvariabelen PGHOST, , PGPORT, PGUSER, PGPASSWORDen PGDATABASEeventueel PGSSLMODE zoals gedefinieerd voor libpq. Deze waarden worden door de PostgresSettings klasse gebruikt om een verbindingsgroep te maken.

U kunt ook rechtstreeks een verzameling maken. De verzameling zelf is een contextbeheerder, zodat u deze in een with blok kunt gebruiken. Als u geen connection pool doorgeeft, maakt de collectie zelf een aan met behulp van de 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
    ...

Datamapping

De Postgres-connector biedt een standaardtoewijzingsfunctie bij het toewijzen van gegevens uit het gegevensmodel aan opslag. De standaard mapper gebruikt de modelaantekeningen of recorddefinitie om het type van elke eigenschap te bepalen en om het model toe te wijzen aan een dict die kan worden geserialiseerd naar Postgres-rijen.

• De eigenschap van het gegevensmodel die is geannoteerd als een sleutel, wordt toegewezen aan de PRIMAIRE SLEUTEL in de Postgres-tabel.
• De eigenschappen van het gegevensmodel die zijn geannoteerd als gegevens, worden toegewezen aan een tabelkolom in Postgres.
• De eigenschappen van het gegevensmodel die zijn geannoteerd als vectoren, worden toegewezen aan een tabelkolom met het pgvectortype VECTOR in Postgres.

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")
);

Vectorindexering

Het hotel_description_embedding bovenstaande Hotel model is een vectoreigenschap met IndexKind.HNSW indexering. Deze index wordt automatisch gemaakt wanneer de verzameling wordt gemaakt. HNSW is het enige indextype dat wordt ondersteund voor het maken van een index. IvFFlat-indexbouw vereist dat er al gegevens in de tabel aanwezig zijn tijdens het maken van de index, en daarom is het niet geschikt voor het maken van een lege tabel. U bent vrij om indexen te maken en te wijzigen op tabellen buiten de connector, die de connector zal gebruiken bij het uitvoeren van query's.

Gebruiken met Entra-verificatie

Azure Database for PostgreSQL biedt de mogelijkheid om verbinding te maken met uw database met behulp van Entra-verificatie. Hiermee hoeft u geen gebruikersnaam en wachtwoord op te slaan in uw verbindingsreeks. Als u Entra-verificatie wilt gebruiken voor een Azure DB for PostgreSQL-database, kunt u de volgende aangepaste AsyncConnection-klasse gebruiken:

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)

U kunt de aangepaste verbindingsklasse gebruiken met de PostgresSettings.get_connection_pool methode om een verbindingsgroep te maken.

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)
    ...

De klasse AsyncEntraConnection maakt standaard gebruik van de klasse DefaultAzureCredential voor verificatie met Azure AD. U kunt indien nodig ook een andere TokenCredential opgeven in de kwargs:

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

De JDBC-connector kan worden gebruikt om verbinding te maken met Postgres.