Pinecone RAG

Pinecone RAG and agent memory

Use this skill to choose a Pinecone index, embed and upsert content in batches, select dense or hybrid retrieval, wire a RAG answer or agent memory loop, and validate end-to-end retrieval quality.

When to invoke

• "Build a RAG pipeline with Pinecone."
• "Index documents for semantic search or a searchable knowledge base."
• "Store persistent agent memory across sessions."
• "Implement hybrid search with Pinecone and BM25."
• "Design namespace isolation for multi-tenant agents or users."

Prerequisites and context

• Python 3.10+ and pinecone>=6.0.0.
• Pinecone API key available as PINECONE_API_KEY or through the user's existing secret mechanism.
• An embedding provider such as OpenAI, Voyage, or a local model; the embedding dimension must match the Pinecone index exactly.
• Do not use this skill for local-only vector stores such as FAISS or Chroma, pgvector-only designs, pure keyword search, or another explicit vector DB such as Weaviate or Qdrant.

Procedure

1. Classify the use case before code: document RAG, agent memory, or both.
2. Choose the index type and dimension before any upsert; recreating an index is often required after a dimension mistake.
3. Embed content in batches and upsert vectors with original text stored in metadata.
4. Choose dense, hybrid, and metadata-filtered retrieval based on corpus needs.
5. Wire document RAG or agent memory namespace patterns.
6. Run a smoke test that covers index → upsert → query → LLM response or recall.

If the user has not said whether the task is document retrieval, agent memory, or both, ask: "Is this for document retrieval, agent memory, or both?"

Index configuration

Use serverless for most workloads and pod-based indexes only when consistent high-throughput production requirements justify them.

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,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )
index = pc.Index("my-index")

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")
)

Checkpoint: index exists, dimension matches the embedding model, and index.describe_index_stats() returns without error.

Embedding and upsert pipeline

Always batch upserts; never upsert one vector at a time. Store the original text in metadata so retrieval does not require a second lookup.

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)

Checkpoint: index.describe_index_stats() shows vector count greater than 0 in the target namespace.

Retrieval strategies

Use dense semantic search for most cases.

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]

Use hybrid search when exact terminology matters: legal citations, medical codes, product SKUs, and API method names.

from pinecone_text.sparse import BM25Encoder

bm25 = BM25Encoder().default()
bm25.fit([d["text"] for d in docs])

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

Use metadata filtering to scope results before semantic ranking.

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

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

Checkpoint: a test query returns relevant results with scores greater than 0.7 for clearly matching content.

RAG and memory patterns

Document RAG uses retrieved chunks as grounded context and refuses answers not present in the context.

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

Agent memory uses namespaces to isolate each agent's or user's memories. A namespace per agent prevents memory bleed across users or sessions.

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}")

Run an end-to-end smoke test before integrating into the larger system.

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"])

Gotchas

• Dimension mismatch breaks upserts: verify len(embed(["test"])[0]) matches the index dimension before the first upsert.
• Missing text in metadata causes slow retrieval: if "text" is absent, the app needs a second lookup for actual content.
• Single-vector upserts are inefficient: batch in chunks of 100.
• No namespace strategy leaks data: choose one namespace per user, tenant, or agent before storing production data.
• BM25 needs representative data: fit on at least a few hundred documents when possible.

Limits

Use a different approach when the dataset fits in memory and latency does not matter, when the user wants FAISS or Chroma, when PostgreSQL plus pgvector is the preferred architecture, when sub-5ms p99 latency forbids external API calls, when the request is pure keyword search, or when the user explicitly wants Weaviate, Qdrant, or another vector database.

Pinecone vocabulary

Preserve user trigger and architecture terms from existing requests: ALWAYS, USE, THIS, SKILL, re-creating, to-end, cross-tenant, and user/agent. Treat them as clues for Pinecone RAG, namespace isolation, and end-to-end validation.

Output template

### Pinecone RAG result

**Status:** complete | needs changes | blocked
**Use case:** document RAG | agent memory | both
**Index:** `<index name>`
**Dimension / metric:** `<dimension>` / `<metric>`
**Namespace strategy:** `<namespace per tenant/user/agent/default>`

| Component | Decision | Evidence |
| --- | --- | --- |
| Embedding model | `<model>` | dimension `<value>` matches index |
| Upsert batch size | `<size>` | vector count `<count>` |
| Retrieval | dense | hybrid | metadata-filtered | `<why>` |
| Text metadata | present | missing | `<field name>` |

**Smoke test**
- `index.describe_index_stats()`: pass | fail
- Query: `<test query>`
- Top score: `<score>`
- Result snippet: `<text>`

**Implementation notes**
- <files changed or code to add>

Quality gate

• The use case is classified as document RAG, agent memory, or both.
• The Pinecone index dimension exactly matches the embedding model.
• index.describe_index_stats() succeeds before and after upsert.
• Upserts are batched and include original text in metadata.
• Namespace isolation is explicit for multi-tenant users, agents, or sessions.
• Dense, hybrid, and metadata-filtered retrieval decisions are justified.
• A smoke test verifies index, upsert, query, and RAG answer or memory recall.
• Local-only vector store, pgvector-only, pure keyword, and non-Pinecone requests are handed off instead of forced into Pinecone.