Pinecone : base vectorielle cloud managée | Guide complet

pip install pinecone-client

from pinecone import Pinecone, ServerlessSpec

# Initialiser avec votre API key
pc = Pinecone(api_key="your-api-key-here")

# Créer un index
pc.create_index(
    name="my-first-index",
    dimension=1536,  # Dimension des [embeddings](/ia/embedding/) (OpenAI)
    metric="cosine",  # ou "euclidean" ou "dotproduct"
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1"  # Choisir région proche de vous
    )
)

print("Index créé avec succès !")

# Obtenir l'index
index = pc.Index("my-first-index")

# Vérifier les stats
stats = index.describe_index_stats()
print(f"Total vecteurs : {stats['total_vector_count']}")
print(f"Dimension : {stats['dimension']}")

from openai import OpenAI

# Générer des embeddings
openai_client = OpenAI(api_key="sk-...")

def get_embedding(text: str) -> list[float]:
    """Générer embedding OpenAI."""
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

# Données à indexer
documents = [
    {
        "id": "doc1",
        "text": "Les bases de données vectorielles sont essentielles pour l'IA moderne.",
        "metadata": {
            "title": "Introduction aux bases vectorielles",
            "category": "tech",
            "date": "2025-01-15",
            "views": 1250
        }
    },
    {
        "id": "doc2",
        "text": "Pinecone est une base vectorielle cloud managée très performante.",
        "metadata": {
            "title": "Guide Pinecone",
            "category": "tech",
            "date": "2025-01-20",
            "views": 890
        }
    },
    {
        "id": "doc3",
        "text": "Le RAG améliore les LLMs en leur donnant accès à des connaissances externes.",
        "metadata": {
            "title": "RAG expliqué",
            "category": "ai",
            "date": "2025-01-18",
            "views": 2100
        }
    }
]

# Préparer les vecteurs
vectors = []
for doc in documents:
    embedding = get_embedding(doc["text"])
    vectors.append({
        "id": doc["id"],
        "values": embedding,
        "metadata": {**doc["metadata"], "text": doc["text"]}
    })

# Upsert dans Pinecone
index.upsert(vectors=vectors)

print(f"{len(vectors)} vecteurs ajoutés !")

def upsert_in_batches(index, vectors, batch_size=100):
    """
    Upsert par batches pour meilleure performance.

    Args:
        index: Index Pinecone
        vectors: Liste de vecteurs
        batch_size: Taille des batches (100-200 optimal)
    """
    for i in range(0, len(vectors), batch_size):
        batch = vectors[i:i+batch_size]
        index.upsert(vectors=batch)
        print(f"Batch {i//batch_size + 1} : {len(batch)} vecteurs ajoutés")

# Exemple : 10K vecteurs
large_vectors = [...]  # 10K vecteurs
upsert_in_batches(index, large_vectors, batch_size=100)

# Question
question = "Qu'est-ce que le RAG ?"

# Générer embedding de la question
query_embedding = get_embedding(question)

# Rechercher
results = index.query(
    vector=query_embedding,
    top_k=3,
    include_metadata=True
)

# Afficher résultats
print(f"Top {len(results['matches'])} résultats:\n")
for i, match in enumerate(results['matches'], 1):
    print(f"{i}. Score: {match['score']:.4f}")
    print(f"   ID: {match['id']}")
    print(f"   Titre: {match['metadata']['title']}")
    print(f"   Texte: {match['metadata']['text'][:100]}...")
    print()

Top 3 résultats:

1. Score: 0.8921
   ID: doc3
   Titre: RAG expliqué
   Texte: Le RAG améliore les LLMs en leur donnant accès à des connaissances externes...

2. Score: 0.7234
   ID: doc1
   Titre: Introduction aux bases vectorielles
   Texte: Les bases de données vectorielles sont essentielles pour l'IA moderne...

3. Score: 0.6789
   ID: doc2
   Titre: Guide Pinecone
   Texte: Pinecone est une base vectorielle cloud managée très performante...

results = index.query(
    vector=query_embedding,
    top_k=5,                    # Nombre de résultats
    include_metadata=True,      # Inclure métadonnées
    include_values=False,       # Inclure vecteurs (généralement False)
    namespace="production"      # Namespace spécifique (optionnel)
)

# === Égalité ===
filter = {"category": {"$eq": "tech"}}
# ou simplement
filter = {"category": "tech"}

# === In (liste de valeurs) ===
filter = {"category": {"$in": ["tech", "ai", "ml"]}}

# === Comparaisons numériques ===
filter = {
    "views": {"$gte": 1000}  # >= 1000
}

filter = {
    "views": {"$gt": 500, "$lt": 2000}  # 500 < views < 2000
}

# === Logique AND ===
filter = {
    "$and": [
        {"category": "tech"},
        {"views": {"$gte": 1000}}
    ]
}

# === Logique OR ===
filter = {
    "$or": [
        {"category": "tech"},
        {"category": "ai"}
    ]
}

# === Complexe (AND + OR) ===
filter = {
    "$and": [
        {"views": {"$gte": 1000}},
        {
            "$or": [
                {"category": "tech"},
                {"category": "ai"}
            ]
        }
    ]
}

# Question
question = "bases de données"

# Filtre : seulement catégorie "tech" avec >= 1000 vues
filter = {
    "$and": [
        {"category": "tech"},
        {"views": {"$gte": 1000}}
    ]
}

# Recherche
results = index.query(
    vector=get_embedding(question),
    top_k=3,
    filter=filter,
    include_metadata=True
)

print(f"Résultats (filtrés) : {len(results['matches'])}")
for match in results['matches']:
    print(f"- {match['metadata']['title']} (vues: {match['metadata']['views']})")

Résultats (filtrés) : 1
- Introduction aux bases vectorielles (vues: 1250)

# Ajouter dans namespace "users-data"
index.upsert(
    vectors=vectors,
    namespace="users-data"
)

# Ajouter dans namespace "products"
index.upsert(
    vectors=vectors,
    namespace="products"
)

# Rechercher dans namespace spécifique
results = index.query(
    vector=query_embedding,
    namespace="users-data",  # Seulement ce namespace
    top_k=5
)

# Stats par namespace
stats = index.describe_index_stats()
print(stats['namespaces'])
# {'users-data': {'vector_count': 1000}, 'products': {'vector_count': 500}}

# Dense embedding (OpenAI)
dense_vector = get_embedding("bases de données vectorielles")

# Sparse vector (TF-IDF like)
# Indices = positions des tokens importants
# Values = scores TF-IDF
sparse_vector = {
    "indices": [10, 45, 123, 456],  # Token IDs
    "values": [0.8, 0.6, 0.4, 0.3]  # Scores
}

# Hybrid search
results = index.query(
    vector=dense_vector,        # Dense (sémantique)
    sparse_vector=sparse_vector,  # Sparse (mots-clés)
    top_k=5,
    alpha=0.7  # 70% dense, 30% sparse (optionnel)
)

# Récupérer vecteurs spécifiques par ID
vectors = index.fetch(ids=["doc1", "doc2", "doc3"])

for id, data in vectors['vectors'].items():
    print(f"{id} : {data['metadata']['title']}")

doc1 : Introduction aux bases vectorielles
doc2 : Guide Pinecone
doc3 : RAG expliqué

# Supprimer documents spécifiques
index.delete(ids=["doc1", "doc2"])

print("Documents supprimés")

# Supprimer tous les documents d'une catégorie
index.delete(filter={"category": "obsolete"})

# Supprimer vieux documents
index.delete(filter={"date": {"$lt": "2024-01-01"}})

# ⚠️ Supprimer TOUS les vecteurs (irréversible !)
index.delete(delete_all=True)

pip install langchain langchain-pinecone langchain-openai

from langchain_pinecone import PineconeVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain.schema import Document

# Embeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Créer vectorstore
vectorstore = PineconeVectorStore(
    index_name="my-first-index",
    embedding=embeddings,
    pinecone_api_key="your-api-key"
)

# Ajouter documents
docs = [
    Document(
        page_content="Les bases vectorielles sont essentielles.",
        metadata={"source": "doc1", "category": "tech"}
    ),
    Document(
        page_content="Pinecone est très performant.",
        metadata={"source": "doc2", "category": "tech"}
    )
]

vectorstore.add_documents(docs)

# Recherche
results = vectorstore.similarity_search(
    "bases de données",
    k=2
)

for doc in results:
    print(f"- {doc.page_content[:50]}...")

User Question
    ↓
[Embedding Model] (OpenAI)
    ↓
[Pinecone Search] (top-3 documents)
    ↓
[Context Construction]
    ↓
[LLM] (GPT-4 with context)
    ↓
Answer

from pinecone import Pinecone
from openai import OpenAI

class RAGChatbot:
    """Chatbot RAG avec Pinecone + OpenAI."""

    def __init__(self, pinecone_api_key: str, openai_api_key: str, index_name: str):
        # Pinecone
        pc = Pinecone(api_key=pinecone_api_key)
        self.index = pc.Index(index_name)

        # OpenAI
        self.openai = OpenAI(api_key=openai_api_key)

    def _get_embedding(self, text: str) -> list[float]:
        """Générer embedding."""
        response = self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding

    def index_documents(self, documents: list[dict]):
        """
        Indexer des documents.

        Args:
            documents: Liste de dicts avec 'id', 'text', 'metadata'
        """
        vectors = []

        for doc in documents:
            embedding = self._get_embedding(doc['text'])
            vectors.append({
                'id': doc['id'],
                'values': embedding,
                'metadata': {**doc['metadata'], 'text': doc['text']}
            })

        # Upsert par batches
        batch_size = 100
        for i in range(0, len(vectors), batch_size):
            batch = vectors[i:i+batch_size]
            self.index.upsert(vectors=batch)

        print(f"{len(vectors)} documents indexés")

    def ask(self, question: str, top_k: int = 3, filter: dict = None) -> dict:
        """
        Poser une question au chatbot.

        Args:
            question: Question de l'utilisateur
            top_k: Nombre de documents à récupérer
            filter: Filtres métadonnées (optionnel)

        Returns:
            Dict avec 'answer', 'sources', 'context'
        """
        # 1. Embedder la question
        query_embedding = self._get_embedding(question)

        # 2. Rechercher dans Pinecone
        results = self.index.query(
            vector=query_embedding,
            top_k=top_k,
            filter=filter,
            include_metadata=True
        )

        # 3. Construire le contexte
        contexts = []
        sources = []

        for match in results['matches']:
            contexts.append(match['metadata']['text'])
            sources.append({
                'id': match['id'],
                'score': match['score'],
                'title': match['metadata'].get('title', 'N/A')
            })

        context = "\n\n---\n\n".join(contexts)

        # 4. Générer réponse avec LLM
        system_prompt = """Tu es un assistant qui répond aux questions en utilisant uniquement le contexte fourni.

Règles:
- Réponds UNIQUEMENT avec les informations du contexte
- Si le contexte ne contient pas la réponse, dis "Je ne trouve pas cette information dans les documents fournis"
- Sois concis et précis
- Cite les sources si pertinent"""

        user_prompt = f"""Contexte:
{context}

Question: {question}

Réponse:"""

        response = self.openai.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            temperature=0.3  # Plus déterministe
        )

        answer = response.choices[0].message.content

        return {
            'answer': answer,
            'sources': sources,
            'context': context
        }

# === Utilisation ===

# Initialiser
chatbot = RAGChatbot(
    pinecone_api_key="your-pinecone-key",
    openai_api_key="your-openai-key",
    index_name="knowledge-base"
)

# Indexer documents (une fois)
documents = [
    {
        'id': 'doc1',
        'text': 'Notre politique de retour autorise les retours dans les 30 jours avec le reçu original.',
        'metadata': {'title': 'Politique de retour', 'category': 'support'}
    },
    {
        'id': 'doc2',
        'text': 'Nous offrons une garantie de 2 ans sur tous les produits électroniques.',
        'metadata': {'title': 'Garantie', 'category': 'support'}
    },
    {
        'id': 'doc3',
        'text': 'La livraison est gratuite pour les commandes supérieures à 50€.',
        'metadata': {'title': 'Livraison', 'category': 'support'}
    }
]

chatbot.index_documents(documents)

# Poser des questions
questions = [
    "Quelle est la durée de la garantie ?",
    "Puis-je retourner un produit ?",
    "Y a-t-il des frais de livraison ?"
]

for question in questions:
    print(f"\n❓ {question}")
    result = chatbot.ask(question)
    print(f"💬 {result['answer']}")
    print(f"📚 Sources: {[s['title'] for s in result['sources']]}")

❓ Quelle est la durée de la garantie ?
💬 Nous offrons une garantie de 2 ans sur tous les produits électroniques.
📚 Sources: ['Garantie', 'Politique de retour']

❓ Puis-je retourner un produit ?
💬 Oui, notre politique de retour autorise les retours dans les 30 jours avec le reçu original.
📚 Sources: ['Politique de retour', 'Garantie']

❓ Y a-t-il des frais de livraison ?
💬 Non, la livraison est gratuite pour les commandes supérieures à 50€.
📚 Sources: ['Livraison', 'Garantie']

stats = index.describe_index_stats()

print(f"Total vecteurs : {stats['total_vector_count']}")
print(f"Dimension : {stats['dimension']}")
print(f"Index fullness : {stats.get('index_fullness', 0):.2%}")

# Stats par namespace
if 'namespaces' in stats:
    for ns, data in stats['namespaces'].items():
        print(f"Namespace '{ns}' : {data['vector_count']} vecteurs")

# ❌ Créer plusieurs index (coûteux)
pc.create_index("users")
pc.create_index("products")
pc.create_index("analytics")

# ✅ Un index avec namespaces (économique)
index.upsert(vectors, namespace="users")
index.upsert(vectors, namespace="products")
index.upsert(vectors, namespace="analytics")

# ❌ Upsert un par un (lent et coûteux)
for vector in vectors:
    index.upsert(vectors=[vector])

# ✅ Batch de 100-200 (optimal)
for i in range(0, len(vectors), 100):
    index.upsert(vectors=vectors[i:i+100])

# ❌ Dimension trop grande si pas nécessaire
dimension = 3072  # OpenAI text-embedding-3-large

# ✅ Dimension optimale pour la plupart des cas
dimension = 1536  # OpenAI text-embedding-3-small
# ou
dimension = 768   # Sentence-BERT base

# Supprimer vieux documents régulièrement
import datetime

cutoff_date = (datetime.datetime.now() - datetime.timedelta(days=365)).isoformat()

index.delete(filter={"date": {"$lt": cutoff_date}})

# Production : Starter plan ($70/mois)
production_index = pc.Index("production")

# Dev/Staging : Free tier (100K vecteurs)
dev_index = pc.Index("dev")

from sklearn.preprocessing import normalize
vectors_normalized = normalize(vectors, norm='l2')

metadata = {
    'title': '...',
    'category': '...',
    'date': '2025-01-15',
    'author': '...',
    'views': 1250,
    'tags': ['ai', 'ml', 'rag']
}

index.upsert(vectors, namespace="production")

# Batches de 100-200 vecteurs

stats = index.describe_index_stats()

# ❌ Query Pinecone à chaque fois
result = index.query(...)

# ✅ Cache les résultats similaires
cache_key = hash(question)
if cache_key in redis_cache:
    return redis_cache[cache_key]

# ❌ Stocker tous les logs
# ✅ Stocker seulement documents importants

# ✅ Combiner similarité + filtres business
filter = {"category": "tech", "views": {"$gte": 1000}}