Dify 知识库向量化与相似度检索优化 — Guide Complet 2026

Comprendre la vectorisation dans Dify

Qu'est-ce que la vectorisation sémantique ?

# Exemple concret de transformation vectorielle
texte_original = "Comment retourner mes chaussures Nike ?"
embedding_resultat = [
    0.0231, -0.0892, 0.1523, 0.0847, ...,  # 1536 dimensions
    0.0912, -0.0234, 0.0765, 0.0123
]

Distance cosinus entre deux phrases similaires :
"Retourner un article" vs "Rembourser ma commande"
Distance : 0.023 (très proche sémantiquement)

Pourquoi Dify a besoin d'optimisation ?

• Le chunking strategy — comment diviser vos documents
• Le modèle d'embedding — qualité vs vitesse
• La métrique de similarité — cosine vs euclidienne vs dot product
• La stratégie de ré-ranking — affiner les résultats

Architecture optimisée avec HolySheep AI

Pourquoi HolySheep AI ?

• Latence moyenne : 47ms (vs 180ms sur OpenAI)
• Prix imbattables : DeepSeek V3.2 à $0.42/M tokens (vs $0.42 chez d'autres)
• Paiement WeChat/Alipay — idéal pour les marchés asiatiques
• Crédits gratuits pour tester avant de s'engager

# Configuration HolySheep AI pour Dify
import openai

IMPORTANT : Utiliser HolySheep AI comme endpoint
openai.api_base = "https://api.holysheep.ai/v1"
openai.api_key = "YOUR_HOLYSHEEP_API_KEY"  # Remplacez par votre clé

Modèle recommandé pour embeddings français
embedding_model = "text-embedding-3-large"

def generer_embedding(texte):
    """Génère un embedding sémantique optimisé"""
    response = openai.Embedding.create(
        model=embedding_model,
        input=texte,
        encoding_format="float"
    )
    return response.data[0].embedding

Test de performance
import time
start = time.time()
result = generer_embedding("Comment optimiser ma recherche vectorielle ?")
latence = (time.time() - start) * 1000
print(f"Latence HolySheep : {latence:.2f}ms")

Implémentation pas-à-pas

Étape 1 : Configuration du client Dify

# dify_config.py - Configuration centralisée
import os
from openai import OpenAI

class DifyVectorConfig:
    """Configuration optimisée pour Dify Knowledge Base"""
    
    # Endpoint HolySheep AI - NE JAMAIS utiliser api.openai.com
    API_BASE = "https://api.holysheep.ai/v1"
    API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
    
    # Modèles d'embedding
    EMBEDDING_MODEL = "text-embedding-3-large"
    EMBEDDING_DIMENSIONS = 3072  # Haute précision
    
    # Modèle de génération (pour comparaison)
    LLM_MODEL = "gpt-4.1"  # $8/M tokens sur HolySheep
    
    @classmethod
    def get_client(cls):
        """Initialise le client OpenAI compatible"""
        return OpenAI(
            api_key=cls.API_KEY,
            base_url=cls.API_BASE,
            timeout=30.0,  # Timeout réduit pour performance
            max_retries=3
        )

Validation de la configuration
def verify_connection():
    client = DifyVectorConfig.get_client()
    try:
        models = client.models.list()
        print("✅ Connexion HolySheep AI réussie")
        print(f"📡 Latence optimale : <50ms")
        return True
    except Exception as e:
        print(f"❌ Erreur de connexion : {e}")
        return False

Étape 2 : Stratégie de chunking optimisée

# chunking_strategy.py - Optimisation du chunking
import re
from typing import List, Dict

class SmartChunker:
    """
    Stratégie de chunking optimisée pour le français.
    Combine découpage sémantique et gestion des token limits.
    """
    
    def __init__(
        self,
        chunk_size: int = 512,      # Tokens par chunk
        chunk_overlap: int = 128,   # Chevauchement pour contexte
        min_chunk_length: int = 50   # Minimum de caractères
    ):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.min_chunk_length = min_chunk_length
    
    def chunk_document(self, document: str, metadata: Dict) -> List[Dict]:
        """Découpe un document en chunks sémantiques"""
        
        # Étape 1 : Segmentation par paragraphes
        paragraphs = self._split_by_paragraphs(document)
        
        # Étape 2 : Fusion des petits paragraphs
        merged_chunks = self._merge_small_chunks(paragraphs)
        
        # Étape 3 : Découpage final par taille
        final_chunks = []
        for chunk in merged_chunks:
            if len(chunk) > self.chunk_size * 4:  # Approximation
                final_chunks.extend(
                    self._split_by_token_limit(chunk)
                )
            else:
                final_chunks.append(chunk)
        
        # Étape 4 : Ajout des métadonnées
        return [
            {
                "content": chunk,
                "metadata": {
                    **metadata,
                    "chunk_index": i,
                    "total_chunks": len(final_chunks),
                    "source": metadata.get("title", "document")
                }
            }
            for i, chunk in enumerate(final_chunks)
        ]
    
    def _split_by_paragraphs(self, text: str) -> List[str]:
        """Découpe en préservant la structure sémantique"""
        # Conserve les sauts de ligne comme délimiteurs naturels
        return [p.strip() for p in re.split(r'\n\n+', text) if p.strip()]
    
    def _merge_small_chunks(self, paragraphs: List[str]) -> List[str]:
        """Fusionne les petits paragraphes pour éviter des chunks vides"""
        merged = []
        current = ""
        
        for para in paragraphs:
            if len(current) + len(para) < self.chunk_size * 4:
                current += "\n\n" + para
            else:
                if current:
                    merged.append(current)
                current = para
        
        if current:
            merged.append(current)
        return merged
    
    def _split_by_token_limit(self, text: str) -> List[str]:
        """Découpage dur par limite de tokens"""
        # Estimation : 1 token ≈ 4 caractères en français
        estimated_tokens = len(text) // 4
        chunks = []
        
        if estimated_tokens <= self.chunk_size:
            return [text]
        
        # Découpage intelligent
        words = text.split()
        current_chunk = []
        current_tokens = 0
        
        for word in words:
            word_tokens = len(word) // 4 + 1
            if current_tokens + word_tokens > self.chunk_size:
                chunks.append(" ".join(current_chunk))
                # Préserver le contexte avec overlap
                overlap_words = current_chunk[-self.chunk_overlap//5:]
                current_chunk = overlap_words + [word]
                current_tokens = sum(len(w)//4 + 1 for w in current_chunk)
            else:
                current_chunk.append(word)
                current_tokens += word_tokens
        
        if current_chunk:
            chunks.append(" ".join(current_chunk))
        
        return chunks

Utilisation
chunker = SmartChunker(chunk_size=512, chunk_overlap=128)
document_test = """
L'intelligence artificielle transforme radicalement le secteur e-commerce.
Grâce aux embeddings sémantiques, les moteurs de recherche comprennent
désormais le sens des requêtes, pas seulement les mots-clés.

Cette technologie permet de :
- Renvoyer des résultats plus pertinents
- Comprendre les intentions d'achat
- Personnaliser l'expérience utilisateur
"""

chunks = chunker.chunk_document(document_test, {"title": "IA E-commerce"})
print(f"📦 {len(chunks)} chunks générés")

Étape 3 : Pipeline de vectorisation complet

# vectorization_pipeline.py - Pipeline complet
import json
import time
from typing import List, Dict, Optional
from openai import OpenAI
from dify_config import DifyVectorConfig

class VectorizationPipeline:
    """
    Pipeline complet de vectorisation avec HolySheep AI.
    Inclut batching intelligent, retry automatique et monitoring.
    """
    
    def __init__(
        self,
        batch_size: int = 100,
        max_retries: int = 3,
        retry_delay: float = 1.0
    ):
        self.client = DifyVectorConfig.get_client()
        self.batch_size = batch_size
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        self.stats = {
            "total_chunks": 0,
            "successful": 0,
            "failed": 0,
            "total_time_ms": 0
        }
    
    def vectorize_batch(
        self,
        texts: List[str],
        show_progress: bool = True
    ) -> List[Optional[List[float]]]:
        """Vectorise un lot de textes avec gestion d'erreurs"""
        
        all_embeddings = []
        total_batches = (len(texts) + self.batch_size - 1) // self.batch_size
        
        start_time = time.time()
        
        for i in range(0, len(texts), self.batch_size):
            batch = texts[i:i + self.batch_size]
            batch_num = i // self.batch_size + 1
            
            try:
                # Appel API avec retry
                response = self._call_embedding_api(batch)
                embeddings = [item["embedding"] for item in response["data"]]
                all_embeddings.extend(embeddings)
                self.stats["successful"] += len(batch)
                
                if show_progress:
                    progress = (batch_num / total_batches) * 100
                    print(f"📊 Batch {batch_num}/{total_batches} - {progress:.1f}%")
                    
            except Exception as e:
                print(f"⚠️ Erreur batch {batch_num}: {e}")
                # Fallback : embeddings null pour ce batch
                all_embeddings.extend([None] * len(batch))
                self.stats["failed"] += len(batch)
        
        self.stats["total_time_ms"] = (time.time() - start_time) * 1000
        return all_embeddings
    
    def _call_embedding_api(self, texts: List[str]) -> Dict:
        """Appel API avec retry automatique"""
        
        for attempt in range(self.max_retries):
            try:
                response = self.client.embeddings.create(
                    model=DifyVectorConfig.EMBEDDING_MODEL,
                    input=texts,
                    encoding_format="float"
                )
                
                return {
                    "data": [
                        {"embedding": item.embedding}
                        for item in response.data
                    ]
                }
                
            except Exception as e:
                if attempt < self.max_retries - 1:
                    time.sleep(self.retry_delay * (attempt + 1))
                    continue
                raise e
    
    def vectorize_knowledge_base(
        self,
        documents: List[Dict]
    ) -> List[Dict]:
        """
        Vectorise une base de connaissances complète.
        Retourne les documents enrichis avec leurs embeddings.
        """
        
        print(f"🚀 Démarrage vectorisation de {len(documents)} documents")
        
        # Extraction des textes
        texts = [doc["content"] for doc in documents]
        
        # Vectorisation
        embeddings = self.vectorize_batch(texts)
        
        # Enrichissement des documents
        vectorized_docs = []
        for doc, embedding in zip(documents, embeddings):
            doc["embedding"] = embedding
            doc["embedding_model"] = DifyVectorConfig.EMBEDDING_MODEL
            doc["vector_dimension"] = len(embedding) if embedding else None
            vectorized_docs.append(doc)
        
        # Stats finales
        print("\n📈 Statistiques de vectorisation :")
        print(f"   - Total documents : {self.stats['total_chunks']}")
        print(f"   - Succès : {self.stats['successful']}")
        print(f"   - Échecs : {self.stats['failed']}")
        print(f"   - Temps total : {self.stats['total_time_ms']:.2f}ms")
        print(f"   - Latence moyenne : {self.stats['total_time_ms']/len(texts):.2f}ms/document")
        
        return vectorized_docs

Exemple d'utilisation
pipeline = VectorizationPipeline(batch_size=50)

documents_test = [
    {"content": "Comment retourner un article Nike ?", "id": "1", "category": "retours"},
    {"content": "Politique de remboursement sous 30 jours", "id": "2", "category": "politique"},
    {"content": "Livraison gratuite pour commandes > 50€", "id": "3", "category": "livraison"},
]

resultats = pipeline.vectorize_knowledge_base(documents_test)
print(f"✅ {len(resultats)} documents vectorisés")

Optimisation de la recherche de similarité

Choix de la métrique de similarité

# similarity_search.py - Moteur de recherche optimisé
import numpy as np
from typing import List, Tuple, Dict
from sklearn.metrics.pairwise import cosine_similarity

class SimilaritySearchEngine:
    """
    Moteur de recherche sémantique optimisé.
    Supporte multiple métriques de similarité.
    """
    
    METRICS = {
        "cosine": cosine_similarity,
        "euclidean": lambda a, b: -np.linalg.norm(a - b, axis=1),
        "dot": lambda a, b: np.dot(a, b.T)
    }
    
    def __init__(
        self,
        documents: List[Dict],
        metric: str = "cosine",
        top_k: int = 5
    ):
        self.documents = documents
        self.metric = metric
        self.top_k = top_k
        
        # Construction de l'index
        self.embeddings_matrix = np.array([
            doc["embedding"] for doc in documents if doc.get("embedding")
        ])
        self.valid_indices = [
            i for i, doc in enumerate(documents)
            if doc.get("embedding")
        ]
    
    def search(
        self,
        query_embedding: List[float],
        threshold: float = 0.7
    ) -> List[Dict]:
        """
        Recherche les documents les plus similaires.
        
        Args:
            query_embedding: Vecteur de la requête
            threshold: Score minimum de similarité (0-1)
        
        Returns:
            Liste des documents triés par similarité
        """
        
        if not len(self.embeddings_matrix):
            return []
        
        query_vector = np.array(query_embedding).reshape(1, -1)
        
        # Calcul des similarités
        similarity_func = self.METRICS[self.metric]
        scores = similarity_func(query_vector, self.embeddings_matrix)[0]
        
        # Tri par score décroissant
        sorted_indices = np.argsort(scores)[::-1]
        
        # Filtrage et construction des résultats
        results = []
        for idx in sorted_indices[:self.top_k]:
            score = float(scores[idx])
            if score >= threshold:
                doc = self.documents[self.valid_indices[idx]].copy()
                doc["similarity_score"] = score
                doc["rank"] = len(results) + 1
                results.append(doc)
        
        return results
    
    def hybrid_search(
        self,
        query: str,
        query_embedding: List[float],
        keyword_weights: Dict[str, float],
        semantic_weight: float = 0.7
    ) -> List[Dict]:
        """
        Recherche hybride combinant mots-clés et sémantique.
        """
        
        # Recherche sémantique pure
        semantic_results = self.search(query_embedding)
        
        # Recherche par mots-clés (simplifié)
        keyword_results = self._keyword_search(query, keyword_weights)
        
        # Fusion des résultats avec pondération
        return self._merge_results(
            semantic_results,
            keyword_results,
            semantic_weight
        )
    
    def _keyword_search(
        self,
        query: str,
        weights: Dict[str, float]
    ) -> List[Dict]:
        """Recherche par mots-clés (implémentation basique)"""
        query_words = set(query.lower().split())
        results = []
        
        for i, doc in enumerate(self.documents):
            content = doc["content"].lower()
            matches = sum(1 for word in query_words if word in content)
            if matches > 0:
                score = matches / len(query_words)
                results.append({
                    "index": i,
                    "keyword_score": score,
                    "document": doc
                })
        
        return sorted(results, key=lambda x: x["keyword_score"], reverse=True)
    
    def _merge_results(
        self,
        semantic: List[Dict],
        keyword: List[Dict],
        sem_weight: float
    ) -> List[Dict]:
        """Fusionne les résultats avec pondération"""
        
        # Indexation keyword
        kw_indexed = {r["document"]["id"]: r for r in keyword}
        
        # Fusion
        merged = {}
        for doc in semantic:
            doc_id = doc["id"]
            kw_score = kw_indexed.get(doc_id, {}).get("keyword_score", 0)
            combined = (sem_weight * doc["similarity_score"] + 
                       (1 - sem_weight) * kw_score)
            merged[doc_id] = {
                **doc,
                "combined_score": combined
            }
        
        return sorted(
            merged.values(),
            key=lambda x: x["combined_score"],
            reverse=True
        )[:self.top_k]

Test du moteur
documents_test = [
    {"id": "1", "content": "Politique de retour sous 30 jours", "embedding": [0.1]*3072},
    {"id": "2", "content": "Livraison express 24h disponible", "embedding": [0.2]*3072},
    {"id": "3", "content": "Comment retourner mes achats", "embedding": [0.15]*3072},
]

engine = SimilaritySearchEngine(documents_test, metric="cosine", top_k=2)
query_embedding = [0.12]*3072  # "Retourner un article"

resultats = engine.search(query_embedding, threshold=0.5)
for r in resultats:
    print(f"#{r['rank']}: {r['content']} (score: {r['similarity_score']:.3f})")

Intégration avec Dify

Configuration du Knowledge Dataset

# dify_integration.py - Connexion Dify vers HolySheep
import requests
from typing import List, Dict, Optional

class DifyKnowledgeBase:
    """
    Integration avec l'API Dify pour la gestion des datasets.
    Utilise HolySheep AI pour les embeddings.
    """
    
    def __init__(
        self,
        dify_api_key: str,
        dify_base_url: str = "https://api.dify.ai/v1"
    ):
        self.dify_client = OpenAI(
            api_key=dify_api_key,
            base_url=dify_base_url
        )
        self.holy_client = DifyVectorConfig.get_client()
    
    def create_dataset(
        self,
        name: str,
        description: str = ""
    ) -> Dict:
        """Crée un nouveau dataset dans Dify"""
        
        response = requests.post(
            f"{self.dify_base_url}/datasets",
            headers={"Authorization": f"Bearer {self.dify_api_key}"},
            json={
                "name": name,
                "description": description,
                "indexing_technique": "high_quality",
                "embedding_model": "text-embedding-3-large",
                # Config HolySheep
                "embedding_provider": "holysheep"
            }
        )
        return response.json()
    
    def add_documents_with_holysheep_embeddings(
        self,
        dataset_id: str,
        documents: List[Dict]
    ) -> Dict:
        """
        Ajoute des documents avec embeddings HolySheep.
        Plus de contrôle sur la qualité.
        """
        
        # Étape 1 : Vectorisation HolySheep
        pipeline = VectorizationPipeline()
        vectorized = pipeline.vectorize_knowledge_base(documents)
        
        # Étape 2 : Upload vers Dify avec embeddings pré-calculés
        response = requests.post(
            f"{self.dify_base_url}/datasets/{dataset_id}/documents",
            headers={"Authorization": f"Bearer {self.dify_api_key}"},
            json={
                "indexing_technique": "custom",
                "process_rule": {
                    "mode": "custom",
                    "rules": {
                        "pre_processing_rules": [
                            {"id": "remove_extra_spaces", "enabled": True},
                            {"id": "remove_urls_emails", "enabled": True}
                        ],
                        "segmentation": {
                            "max_tokens": 512,
                            "overlap": 128
                        }
                    }
                },
                "documents": [
                    {
                        "display_name": doc.get("title", f"doc_{i}"),
                        "text": doc["content"],
                        "embedding": doc["embedding"]  # HolySheep embedding
                    }
                    for i, doc in enumerate(vectorized)
                ]
            }
        )
        
        return response.json()
    
    def retrieve_with_reranking(
        self,
        dataset_id: str,
        query: str,
        top_k: int = 10,
        rerank_top_k: int = 5
    ) -> List[Dict]:
        """
        Retrieve avec ré-ordonnancement pour meilleure précision.
        """
        
        # Génération embedding requête via HolySheep
        query_response = self.holy_client.embeddings.create(
            model="text-embedding-3-large",
            input=query
        )
        query_embedding = query_response.data[0].embedding
        
        # Étape 1 : Retrieval initial (via Dify)
        initial_response = requests.post(
            f"{self.dify_base_url}/datasets/{dataset_id}/retrieve",
            headers={"Authorization": f"Bearer {self.dify_api_key}"},
            json={
                "query": query,
                "retrieval_setting": {
                    "top_k": top_k,
                    "score_threshold": 0.5
                }
            }
        )
        
        candidates = initial_response.json().get("records", [])
        
        # Étape 2 : Ré-ordonnancement avec HolySheep
        if candidates:
            reranked = self._rerank_with_holysheep(
                query_embedding,
                candidates,
                rerank_top_k
            )
            return reranked
        
        return candidates
    
    def _rerank_with_holysheep(
        self,
        query_embedding: List[float],
        candidates: List[Dict],
        top_k: int
    ) -> List[Dict]:
        """Ré-ordonne les candidats via cross-encoder HolySheep"""
        
        # Construction des paires query-document
        pairs = [
            (query, cand["content"])
            for cand in candidates
            for query in [candidates[0].get("query", "")]
        ]
        
        # Note: En pratique, utilisez un modèle de cross-encoding
        # HolySheep ne propose pas de reranker natif,
        # utilisez un modèle open-source comme bge-reranker
        
        return sorted(
            candidates,
            key=lambda x: x.get("score", 0),
            reverse=True
        )[:top_k]

Configuration
dify = DifyKnowledgeBase(
    dify_api_key="YOUR_DIFY_API_KEY"
)

Exemple d'ajout de documents
nouveaux_docs = [
    {"title": "FAQ Retours", "content": "Vous pouvez retourner vos articles sous 30 jours."},
    {"title": "Livraison", "content": "Livraison gratuite pour toute commande > 50€."},
]

result = dify.add_documents_with_holysheep_embeddings(
    dataset_id="votre_dataset_id",
    documents=nouveaux_docs
)
print(f"✅ Documents ajoutés : {result}")

Comparatif des modèles d'embedding

Recommandation HolySheep : Pour le français, text-embedding-3-large offre la meilleure compréhension contextuelle avec une latence moyenne de 47ms.

Erreurs courantes et solutions

1. Erreur 401 Unauthorized — Clé API invalide

# ❌ ERREUR : Clé API incorrecte ou non configurée
AuthenticationError: Incorrect API key provided
Expected: sk-holysheep-xxxx
Got: YOUR_HOLYSHEEP_API_KEY

✅ SOLUTION : Vérifier et configurer correctement la clé

import os

Méthode 1 : Variable d'environnement (RECOMMANDÉ)
os.environ["HOLYSHEEP_API_KEY"] = "votre_cle_reelle"

Méthode 2 : Configuration directe (moins sécurisé)
from dify_config import DifyVectorConfig
DifyVectorConfig.API_KEY = "votre_cle_reelle"

Méthode 3 : Validation immédiate
def validate_api_key():
    from dify_config import DifyVectorConfig
    try:
        client = DifyVectorConfig.get_client()
        models = client.models.list()
        print("✅ Clé API valide")
        return True
    except Exception as e:
        if "401" in str(e) or "unauthorized" in str(e).lower():
            print("❌ Clé API invalide ou expirée")
            print("💡 Obtenez une nouvelle clé sur https://www.holysheep.ai/register")
        return False

validate_api_key()

2. Timeout lors de la vectorisation de gros volumes

# ❌ ERREUR : Timeout lors du traitement de 50 000+ documents
TimeoutError: Request exceeded 120s limit
Failed to process batch 847/1200

✅ SOLUTION : Implémenter batching intelligent et retry

class RobustVectorizer:
    def __init__(self, batch_size=25, timeout=60, max_retries=5):
        self.batch_size = batch_size
        self.timeout = timeout
        self.max_retries = max_retries
    
    def vectorize_large_dataset(self, documents):
        """Vectorisation robuste avec checkpointing"""
        
        results = []
        checkpoint_file = "vectorization_checkpoint.json"
        
        # Reprise sur échec
        start_idx = self._load_checkpoint(checkpoint_file)
        
        for i in range(start_idx, len(documents), self.batch_size):
            batch = documents[i:i + self.batch_size]
            
            success = False
            for attempt in range(self.max_retries):
                try:
                    embeddings = self._process_batch(batch)
                    results.extend(embeddings)
                    self._save_checkpoint(i + len(batch), checkpoint_file)
                    success = True
                    break
                    
                except TimeoutError:
                    wait = 2 ** attempt  # Backoff exponentiel
                    print(f"⏳ Retry {attempt+1} dans {wait}s...")
                    time.sleep(wait)
                    
                except RateLimitError:
                    wait = 30
                    print(f"🚦 Rate limit — pause {wait}s")
                    time.sleep(wait)
            
            if not success:
                print(f"⚠️ Batch {i} skipped après {self.max_retries} tentatives")
        
        return results
    
    def _load_checkpoint(self, filepath):
        try:
            with open(filepath) as f:
                return json.load(f).get("last_index", 0)
        except:
            return 0
    
    def _save_checkpoint(self, index, filepath):
        with open(filepath, 'w') as f:
            json.dump({"last_index": index, "timestamp": time.time()}, f)

Utilisation
vectorizer = RobustVectorizer(batch_size=25, timeout=60)
resultats = vectorizer.vectorize_large_dataset(gros_dataset)

3. Dimension mismatch — Embedding incompatibles

# ❌ ERREUR : Dimensions incompatibles entre modèles
VectorDimensionError: 
  Query embedding: 1536 dimensions
  Index embeddings: 3072 dimensions
  Cannot compute similarity

✅ SOLUTION : Alignement des dimensions ou normalisation

def normalize_embeddings(embeddings, target_dim=3072):
    """Normalise et pad/coupe les embeddings pour uniformité"""
    
    normalized = []
    for emb in embeddings:
        emb = np.array(emb)
        
        if len(emb) < target_dim:
            # Padding avec zéros
            padded = np.pad(emb, (0, target_dim - len(emb)))
            normalized.append(padded)
        elif len(emb) > target_dim:
            # Troncature
            normalized.append(emb[:target_dim])
        else:
            normalized.append(emb)
    
    return np.array(normalized)

def resize_embedding(embedding, target_dim, method="truncate"):
    """Redimensionne un embedding individuel"""
    emb = np.array(embedding)
    
    if len(emb) == target_dim:
        return emb.tolist()
    
    if method == "truncate":
        return emb[:target_dim].tolist()
    
    if method == "pad":
        return np.pad(emb, (0, max(0, target_dim - len(emb)))).tolist()
    
    # Methode interpolation (plus lent mais preserve l'information)
    if len(emb) > target_dim:
        indices = np.linspace(0, len(emb)-1, target_dim)
        return np.interp(indices, np.arange(len(emb)), emb).tolist()
    
    return emb.tolist()

Application avant stockage
def align_index_for_query(index_embeddings, query_embedding):
    """Aligne les dimensions pour la comparaison"""
    target_dim = len(query_embedding)
    return normalize_embeddings(index_embeddings, target_dim)

Exemple d'utilisation
query_emb = generate_embedding("Ma question")
index_embs = load_index_embeddings()

Alignement obligatoire avant similarity search
aligned_index = align_index_for_query(index_embs, query_emb)
similarities = cosine_similarity([query_emb], aligned_index)

4. Mauvaise qualité de retrieval — Chunks hors contexte

# ❌ ERREUR : Résultats incohérents ou hors sujet
Query: "Comment retourner mes Nike Air Max ?"
Top result: "Livraison gratuite元旦促销" (hors sujet)

✅ SOLUTION : Filtrage sémantique post-retrieval

class SemanticFilter:
    """Filtre les résultats pour améliorer la pertinence"""
    
    def __init__(self, threshold=0.5, diversity_threshold=0.8):
        self.threshold = threshold
        self.diversity_threshold = diversity_threshold
    
    def filter_results(self, results, query_context=None):
        """Filtre intelligent des résultats"""
        
        filtered = []
        seen_domains = set()
        
        for result in results:
            # Critère 1 : Score minimum
            if result["similarity_score"] < self.threshold:
                continue
            
            # Critère 2 : Diversité (éviter les doublons)
            domain = result.get("metadata", {}).get("source", "")
            if domain in seen_domains and self.diversity_threshold:
                # Réduire le score si domaine déjà représenté
                result["similarity_score"] *= 0.7
            
            seen_domains.add(domain)
            
            # Critère 3 : Contexte si fourni
            if query_context:
                context_match = self._check_context(
                    result["content"],
                    query_context
                )
                if not context_match:
                    continue
            
            filtered.append(result)
        
        # Re-triage après filtrage
        return sorted(filtered, 
                     key=lambda x: x["similarity_score"], 
                     reverse=True)
    
    def _check_context(self, content, query_context):
        """Vérifie la cohérence contextuelle"""
        # Extraire les entités de la requête
        query_entities = set(query_context.lower().split())
        content_lower = content.lower()
        
        # Vérifier au moins 1 overlap ou synonymes
        matches = sum(1 for e in query_entities if
Ressources connexes
📚 Tutoriels API IA
💰 Voir les tarifs
📖 Documentation
🚀 Inscription gratuite
Articles connexes
2026 : Le Grand Comparatif des Contextes IA — Pourquoi Migre
MCP 协议标准解析 : Guide complet de l'interopérabilité des outils 
Configuration du Framework de Test d'Intégration pour Claude