Aller au contenu principal

Fine-tuning d'embeddings

Objectifs

  • Comprendre quand le fine-tuning d’embeddings est nécessaire
  • Préparer un dataset de paires pour l’entraînement
  • Implémenter le fine-tuning avec des adaptateurs légers

Quand fine-tuner ?

Les modèles d’embedding pré-entraînés fonctionnent bien pour la majorité des cas. Le fine-tuning est utile quand :

  • Votre domaine utilise un vocabulaire très spécialisé (médecine, droit, industrie)
  • Les performances de recherche stagnent malgré l’optimisation du pipeline
  • Vous avez des paires de textes annotées (similaires/dissimilaires)

Quand ne PAS fine-tuner

  • Vous n’avez pas assez de données annotées (< 1 000 paires)
  • Les performances sont déjà satisfaisantes
  • Vous pouvez résoudre le problème en améliorant le chunking ou le prompt

Approche 1 : adaptateur linéaire (Matryoshka)

Entraînez une couche de projection pour adapter l’espace d’embedding à votre domaine :

import numpy as np
from openai import OpenAI
from sklearn.model_selection import train_test_split

client = OpenAI()

# Dataset de paires (query, document_positif, document_négatif)
triplets = [
    {
        "query": "effets secondaires du paracétamol",
        "positif": "Le paracétamol peut causer des troubles hépatiques...",
        "negatif": "Le paracétamol est disponible en pharmacie..."
    },
    {
        "query": "posologie aspirine enfant",
        "positif": "Pour les enfants de 6 à 12 ans, la dose recommandée...",
        "negatif": "L'aspirine a été découverte au XIXe siècle..."
    },
    # ... plus de triplets
]

def preparer_embeddings(triplets: list[dict]) -> dict:
    """Génère les embeddings pour les triplets."""
    tous_textes = []
    for t in triplets:
        tous_textes.extend([t["query"], t["positif"], t["negatif"]])

    response = client.embeddings.create(
        input=tous_textes,
        model="text-embedding-3-large"
    )
    embs = np.array(
        [d.embedding for d in sorted(response.data, key=lambda x: x.index)]
    )

    n = len(triplets)
    return {
        "queries": embs[0::3],
        "positifs": embs[1::3],
        "negatifs": embs[2::3],
    }

Entraîner l’adaptateur

import torch
import torch.nn as nn
import torch.optim as optim

class AdaptateurEmbedding(nn.Module):
    """Couche de projection pour adapter les embeddings."""
    def __init__(self, dim_input: int = 3072, dim_output: int = 256):
        super().__init__()
        self.projection = nn.Sequential(
            nn.Linear(dim_input, dim_output),
            nn.ReLU(),
            nn.Linear(dim_output, dim_output),
        )

    def forward(self, x):
        projected = self.projection(x)
        return projected / projected.norm(dim=-1, keepdim=True)

def triplet_loss(anchor, positive, negative, margin=0.2):
    """Triplet margin loss."""
    dist_pos = 1 - torch.cosine_similarity(anchor, positive)
    dist_neg = 1 - torch.cosine_similarity(anchor, negative)
    return torch.relu(dist_pos - dist_neg + margin).mean()

def entrainer_adaptateur(data: dict, epochs: int = 50):
    """Entraîne l'adaptateur sur les triplets."""
    model = AdaptateurEmbedding()
    optimizer = optim.Adam(model.parameters(), lr=1e-3)

    queries = torch.tensor(data["queries"], dtype=torch.float32)
    positifs = torch.tensor(data["positifs"], dtype=torch.float32)
    negatifs = torch.tensor(data["negatifs"], dtype=torch.float32)

    for epoch in range(epochs):
        optimizer.zero_grad()

        q_proj = model(queries)
        p_proj = model(positifs)
        n_proj = model(negatifs)

        loss = triplet_loss(q_proj, p_proj, n_proj)
        loss.backward()
        optimizer.step()

        if (epoch + 1) % 10 == 0:
            print(f"Epoch {epoch + 1}/{epochs}, Loss: {loss.item():.4f}")

    return model

# Entraîner
data = preparer_embeddings(triplets)
adaptateur = entrainer_adaptateur(data)

# Sauvegarder
torch.save(adaptateur.state_dict(), "adaptateur_medical.pt")

Approche 2 : fine-tuning avec Sentence Transformers

Pour un fine-tuning plus poussé, utilisez la bibliothèque Sentence Transformers :

from sentence_transformers import (
    SentenceTransformer, InputExample, losses
)
from torch.utils.data import DataLoader

# Charger un modèle de base
model = SentenceTransformer("BAAI/bge-large-en-v1.5")

# Préparer les exemples d'entraînement
train_examples = [
    InputExample(
        texts=["effets secondaires paracétamol",
               "Le paracétamol peut causer des troubles hépatiques"],
        label=1.0
    ),
    InputExample(
        texts=["effets secondaires paracétamol",
               "Le paracétamol est vendu en pharmacie"],
        label=0.2
    ),
]

train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
train_loss = losses.CosineSimilarityLoss(model)

# Entraîner
model.fit(
    train_objectives=[(train_dataloader, train_loss)],
    epochs=3,
    warmup_steps=100,
    output_path="./modele_medical_finetuned"
)

# Utiliser le modèle fine-tuné
embeddings = model.encode([
    "douleurs abdominales après prise de médicament",
    "effets indésirables médicamenteux"
])

Préparer un dataset de qualité

Générer des paires avec un LLM

def generer_paires_entrainement(
    documents: list[str],
    n_par_doc: int = 3
) -> list[dict]:
    """Génère des paires query-document pour l'entraînement."""
    paires = []

    for doc in documents:
        response = client.chat.completions.create(
            model="gpt-5.3",
            messages=[{
                "role": "user",
                "content": (
                    f"Voici un document :\n\n{doc[:500]}\n\n"
                    f"Génère {n_par_doc} questions auxquelles ce document "
                    f"répond précisément. Une question par ligne."
                )
            }],
            temperature=0.7
        )
        questions = response.choices[0].message.content.strip().split("\n")

        for q in questions[:n_par_doc]:
            q = q.strip().lstrip("0123456789.-) ")
            if q:
                paires.append({"query": q, "document": doc})

    return paires

Évaluer le fine-tuning

def evaluer_avant_apres(
    queries: list[str],
    docs_pertinents: list[list[str]],
    corpus: list[str],
    adaptateur=None
):
    """Compare les performances avant et après fine-tuning."""
    resp = client.embeddings.create(
        input=corpus, model="text-embedding-3-large"
    )
    corpus_embs = np.array(
        [d.embedding for d in sorted(resp.data, key=lambda x: x.index)]
    )

    if adaptateur:
        corpus_embs_adapted = adaptateur(
            torch.tensor(corpus_embs, dtype=torch.float32)
        ).detach().numpy()
    else:
        corpus_embs_adapted = corpus_embs

    recall_scores = []
    for query, pertinents in zip(queries, docs_pertinents):
        q_resp = client.embeddings.create(
            input=query, model="text-embedding-3-large"
        )
        q_emb = np.array(q_resp.data[0].embedding)

        if adaptateur:
            q_emb = adaptateur(
                torch.tensor(q_emb, dtype=torch.float32).unsqueeze(0)
            ).detach().numpy()[0]

        scores = corpus_embs_adapted @ q_emb
        top_5 = set(np.argsort(scores)[-5:])
        pertinent_ids = set(
            i for i, d in enumerate(corpus) if d in pertinents
        )
        recall = len(top_5 & pertinent_ids) / len(pertinent_ids)
        recall_scores.append(recall)

    return np.mean(recall_scores)

Résumé

  • Le fine-tuning d’embeddings améliore les performances sur des domaines spécialisés
  • Un adaptateur linéaire est la méthode la plus simple et souvent suffisante
  • Sentence Transformers permet un fine-tuning plus poussé du modèle complet
  • Un minimum de 1 000 paires annotées est recommandé
  • Évaluez toujours avant et après pour valider le gain
← Lecon precedenteDétection d'anomalies et de duplicatesLecon suivante →Production : scaling et caching