Recherche hybride dans PostgreSQL | Configuration VectorChord

Recherche hybride sur PostgreSQL et VectorChord : Configuration de l'infrastructure

La recherche hybride, qui combine les méthodes vectorielles et à texte intégral, devient la norme pour les systèmes de recherche modernes. Dans cet article, nous plongeons dans la configuration de l'infrastructure basée sur PostgreSQL et VectorChord — une solution qui simplifie la mise en œuvre de la recherche hybride sans nécessiter d'installation manuelle d'extensions ni de configuration complexe.

Configuration de l'environnement : Docker et PostgreSQL

Pour commencer, nous aurons besoin d'un environnement conteneurisé. Nous utiliserons Docker Compose pour déployer PostgreSQL avec les extensions VectorChord préinstallées. Note : dans la configuration, nous ne spécifions pas explicitement les extensions — elles seront ajoutées automatiquement lors de la première connexion via VechordRegistry.

Exemple de fichier docker-compose-dev.yml :

services:
  postgres:
    image: tensorchord/vchord-suite:pg18-latest
    environment:
      POSTGRES_DB: ${DB__NAME}
      POSTGRES_USER: ${DB__USER}
      POSTGRES_PASSWORD: ${DB__PASSWORD}
    volumes:
      - pgdata:/var/lib/postgresql
    ports:
      - "5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U ${DB__USER} -d ${DB__NAME}"]
      interval: 5s
      timeout: 5s
      retries: 5
volumes:
  pgdata:

Point clé : l'image tensorchord/vchord-suite contient déjà tous les composants nécessaires, mais les extensions ne sont activées qu'au premier accès via VechordRegistry. Cela évite la configuration manuelle et garantit la compatibilité des versions.

Conception des tables pour la recherche hybride

La structure de données repose sur trois tables. La table de base BaseTable contient les champs communs :

from datetime import datetime, timezone
from functools import partial
import msgspec
from psycopg.types.json import Jsonb
from vechord import Table

class BaseTable(Table, kw_only=True):
    metadata: Jsonb
    created_at: datetime = msgspec.field(
        default_factory=partial(datetime.now, timezone.utc))
    updated_at: datetime = msgspec.field(
        default_factory=partial(datetime.now, timezone.utc))

Le champ metadata stocke des données arbitraires au format JSONB, et les horodatages sont mis à jour automatiquement lors de la création d'un enregistrement. Note : updated_at n'est pas mis à jour automatiquement lors des modifications d'enregistrement — cela doit être fait manuellement dans les méthodes de mise à jour.

La table Document :

import msgspec
from vechord.spec import PrimaryKeyUUID
from .base import BaseTable

class Document(BaseTable, kw_only=True):
    uid: PrimaryKeyUUID = msgspec.field(default_factory=PrimaryKeyUUID.factory)
    title: str
    text: str

Un UUID est utilisé comme clé primaire. La table Chunk est liée au document via une clé étrangère :

from vechord.spec import ForeignKey, Keyword, PrimaryKeyUUID, Vector

class Chunk(BaseTable, kw_only=True):
    uid: PrimaryKeyUUID = msgspec.field(default_factory=PrimaryKeyUUID.factory)
    doc_id: Annotated[UUID, ForeignKey[Document.uid]]
    content: str
    content_tsv: Keyword
    embedding: DenseVector
    chunk_index: int

Ici, content_tsv est destiné à la recherche en texte intégral (BM25), et embedding stocke la représentation vectorielle du chunk. La clé étrangère doc_id implémente automatiquement la suppression en cascade.

Important : tenter de déplacer uid vers la classe de base provoque une erreur, car ForeignKey commence à chercher le champ dans BaseTable plutôt que dans la table spécifique. Par conséquent, la clé primaire est dupliquée dans chaque table.

VechordRegistry : Gestion centralisée de la base de données

La classe VechordRegistry sert d'interface unique pour travailler avec la base de données. Elle automatise :

• L'installation des extensions (vchord, vchord_bm25, pg_tokenizer)
• La création des tables et des tokenizers
• Les opérations CRUD et de recherche

Initialisation :

from vechord import VechordRegistry
from core import settings
from .document import Document
from .doc_chunk import Chunk

vr = VechordRegistry(
    namespace=settings.db.namespace,
    url=settings.db.db_url,
    tables=[Document, Chunk]
)

Lors de la première utilisation (dans le contexte async with vr :), la méthode init_extension s'exécute, installant les extensions et configurant search_path. Cela élimine le besoin de configuration manuelle et minimise les erreurs de configuration.

Exemple d'utilisation pour insérer un document :

async with vr:
    doc = Document(title='Note', text='Some text', metadata=Jsonb({}))
    await vr.insert(doc)

Après exécution de ce code, toutes les extensions nécessaires, les tables et l'enregistrement apparaîtront dans la base de données.

Pipelines de traitement des données

Pour des opérations complexes comme la création d'un document suivie d'un chunking, on utilise VechordPipeline. Cela permet de combiner plusieurs étapes en un pipeline, où chaque étape enregistre automatiquement les résultats dans la base de données.

Étape 1 : Création du document. Le décorateur @vr.inject(output=Document) garantit la persistance de l'objet :

@vr.inject(output=Document)
async def _create_document(doc_data: DocumentCreate) -> Document:
    doc = Document(
        title=doc_data.title,
        text=doc_data.text,
        metadata=Jsonb(doc_data.metadata)
    )
    return doc

Étape 2 : Génération des chunks. Prend les données de la table Document et retourne une liste de chunks :

@vr.inject(input=Document, output=Chunk)
async def create_chunks(uid: UUID, text: str) -> list[Chunk]:
    chunks = await _chunker.segment(text)
    return [
        Chunk(
            doc_id=uid,
            content=chunk,
            content_tsv=Keyword(chunk),
            embedding=DenseVector(await _embedder.vectorize_chunk(chunk)),
            metadata=Jsonb({}),
            chunk_index=i
        )
        for i, chunk in enumerate(chunks, start=1)
    ]

Assemblage du pipeline :

from db_models import vr
from .utils import create_chunks, create_document

class DocumentService:
    async def create_document(self, doc_data: DocumentCreate):
        pipeline = vr.pipeline(
            _create_document,
            create_chunks
        )
        await pipeline(doc_data)

Cette approche garantit l'atomicité des opérations et minimise les accès à la base de données.

Points clés : Aspects essentiels de l'implémentation

Lors de la mise en œuvre de la recherche hybride basée sur VectorChord, prêtez attention à :

• Initialisation automatique des extensions : VechordRegistry installe tous les composants nécessaires lors de la première connexion, simplifiant le déploiement.
• Structure des tables : séparer les documents et les chunks est crucial pour une recherche efficace ; les clés étrangères garantissent l'intégrité des données.
• Pipelines de traitement : l'utilisation de VechordPipeline encapsule les opérations complexes et assure la cohérence des données.

Ces éléments forment la base d'un système de recherche scalable combinant la précision de la recherche vectorielle et la flexibilité de la recherche en texte intégral.

— Editorial Team