Utilisation du connecteur Postgres Vector Store (préversion)

Aperçu

Le connecteur Postgres Vector Store peut être utilisé pour accéder aux données et les gérer dans Postgres et prend également en charge Neon Serverless Postgres.

Le connecteur présente les caractéristiques suivantes.

Limites

Voici un exemple de comment appeler UseVector.

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

Lorsque vous utilisez la méthode d’inscription d’injection de dépendances AddPostgresVectorStore avec un chaîne de connexion, la source de données sera construite par cette méthode et aura automatiquement UseVector appliquée.

Commencer

Ajoutez le package NuGet du connecteur Postgres Vector Store à votre projet.

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

Vous pouvez ajouter le magasin de vecteurs au conteneur d’injection de dépendances IServiceCollection à l’aide de méthodes d’extension fournies par Noyau sémantique.

using Microsoft.Extensions.DependencyInjection;

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

Où <Connection String> est une chaîne de connexion à l'instance Postgres, au format attendu par Npgsql, par exemple Host=localhost;Port=5432;Database=postgres;Username=postgres;Password=postgres.

Les méthodes d’extension qui ne prennent aucun paramètre sont également fournies. Celles-ci nécessitent qu’une instance de NpgsqlDataSource soit inscrite séparément auprès du conteneur d’injection de dépendances. Notez que UseVector doit être invoqué sur le générateur pour activer la prise en charge des vecteurs via pgvector-dotnet :

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

Vous pouvez construire une instance postgres Vector Store directement avec une source de données personnalisée ou avec un chaîne de connexion.

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

Il est possible de construire une référence directe à une collection nommée avec une source de données personnalisée ou avec un chaîne de connexion.

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

Mappage des données

Le connecteur Postgres fournit un mappeur par défaut lors du mappage des données du modèle de données au stockage. Le mappeur par défaut utilise les annotations de modèle ou la définition d’enregistrement pour déterminer le type de chaque propriété et pour mapper le modèle dans un dictionnaire qui peut être sérialisé sur Postgres.

• La propriété de modèle de données annotée en tant que clé est mappée à la clé PRIMAIRE dans la table Postgres.
• Les propriétés du modèle de données annotées en tant que données sont mappées à une colonne de table dans Postgres.
• Les propriétés du modèle de données annotées en tant que vecteurs sont mappées à une colonne de table qui a le type pgvector VECTOR dans Postgres.

Surcharge du nom de propriété

Vous pouvez fournir des noms de champs personnalisés à utiliser dans le stockage, qui sont différents des noms de propriétés du modèle de données. Cela vous permet de faire correspondre les noms de colonnes de table même s’ils ne correspondent pas aux noms de propriétés sur le modèle de données.

Le remplacement du nom de propriété est effectué en définissant l’option StorageName via les attributs du modèle de données ou la définition d’enregistrement.

Voici un exemple de modèle de données avec StorageName défini sur ses attributs et comment il sera représenté dans Postgres en tant que table, en supposant que le nom de la collection est 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)
);

Indexation de vecteurs

Le hotel_description_embedding modèle ci-dessus est une propriété vectorielle avec une indexation Hotel. Cet index est créé automatiquement lors de la création de la collection. HNSW est le seul type d’index pris en charge pour la création d’index. La génération d’index IVFFlat nécessite que les données existent déjà dans la table au moment de la création de l’index, et il n’est donc pas approprié pour la création d’une table vide. Vous êtes libre de créer et de modifier des index sur des tables en dehors du connecteur, qui seront utilisés par le connecteur lors de l’exécution de requêtes.

Utilisation de l’authentification Entra

Azure Database pour PostgreSQL permet de se connecter à votre base de données à l’aide de l’authentification Entra. Cela supprime la nécessité de stocker un nom d’utilisateur et un mot de passe dans votre chaîne de connexion. Pour utiliser l’authentification Entra pour une base de données Azure DB pour PostgreSQL, vous pouvez utiliser la méthode d’extension Npgsql suivante et définir un chaîne de connexion qui n’a pas de nom d’utilisateur ou de mot de passe :

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

Vous pouvez maintenant utiliser la méthode UseEntraAuthentication pour configurer l’chaîne de connexion pour le connecteur Postgres :

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

Par défaut, la méthode UseEntraAuthentication utilise le DefaultAzureCredential pour s’authentifier auprès de Azure AD. Vous pouvez également fournir une implémentation personnalisée TokenCredential si nécessaire.

Commencer

Installez le noyau sémantique avec les extras Postgres, qui inclut le client Postgres.

pip install semantic-kernel[postgres]

Vous pouvez ensuite créer une instance de magasin de vecteurs à l’aide de la PostgresStore classe. Vous pouvez passer un psycopg_poolAsyncConnectionPool directement ou utiliser celui-ci PostgresSettings pour créer un pool de connexions à partir de variables d’environnement.

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

Vous pouvez définir POSTGRES_CONNECTION_STRING dans votre environnement ou les variables PGHOSTd’environnement , , PGPORT, PGUSERPGPASSWORD, , PGDATABASEet éventuellement PGSSLMODE comme défini pour libpq. Ces valeurs seront utilisées par la PostgresSettings classe pour créer un pool de connexions.

Vous pouvez également créer une collection directement. La collection elle-même est un gestionnaire de contexte. Vous pouvez donc l’utiliser dans un with bloc. Si vous ne passez pas dans un pool de connexions, la collection en crée un en utilisant la classe PostgresSettings.

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

Mappage des données

Le connecteur Postgres fournit un mappeur par défaut lors du mappage des données du modèle de données au stockage. Le mappeur par défaut utilise les annotations de modèle ou la définition d’enregistrement pour déterminer le type de chaque propriété et pour mapper le modèle dans un dict modèle qui peut être sérialisé sur des lignes Postgres.

• La propriété de modèle de données annotée en tant que clé est mappée à la clé PRIMAIRE dans la table Postgres.
• Les propriétés du modèle de données annotées en tant que données sont mappées à une colonne de table dans Postgres.
• Les propriétés du modèle de données annotées en tant que vecteurs sont mappées à une colonne de table qui a le type pgvector VECTOR dans 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")
);

Indexation de vecteurs

Le hotel_description_embedding modèle ci-dessus est une propriété vectorielle avec une indexation Hotel. Cet index est créé automatiquement lors de la création de la collection. HNSW est le seul type d’index pris en charge pour la création d’index. La génération d’index IVFFlat nécessite que les données existent déjà dans la table au moment de la création de l’index, et il n’est donc pas approprié pour la création d’une table vide. Vous êtes libre de créer et de modifier des index sur des tables en dehors du connecteur, qui seront utilisés par le connecteur lors de l’exécution de requêtes.

Utilisation de l’authentification Entra

Azure Database pour PostgreSQL permet de se connecter à votre base de données à l’aide de l’authentification Entra. Cela supprime la nécessité de stocker un nom d’utilisateur et un mot de passe dans votre chaîne de connexion. Pour utiliser l’authentification Entra pour une base de données Azure DB pour PostgreSQL, vous pouvez utiliser la classe AsyncConnection personnalisée suivante :

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)

Vous pouvez utiliser la classe de connexion personnalisée avec la PostgresSettings.get_connection_pool méthode pour créer un pool de connexions.

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

Par défaut, la classe AsyncEntraConnection utilise la DefaultAzureCredential pour s’authentifier auprès de Azure AD. Vous pouvez également fournir une autre TokenCredential, si nécessaire, dans 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

Le connecteur JDBC peut être utilisé pour se connecter à Postgres.