Skill Pinecone RAG

Ce skill vous guide dans la construction d'un pipeline RAG de production ou d'un système de mémoire persistante pour agent utilisant Pinecone. Suivez le workflow du début à la fin — ne sautez pas d'étapes et ne passez au code que lorsque vous comprenez réellement ce dont l'utilisateur a besoin.

Avant de commencer — posez une question

Avant d'écrire du code, identifiez lequel de ces deux cas d'usage s'applique :

A — RAG sur documents : L'utilisateur veut indexer un corpus (PDFs, docs, code, pages web) et récupérer des chunks pertinents pour ancrer les réponses du LLM.

B — Mémoire d'agent : L'utilisateur veut qu'un agent se souvienne de faits, de décisions ou du contexte entre les sessions ou entre plusieurs agents partageant une base de connaissances.

La configuration est similaire mais la stratégie de namespace et les patterns de récupération diffèrent. Si l'utilisateur ne l'a pas précisé, demandez : « Est-ce pour la récupération de documents, la mémoire d'agent, ou les deux ? » Puis suivez le workflow pertinent ci-dessous.

Étape 1 — Choisir la configuration de votre index

Choisissez le type d'index avant d'écrire du code. Une mauvaise configuration signifie recréer l'index plus tard.

Serverless (recommandé pour la plupart des cas)

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="PINECONE_API_KEY")

if "my-index" not in pc.list_indexes().names():
    pc.create_index(
        name="my-index",
        dimension=1536,        # doit correspondre exactement à votre modèle d'embedding
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )
index = pc.Index("my-index")

Pod-based (pour un débit élevé constant en production)

from pinecone import PodSpec

pc.create_index(
    name="my-index-prod",
    dimension=1536,
    metric="cosine",
    spec=PodSpec(environment="us-east1-gcp", pod_type="p1.x1")
)

Référence rapide des dimensions — correspondre exactement à votre modèle d'embedding : | Modèle | Dimension | |---|---| | text-embedding-3-small | 1536 | | text-embedding-3-large | 3072 | | voyage-3 / voyage-multimodal-3 | 1024 | | BAAI/bge-large-en-v1.5 | 1024 | | intfloat/multilingual-e5-large (Arabic, Malay, Chinese) | 1024 |

Checkpoint : L'index existe, la dimension correspond au modèle d'embedding, index.describe_index_stats() s'exécute sans erreur.

Étape 2 — Embedder et upsert les documents

Toujours faire un batch des upserts — ne jamais upsert un seul vecteur à la fois.

from openai import OpenAI

client = OpenAI()

def embed(texts: list[str]) -> list[list[float]]:
    res = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return [r.embedding for r in res.data]

def upsert_docs(index, docs: list[dict], namespace: str = "default"):
    """docs = [{"id": "...", "text": "...", "metadata": {...}}]"""
    BATCH = 100
    for i in range(0, len(docs), BATCH):
        batch = docs[i:i + BATCH]
        vecs = [
            {
                "id": d["id"],
                "values": emb,
                "metadata": {**d.get("metadata", {}), "text": d["text"]}
            }
            for d, emb in zip(batch, embed([d["text"] for d in batch]))
        ]
        index.upsert(vectors=vecs, namespace=namespace)

Toujours stocker le texte original dans les métadonnées — cela évite une deuxième recherche au moment de la récupération.

Checkpoint : index.describe_index_stats() affiche un nombre de vecteurs > 0 dans le namespace cible.

Étape 3 — Choisir la stratégie de récupération

Recherche dense (sémantique) — à utiliser dans la plupart des cas

def search(index, query: str, top_k: int = 5, namespace: str = "default",
           filter: dict = None) -> list[dict]:
    [q_emb] = embed([query])
    results = index.query(
        vector=q_emb, top_k=top_k, namespace=namespace,
        include_metadata=True, filter=filter
    )
    return [{"text": m.metadata["text"], "score": m.score, "id": m.id}
            for m in results.matches]

Recherche hybride (sémantique + BM25 par mot-clé) — à utiliser quand le corpus a une terminologie précise

Utilisez la recherche hybride quand le domaine a des termes précis que la recherche sémantique manque : citations légales, codes médicaux, SKU de produits, noms de méthodes API.

from pinecone_text.sparse import BM25Encoder

bm25 = BM25Encoder().default()
bm25.fit([d["text"] for d in docs])  # fit une seule fois sur votre corpus

def hybrid_search(index, query: str, top_k: int = 5, alpha: float = 0.7):
    """alpha=1.0 est pur dense ; alpha=0.0 est pur sparse."""
    dense = [v * alpha for v in embed([query])[0]]
    sparse_raw = bm25.encode_queries(query)
    sparse = {
        "indices": sparse_raw["indices"],
        "values": [v * (1 - alpha) for v in sparse_raw["values"]]
    }
    return index.query(vector=dense, sparse_vector=sparse,
                       top_k=top_k, include_metadata=True).matches

Filtrage par métadonnées — à utiliser pour limiter les résultats avant le classement sémantique

# Correspondance exacte
results = index.query(vector=emb, filter={"source": {"$eq": "confluence"}})

# Filtre combiné
results = index.query(vector=emb, filter={
    "$and": [
        {"category": {"$eq": "engineering"}},
        {"language": {"$in": ["en", "ar"]}}
    ]
})

Checkpoint : Une requête de test retourne des résultats pertinents avec des scores > 0,7 pour le contenu clairement correspondant.

Étape 4A — Pipeline RAG complet (cas d'usage document)

def rag_answer(index, question: str, namespace: str = "default",
               model: str = "gpt-4o-mini") -> str:
    hits = search(index, question, top_k=5, namespace=namespace)
    context = "\n\n".join(h["text"] for h in hits)

    return client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "system",
                "content": (
                    "Answer using only the provided context. "
                    "If the answer isn't in the context, say so.\n\n"
                    f"Context:\n{context}"
                )
            },
            {"role": "user", "content": question}
        ]
    ).choices[0].message.content

Étape 4B — Mémoire d'agent (cas d'usage mémoire)

Utilisez les namespaces pour isoler complètement les mémoires de chaque agent ou utilisateur. Un namespace par agent prévient les fuites de mémoire entre utilisateurs ou sessions.

import time, hashlib

def remember(index, agent_id: str, content: str,
             memory_type: str = "fact"):
    """Stocker une mémoire pour un agent."""
    mem_id = hashlib.md5(
        f"{agent_id}{content}{time.time()}".encode()
    ).hexdigest()
    [emb] = embed([content])
    index.upsert(
        vectors=[{
            "id": mem_id,
            "values": emb,
            "metadata": {
                "text": content,
                "type": memory_type,
                "timestamp": time.time(),
                "agent_id": agent_id
            }
        }],
        namespace=f"agent_{agent_id}"
    )

def recall(index, agent_id: str, query: str,
           top_k: int = 5) -> list[str]:
    """Récupérer les mémoires pertinentes pour un agent."""
    return [h["text"] for h in
            search(index, query, top_k=top_k,
                   namespace=f"agent_{agent_id}")]

def forget(index, agent_id: str):
    """Effacer toutes les mémoires d'un agent (par ex. sur demande de l'utilisateur)."""
    index.delete(delete_all=True, namespace=f"agent_{agent_id}")

Étape 5 — Assembler et tester bout en bout

Lancez un test rapide avant d'intégrer dans le système plus large :

# Smoke test
upsert_docs(index, [
    {"id": "t1", "text": "Pinecone is a vector database for semantic search."},
    {"id": "t2", "text": "RAG combines retrieval with language model generation."},
])

hits = search(index, "What is Pinecone?")
assert hits[0]["score"] > 0.7, f"Expected high similarity, got {hits[0]['score']}"
print("Smoke test passed:", hits[0]["text"])

Checkpoint : Le smoke test réussit. Bout en bout : index → upsert → query → réponse LLM fonctionne sans erreur.

Pièges courants — corriger ces avant qu'ils ne deviennent des bugs

Dimension mismatch : toujours vérifier que len(embed(["test"])[0]) correspond à la dimension de l'index avant le premier upsert.
Texte manquant dans les métadonnées : si vous ne stockez pas "text" dans les métadonnées, vous aurez besoin d'une deuxième recherche pour obtenir le contenu réel au moment de la requête.
Upserts de vecteurs uniques dans une boucle : toujours faire un batch par chunks de 100.
Pas de stratégie de namespace : décider à l'avance — un namespace par utilisateur/agent prévient les fuites de données cross-tenant qui sont difficiles à corriger plus tard.
Fitting BM25 sur un petit corpus : BM25 a besoin d'un corpus représentatif pour construire de bonnes fréquences de termes. Faire un fit sur au moins quelques centaines de documents.

Quand NE PAS utiliser ce skill

Utiliser une approche différente quand :

L'ensemble de données tient en mémoire et la latence n'a pas d'importance → utiliser FAISS ou Chroma
Vous êtes déjà sur PostgreSQL et voulez éviter un nouveau service → utiliser pgvector
Vous avez besoin d'une latence p99 < 5ms sans appels API externes → local vector store
L'utilisateur veut explicitement une base de données vectorielle différente (Weaviate, Qdrant, etc.)

pinecone-rag