pinecone-rag

---
name: pinecone-rag
description: >
  Build production RAG pipelines and persistent agent memory using Pinecone as
  the vector database backend. ALWAYS USE THIS SKILL when the user mentions
  Pinecone, wants to index documents for semantic search, build a
  retrieval-augmented generation system, store agent memory across sessions,
  implement hybrid search, or connect an LLM to a searchable knowledge base —
  even if they don't say "Pinecone" explicitly. Also use when the user asks
  about vector databases for RAG, namespace isolation for multi-tenant agents,
  embedding pipelines, or scaling a knowledge base beyond what local storage
  can handle. DO NOT use for local-only vector stores (Chroma, FAISS, pgvector)
  or pure keyword search with no semantic component.
license: Apache-2.0
compatibility: "pinecone>=6.0.0, Python 3.10+"
---

# Pinecone RAG Skill

This skill guides you through building a production RAG pipeline or persistent
agent memory system using Pinecone. Follow the workflow from start to finish —
don't skip steps or jump to code before understanding what the user actually
needs.

## Before you start — ask one question

Before writing any code, identify which of these two use cases applies:

**A — RAG over documents**: User wants to index a corpus (PDFs, docs, code,
web pages) and retrieve relevant chunks to ground LLM responses.

**B — Agent memory**: User wants an agent to remember facts, decisions, or
context across sessions or across multiple agents sharing a knowledge base.

The setup is similar but the namespace strategy and retrieval patterns differ.
If the user hasn't said, ask: *"Is this for document retrieval, agent memory,
or both?"* Then follow the relevant workflow below.

---

## Step 1 — Choose your index configuration

Pick the index type before writing any code. Getting this wrong means
re-creating the index later.

**Serverless (recommended for most cases)**
```python
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,        # must match your embedding model exactly
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )
index = pc.Index("my-index")
```

**Pod-based (for consistent high-throughput production)**
```python
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")
)
```

**Dimension quick reference — match this exactly to your embedding model:**
| Model | 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**: Index exists, dimension matches embedding model, `index.describe_index_stats()` returns without error.

---

## Step 2 — Embed and upsert documents

Always batch upserts — never upsert one vector at a time.

```python
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)
```

**Always store the original text in metadata** — this avoids a second lookup
at retrieval time.

> **Checkpoint**: `index.describe_index_stats()` shows vector count > 0 in the
> target namespace.

---

## Step 3 — Choose retrieval strategy

### Dense (semantic) search — use for most cases
```python
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]
```

### Hybrid search (semantic + BM25 keyword) — use when corpus has exact terminology
Use hybrid when the domain has precise terms that semantic search misses:
legal citations, medical codes, product SKUs, API method names.

```python
from pinecone_text.sparse import BM25Encoder

bm25 = BM25Encoder().default()
bm25.fit([d["text"] for d in docs])  # fit once on your corpus

def hybrid_search(index, query: str, top_k: int = 5, alpha: float = 0.7):
    """alpha=1.0 is pure dense; alpha=0.0 is pure 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
```

### Metadata filtering — use to scope results before semantic ranking
```python
# Exact match
results = index.query(vector=emb, filter={"source": {"$eq": "confluence"}})

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

> **Checkpoint**: A test query returns relevant results with scores > 0.7 for
> clearly matching content.

---

## Step 4A — Full RAG pipeline (document use case)

```python
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
```

---

## Step 4B — Agent memory (memory use case)

Use namespaces to isolate each agent's or user's memories completely.
Namespace per agent prevents memory bleed across users or sessions.

```python
import time, hashlib

def remember(index, agent_id: str, content: str,
             memory_type: str = "fact"):
    """Store a memory for an 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]:
    """Recall relevant memories for an 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):
    """Wipe all memories for an agent (e.g., on user request)."""
    index.delete(delete_all=True, namespace=f"agent_{agent_id}")
```

---

## Step 5 — Wire it together and test end to end

Run a quick smoke test before integrating into the larger system:

```python
# 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**: Smoke test passes. End-to-end: index → upsert → query →
> LLM response works without errors.

---

## Common pitfalls — fix these before they become bugs

- **Dimension mismatch**: always verify `len(embed(["test"])[0])` matches
  the index dimension before your first upsert.
- **Missing text in metadata**: if you don't store `"text"` in metadata,
  you'll need a second lookup to get the actual content at query time.
- **Single-vector upserts in a loop**: always batch in chunks of 100.
- **No namespace strategy**: decide upfront — one namespace per user/agent
  prevents cross-tenant data leaks that are hard to fix later.
- **Fitting BM25 on a small corpus**: BM25 needs a representative corpus to
  build good term frequencies. Fit on at least a few hundred documents.

## When NOT to use this skill

Use a different approach when:
- The dataset fits in memory and latency doesn't matter → use FAISS or Chroma
- You're already on PostgreSQL and want to avoid a new service → use pgvector
- You need sub-5ms p99 latency with no external API calls → local vector store
- The user explicitly wants a different vector DB (Weaviate, Qdrant, etc.)