Pinecone Vector Database: A Hands-On Python Tutorial

A Pinecone vector database stores text as lists of numbers, called embeddings, and finds the entries closest in meaning to a question. That lets you search by what someone meant instead of what they typed. This tutorial builds one from scratch in Python: embeddings, an index, upserts, filters and a small retrieval-augmented generation (RAG) loop.

Say you keep a trading journal. In January you wrote “Sold NVDA in a panic after a red day.” In February you wrote “Closed my chip stock early because I was scared.” Now you ask, “When do I sell winners out of fear?”

A keyword search finds neither entry, because neither contains the word “fear.” A vector search finds both. That gap is the whole reason vector databases exist.

I checked every Pinecone call in this post against the pinecone Python SDK 10.0.0 (Python 3.10 or newer) and Pinecone’s docs as of September 2026. I ran the data-plane calls (upsert, query, filters, fetch, list, delete) against Pinecone’s local emulator. I did not run index creation against the live service, the OpenAI calls or the chat step, so those follow each vendor’s docs.

The design lessons come from two projects of mine: Tradion, an AI trading platform in production, and RAGnetic, an open source project I built and never launched. One of them taught me by working. The other taught me by failing.

What is a Pinecone vector database, and what does it do?

Pinecone is a hosted vector database. You send it vectors, and it returns the stored ones closest to a query vector. Two pieces make that work.

1. An embedding model turns text into a fixed-length list of numbers, so that texts with similar meaning get similar numbers.
2. A vector database stores those lists and, given a new one, quickly returns the closest matches.

That is the whole idea. Everything else in this post is engineering around those two steps.

Use case	What gets embedded	What gets searched
Semantic search	Product descriptions, articles	A user’s search phrase
Chat with your documents (RAG)	Chunks of PDFs, wikis, notes	A user’s question
Long-term memory for AI assistants	Facts learned about a user	The current conversation
Recommendations	Items, users	“More like this”
Duplicate and anomaly detection	Tickets, logs, transactions	A new item

Tradion uses the third row. When a user chats with its assistant, the app searches vectors built from that user’s own past trades, post-trade reviews, chart analyses, alerts and earlier conversations. It hands the closest matches to the model as context. The assistant can then say something specific to this trader instead of giving generic advice.

How do embeddings turn text into vectors?

An embedding is a list of floating point numbers, for example 1,536 of them. You never read the numbers directly. What matters is geometry: each text becomes a point in a very high-dimensional space, and texts with similar meaning land near each other.

To build intuition, shrink the space to four dimensions we can name: [fear, tech, urgency, discipline]. Real models learn their own axes, and nobody can label them this neatly, but the math is identical. This script needs no libraries and no API keys.

# toy_similarity.py
import math

def dot(a, b):
    return sum(x * y for x, y in zip(a, b))

def norm(a):
    return math.sqrt(dot(a, a))

def cosine(a, b):
    return dot(a, b) / (norm(a) * norm(b))

def euclid(a, b):
    return math.sqrt(sum((x - y) ** 2 for x, y in zip(a, b)))

docs = {
    "Sold NVDA in a panic after a red day":            [0.9, 0.3, 0.9, 0.1],
    "Closed my chip stock early because I was scared": [0.8, 0.4, 0.8, 0.2],
    "Held TSLA to target and followed my plan":        [0.2, 0.3, 0.1, 0.9],
    "Recipe for banana bread":                         [0.0, 0.0, 0.0, 0.1],
}

# "When do I sell winners out of fear?"
query = [0.85, 0.35, 0.85, 0.1]

for text, vec in docs.items():
    print(f"{text:<50} cos={cosine(query, vec):.3f} "
          f"dot={dot(query, vec):.3f} euclid={euclid(query, vec):.3f}")

Sold NVDA in a panic after a red day               cos=0.999 dot=1.645 euclid=0.087
Closed my chip stock early because I was scared    cos=0.995 dot=1.520 euclid=0.132
Held TSLA to target and followed my plan           cos=0.368 dot=0.450 euclid=1.276
Recipe for banana bread                            cos=0.080 dot=0.010 euclid=1.252

Notice three things:

• The two fear entries score near 1.0 even though they share almost no words.
• The disciplined trade scores much lower, and banana bread scores near zero.
• The ranking is what a vector database gives you. The absolute numbers depend on the model and the metric.

Which similarity metric should you use?

Metric	Measures	Bigger means	Typical use
Cosine	The angle between two vectors, ignoring length	More similar (max 1.0)	Text embeddings. The safe default
Dot product	Angle and length	More similar	Recommenders where magnitude carries meaning. Identical to cosine when vectors are unit length
Euclidean	Straight-line distance	Less similar (it is a distance)	Some image and numeric embeddings

Two rules follow. First, use the metric your embedding model was designed for. For OpenAI text embeddings, cosine is the standard choice.

Second, know which direction “better” points. With cosine and dot product, higher is better. With Euclidean distance, lower is better. Mixing this up silently reverses your ranking.

RAGnetic’s debug command prints every retrieved chunk under the heading “Distance Score … Lower is More Relevant.” It prints that for every vector store it supports, including Pinecone, where a cosine index returns higher-is-better scores. It was confidently wrong, in bold cyan.

How do you get real embeddings?

You call an embedding model. With OpenAI:

from openai import OpenAI

client = OpenAI()  # reads OPENAI_API_KEY from the environment
response = client.embeddings.create(
    model="text-embedding-3-small",
    input=["Sold NVDA in a panic after a red day"],
)
vector = response.data[0].embedding   # a list of 1,536 floats

The OpenAI embeddings guide documents the facts worth knowing:

• text-embedding-3-small returns 1,536 numbers by default and text-embedding-3-large returns 3,072.
• Both accept up to 8,192 tokens of input per text.
• A dimensions parameter returns a shortened embedding, trading a little accuracy for less storage.

Two more rules are on you. The dimension of your index must equal the dimension of your vectors, which is the most common beginner error. And embed your documents and your queries with the same model, because vectors from different models live in different spaces and cannot be compared.

Tradion embeds with text-embedding-3-large at 3,072 dimensions, double the size of the 1,536 used here. Bigger is not automatically better. It is a cost-versus-quality decision you should measure on your own data.

What should you embed?

The text you feed in decides the quality. There are two approaches.

• Chunking: split long documents into passages of a few hundred tokens, often with slight overlap, and embed each one. Use this for PDFs, articles and wikis, where one document holds many ideas.
• Field concatenation: for short structured records like a trade, a note or a ticket, join the meaningful fields into one sentence and embed that.

Tradion’s records are short and structured, so it does not chunk. A trade becomes one text: the symbol, direction, result, date and prices, followed by the trader’s stated reason, emotion and notes. One record, one vector, one idea. The retrieval unit is the business object.

Exact vs approximate search: why not loop over everything?

The simplest search compares your query against every stored vector and keeps the best. That is exact nearest-neighbor search, the same core as the k-nearest neighbors algorithm. For a few thousand vectors it is fine in plain Python.

It stops working at scale. Ten million vectors of 1,536 floats is about 60 GB of raw data, and scanning all of it for every query is far too slow.

Approximate nearest-neighbor (ANN) algorithms trade a tiny amount of accuracy for a huge speedup. They pre-organize vectors into graphs or clusters, so a query only examines a small, promising slice. You get the true top result almost always, in milliseconds.

Three words come up constantly. Recall is the fraction of the true top-k neighbors the search returned, so 0.95 means it found 95%. Latency is how long one query takes. Index is the pre-built structure that makes search fast, though Pinecone also uses the word for the container that holds your vectors.

Pinecone runs the ANN machinery for you. Your job is to choose a metric and a dimension, put good vectors in and query sensibly.

How does Pinecone organize your data?

Five nouns explain the whole API.

• Index: the top-level container. You choose its dimension and metric when you create it, and you cannot change them later. On Pinecone’s serverless plans you do not manage servers.
• Record (also called a vector): an id string, values (the vector) and optional metadata.
• Metadata: a flat key-value dictionary on a record. It lets you filter searches (“only NVDA trades from this year”) and carry the original text, so a result is usable immediately.
• Namespace: a partition inside an index. Every read and write targets exactly one namespace, and a query only sees vectors in the namespace it targets. Namespaces appear automatically the first time you upsert into them, and they are Pinecone’s tool for multitenancy.
• Upsert: insert or overwrite. If a record with that ID already exists in that namespace, the whole record is replaced.

What limits should you know?

These come from Pinecone’s limits pages as of September 2026. Plans change, so check before you design around them.

Thing	Limit
Records per upsert request	1,000
Upsert request size	2 MB
Metadata per record	40 KB, flat (strings, numbers, booleans, lists of strings)
top_k per query	10,000
IDs per fetch or delete request	1,000
Values per $in / $nin filter	10,000
Record ID length	1 to 512 ASCII characters
Max dense dimension	20,000
Namespaces per serverless index	100 (Starter), 1,000 (Builder), 100,000 (Standard), 1,000,000 (Enterprise)
Serverless indexes per project	5 / 10 / 20 / 200 on the same plans

Why can a query miss data you just wrote?

Pinecone is eventually consistent. A record you just upserted may not show up in a query for a moment, and updates and deletes take a moment too. Every write gets a log sequence number, and a query reflects a write once the query’s number is at least the write’s. The data freshness docs explain how to check.

For most apps, the rule is simple: wait a second in tests and ignore it in production. Just never write test code that upserts and then immediately asserts on a query result. Yes, you. I know you were about to.

What does it cost?

Serverless pricing has three parts: storage, read units and write units. Pinecone’s cost docs say a query uses one read unit for every 1 GB of namespace size, with a minimum of 0.25 per query. They also say top_k, include_metadata and include_values do not change the read cost. Only the size of the namespace being searched does.

Writes scale with size. An upsert uses one write unit per 1 KB of the request, with a minimum of five per request, so tiny records are cheaper when you batch them.

Two consequences follow. Small namespaces make queries cheaper, so per-user namespaces mean each query pays for one user’s data instead of the whole index. And big vectors or big metadata cost you on every write and every stored GB, so do not store a novel in metadata just in case.

How do you build a Pinecone vector database in Python?

We will build a small semantic memory for the trading journal. By the end you can ask “why do I sell winners out of fear?” and get the relevant entries back.

Setup

python -m venv .venv
source .venv/bin/activate
pip install "pinecone>=10,<11" openai

Keep your keys in the environment, never in source code and never in git.

export PINECONE_API_KEY="..."     # from the Pinecone console
export OPENAI_API_KEY="..."
export CHAT_MODEL="..."           # a chat model name from your provider's current docs

The project is a small package. Create an empty journal/__init__.py, then the files below.

journal/
  __init__.py
  config.py
  embeddings.py
  sample_data.py
  create_index.py
  ingest.py
  search.py
  rag.py
  maintenance.py
  evaluate.py

# journal/config.py
from openai import OpenAI
from pinecone import Pinecone

INDEX_NAME = "trade-journal-demo"
EMBED_MODEL = "text-embedding-3-small"
DIMENSION = 1536   # must match EMBED_MODEL's output size
METRIC = "cosine"

# Construct these once and reuse them: each holds a connection pool.
pc = Pinecone()             # reads PINECONE_API_KEY
openai_client = OpenAI()    # reads OPENAI_API_KEY


def get_index():
    """Return a data-plane handle for our index.

    pc.index(name) looks up the index's host for you. In production, look the
    host up once (pc.indexes.describe(name).host) and pass pc.index(host=...)
    to skip that extra call on every cold start.
    """
    return pc.index(INDEX_NAME)

The SDK has two planes. pc.indexes.* is the control plane, for creating, describing and deleting indexes. pc.index(...) returns an Index object for the data plane, which handles upsert, query, fetch and delete.

# journal/embeddings.py
from journal import config


def embed_texts(texts: list[str]) -> list[list[float]]:
    """Embed a batch of texts. Order of results matches order of inputs."""
    response = config.openai_client.embeddings.create(
        model=config.EMBED_MODEL,
        input=texts,
    )
    return [item.embedding for item in response.data]

Batch texts into one call whenever you can. One round trip for 100 texts is far cheaper than 100 round trips.

Some journal data

In a real app this comes from your primary database. That database is the source of truth, and Pinecone is an index over it. You should always be able to rebuild the vectors from the database, never the other way around.

# journal/sample_data.py
TRADES = [
    {"id": "trade-001", "symbol": "NVDA", "emotion": "fear", "pnl": -120.0, "ts": 1767571200,
     "tags": ["panic", "earnings"],
     "text": "Sold NVDA in a panic after a red day. Felt fear and abandoned my stop plan."},
    {"id": "trade-002", "symbol": "AMD", "emotion": "fear", "pnl": -40.0, "ts": 1768000000,
     "tags": ["early-exit"],
     "text": "Closed my chip stock early because I was scared of a pullback."},
    {"id": "trade-003", "symbol": "TSLA", "emotion": "calm", "pnl": 310.0, "ts": 1768500000,
     "tags": ["plan"],
     "text": "Held TSLA to target and followed my plan. Calm and patient the whole time."},
    {"id": "trade-004", "symbol": "NVDA", "emotion": "fomo", "pnl": -85.0, "ts": 1769000000,
     "tags": ["chase"],
     "text": "Chased a breakout on NVDA out of fomo and bought the top."},
    {"id": "trade-005", "symbol": "AAPL", "emotion": "calm", "pnl": 150.0, "ts": 1769500000,
     "tags": ["earnings", "plan"],
     "text": "Traded the AAPL earnings gap exactly as planned, took profit at the first target."},
    {"id": "trade-006", "symbol": "MSFT", "emotion": "greed", "pnl": -60.0, "ts": 1770000000,
     "tags": ["oversized"],
     "text": "Doubled my position size on MSFT because I felt certain. It reversed."},
]


def load_trades() -> list[dict]:
    """Stand-in for 'SELECT * FROM trades WHERE user_id = ...'."""
    return TRADES

Create the index

# journal/create_index.py
from journal import config


def main() -> None:
    if config.pc.indexes.exists(config.INDEX_NAME):
        print("index already exists")
        return

    model = config.pc.indexes.create(
        name=config.INDEX_NAME,
        schema={
            "fields": {
                "_values": {
                    "type": "dense_vector",
                    "dimension": config.DIMENSION,
                    "metric": config.METRIC,
                }
            }
        },
        deployment={"deployment_type": "managed", "cloud": "aws", "region": "us-east-1"},
        # By default this call blocks until the index is ready.
    )
    print("created", model.name, "host:", model.host)


if __name__ == "__main__":
    main()

Run it once with python -m journal.create_index. A few points are worth knowing.

• _values is a reserved field name. In the SDK source it marks the ordinary vector you supply yourself, the kind you pass to upsert(vectors=...) and query(vector=...). Use it for the classic upsert-and-query workflow shown here.
• Dimension and metric are permanent. Changing your embedding model later means creating a new index and re-embedding everything.
• Region matters for latency and data residency, so choose it on purpose.
• Older tutorials show pc.create_index(name=..., dimension=..., metric=..., spec=ServerlessSpec(...)). SDK 10 still accepts that legacy form, so you will meet it in old code. New code should use pc.indexes.create(schema=..., deployment=...).
• Consider deletion_protection="enabled" for anything you cannot afford to lose. It blocks deleting the index until you turn it off.

RAGnetic connects to Pinecone and never creates the index. It calls from_documents(..., index_name=...) and assumes one already exists, and there is no index-creation call anywhere in the app or the CLI. That is a bold assumption to make about infrastructure I had not built. Creating the index explicitly removes a whole class of confusing first-run errors.

Ingest: embed and upsert

Four rules make ingestion safe to run twice.

1. Deterministic IDs. Use the row’s primary key (trade-001), not a random UUID or a timestamp. Re-running ingestion then overwrites instead of duplicating.
2. Skip unchanged records. Store a hash of the embedded text in metadata. On re-ingest, only re-embed records whose hash changed, because embedding costs money and time.
3. Keep useful metadata. Store filterable fields (symbol, date as a number, tags as a list) plus the display text.
4. Batch by size, not just by count. The next section shows why.

# journal/ingest.py
import hashlib

from journal import config, embeddings
from journal.sample_data import load_trades

FETCH_LIMIT = 1000   # Pinecone allows up to 1,000 IDs per fetch request


def content_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]


def _existing_hashes(index, ids: list[str], namespace: str) -> dict[str, str]:
    """Return {id: stored content hash} for the ids that already exist."""
    found: dict[str, str] = {}
    for start in range(0, len(ids), FETCH_LIMIT):
        chunk = ids[start:start + FETCH_LIMIT]
        response = index.fetch(ids=chunk, namespace=namespace)
        for record_id, vector in response.vectors.items():
            found[record_id] = (vector.metadata or {}).get("hash", "")
    return found


def ingest(trades: list[dict], namespace: str) -> int:
    """Embed and upsert only new or changed trades. Returns how many were written."""
    index = config.get_index()
    existing = _existing_hashes(index, [t["id"] for t in trades], namespace)

    changed = [t for t in trades if existing.get(t["id"]) != content_hash(t["text"])]
    if not changed:
        return 0

    vectors = embeddings.embed_texts([t["text"] for t in changed])
    records = [
        {
            "id": trade["id"],
            "values": vector,
            "metadata": {
                "symbol": trade["symbol"],
                "emotion": trade["emotion"],
                "tags": trade["tags"],          # a list of strings is allowed
                "pnl": trade["pnl"],
                "ts": trade["ts"],              # numbers, so we can range-filter
                "text": trade["text"],          # so a hit is immediately displayable
                "hash": content_hash(trade["text"]),
            },
        }
        for trade, vector in zip(changed, vectors)
    ]

    # batch_size counts records, not bytes. 25 is safe at 1,536 dimensions.
    response = index.upsert(vectors=records, namespace=namespace, batch_size=25, show_progress=False)
    if response.has_errors:
        # A partial failure does not raise. Retry what did not land, then fail loudly.
        retry = index.upsert(vectors=response.failed_items, namespace=namespace,
                             batch_size=10, show_progress=False)
        if retry.has_errors:
            raise RuntimeError(f"{retry.failed_item_count} records failed to upsert")
    return len(records)


if __name__ == "__main__":
    written = ingest(load_trades(), namespace="user-alex")
    print(f"upserted {written} records")

Run python -m journal.ingest twice. The second run writes nothing, because every hash matches. I confirmed that hash check works with fetch against the emulator.

Records go in as dicts with id, values and metadata. The SDK also accepts (id, values) and (id, values, metadata) tuples. Metadata allows strings, numbers, booleans and lists of strings, with no nesting. Store dates as numbers, such as a Unix timestamp, so you can filter with $gte and $lte.

Why is batch_size=100 a trap with real embeddings?

The upsert limit is 1,000 records or 2 MB, whichever you hit first. The SDK’s batch_size only counts records. It never looks at bytes.

Bytes are where real embeddings bite. The default Python client sends JSON, so every float travels as text. The OpenAI SDK decodes embeddings into float32 values held as Python floats, and those print as long decimals. I measured one journal record (a unit-length float32 vector plus the metadata from this tutorial) at each size.

At 1,536 dimensions a record is about 33 KB as JSON. A batch of 100 is about 3.3 MB, over the 2 MB limit. At 3,072 dimensions, which Tradion uses, one record is about 66 KB.

The dangerous part is what happens next. With batch_size set, a failed batch does not raise an exception. The SDK reports it on the response as has_errors, errors and failed_items, and the code above has to check for it. Code that skips the check reports success while whole batches sit in failed_items.

Start with roughly half of what your measurement says fits, and tune from there. Measure your own records, because long metadata text changes the answer.

Waiting for freshness in tests

Because of eventual consistency, poll instead of asserting right after a write.

import time
from journal import config


def wait_for_count(namespace: str, expected: int, timeout: float = 30.0) -> None:
    index = config.get_index()
    deadline = time.time() + timeout
    while time.time() < deadline:
        stats = index.describe_index_stats()
        summary = stats.namespaces.get(namespace)
        if summary and summary.vector_count >= expected:
            return
        time.sleep(1)
    raise TimeoutError(f"{namespace} did not reach {expected} records in {timeout}s")

Query

# journal/search.py
from pinecone import Field

from journal import config, embeddings


def search(question: str, namespace: str, top_k: int = 3, metadata_filter: dict | None = None):
    """Return the top_k closest journal entries to `question`."""
    index = config.get_index()
    query_vector = embeddings.embed_texts([question])[0]
    response = index.query(
        vector=query_vector,
        top_k=top_k,
        namespace=namespace,
        filter=metadata_filter,
        include_metadata=True,
    )
    return response.matches


def show(matches) -> None:
    for match in matches:
        meta = match.metadata
        print(f"{match.score:.3f}  {match.id}  [{meta['symbol']}, {meta['emotion']}]  {meta['text']}")


if __name__ == "__main__":
    question = "When do I sell winners out of fear?"

    print("--- plain semantic search")
    show(search(question, "user-alex"))

    print("--- only NVDA, and only fear or fomo")
    nvda_emotional = (Field("symbol") == "NVDA") & Field("emotion").is_in(["fear", "fomo"])
    show(search(question, "user-alex", metadata_filter=nvda_emotional.to_dict()))

You get a list of matches, each with id, score and metadata. With cosine, a higher score means closer. Metadata filters narrow the candidate set before ranking, and they are plain dictionaries:

{"symbol": {"$eq": "NVDA"}}
{"$and": [{"symbol": {"$eq": "NVDA"}}, {"ts": {"$gte": 1767571200}}]}
{"tags": {"$in": ["earnings"]}}      # matches records whose tag list contains "earnings"

Nested dictionaries are easy to get wrong, so the Python SDK ships a small builder, pinecone.Field, that produces the same dictionaries.

(Field("symbol") == "NVDA")                    # {'symbol': {'$eq': 'NVDA'}}
Field("ts").gte(1767571200)                    # {'ts': {'$gte': 1767571200}}
Field("tags").is_in(["earnings"])              # {'tags': {'$in': ['earnings']}}
(cond_a & cond_b) | cond_c                     # combines into $and / $or

Ordering comparisons (gt, gte, lt, lte) are methods and numeric only. Only == and != are overloaded as Python operators.

To grab records you already know by ID, use fetch. IDs that do not exist are simply missing from the result.

response = index.fetch(ids=["trade-001"], namespace="user-alex")
record = response.vectors["trade-001"]
print(record.metadata["text"])

To search several namespaces at once, query_namespaces runs one query per namespace in parallel and merges the results into one ranking.

merged = index.query_namespaces(
    vector=query_vector,
    namespaces=["user-alex", "user-bob"],
    metric="cosine",            # you must pass the index's metric: the merge ranks by it
    top_k=5,
    include_metadata=True,
)

A merged match does not say which namespace produced it. When I ran this with the same trade ID in two namespaces, it came back twice with nothing to tell the copies apart. Put a tag in metadata if you need to know.

Update and delete

index = config.get_index()

# Change metadata only. Fields you don't mention are left alone.
index.update(id="trade-001", set_metadata={"reviewed": True}, namespace="user-alex")

# If the TEXT changes, the meaning changes, so re-embed and upsert.
# (That is what ingest() does when the hash differs.)

# Delete specific records.
index.delete(ids=["trade-002", "trade-004"], namespace="user-alex")

# Empty a namespace but keep it.
index.delete(delete_all=True, namespace="user-alex")

# Remove the namespace entirely.
index.delete_namespace(name="user-alex")

Each of these surprises beginners.

• update applies asynchronously. A read right after can still show the old value.
• Deleting IDs that don’t exist raises no error. The SDK’s own docstring says so: a successful call “is not evidence anything was deleted.” If you need proof, fetch afterwards.
• delete_all keeps the namespace and delete_namespace removes it. Both are irreversible.
• Delete by metadata filter (index.delete(filter=...)) exists in the SDK, but support depends on your index. The emulator refused it, saying serverless and starter indexes do not support deleting with metadata filtering. Verify it on your own index.

A portable alternative is to give records prefixed IDs and list them. Delete requests take up to 1,000 IDs each.

ids = [item.id for page in index.list(prefix="trade-", namespace="user-alex") for item in page.vectors]
for start in range(0, len(ids), 1000):
    index.delete(ids=ids[start:start + 1000], namespace="user-alex")

Can you try it without an API key or a bill?

Pinecone publishes Pinecone Local, an in-memory emulator that runs in Docker. This is the command from Pinecone’s docs.

docker run -d --name pinecone-local -e PORT=5080 -e PINECONE_HOST=localhost \
  -p 5080-5090:5080-5090 --platform linux/amd64 ghcr.io/pinecone-io/pinecone-local:latest

The docs list its limits: 100,000 records per index, nothing persists after a restart, and no backups or hosted inference. It also speaks an older API version than SDK 10, and that matters. Here is what happened when I ran this tutorial’s code against it.

• The data-plane calls worked: batched upsert, query, Field filters, $and and $in filters, fetch, query_namespaces, list and delete by ID. Point the client at Pinecone(api_key="pclocal", host="http://localhost:5080") and use pc.index(host="http://localhost:5081") for the index.
• pc.indexes.create(schema=...) was rejected with a 422 asking for a top-level dimension, so I created the index with a plain HTTP request.
• The SDK could not parse the emulator’s older describe response, so pc.index(name) and pc.indexes.exists fail once the index exists. Passing host= avoids it.
• update applied, but the SDK errored while parsing the emulator’s reply.
• delete_namespace answered 404, and the namespace and its records were still there afterwards.

It is excellent for exercising your own code paths. It is not a substitute for the real service.

The RAG loop: why numbers must come from code

RAG, retrieval-augmented generation, is the pattern behind “chat with your data.” You retrieve the most relevant records, add them to the prompt as context, then let a model generate the answer.

The model never looks anything up. You look things up and paste the results into the prompt, which is why retrieval quality decides answer quality.

Language models are fluent and confident, and one will invent a win rate without blinking. So split the jobs. Code computes every figure: counts, sums, ratios, dates. The model interprets those figures in words. Vectors supply qualitative context, meaning what the user wrote and felt, and never quantities.

# journal/rag.py
import json
import os

from journal import config
from journal.sample_data import load_trades
from journal.search import search

SYSTEM = (
    "You are a trading-journal assistant. Answer using ONLY the statistics and "
    "journal excerpts provided. If they do not contain the answer, say so plainly. "
    "Never invent numbers."
)


def compute_stats(trades: list[dict]) -> dict:
    """Exact figures, computed in code from the source-of-truth rows."""
    total = len(trades)
    wins = [t for t in trades if t["pnl"] > 0]
    by_emotion: dict[str, dict] = {}
    for t in trades:
        bucket = by_emotion.setdefault(t["emotion"], {"trades": 0, "total_pnl": 0.0})
        bucket["trades"] += 1
        bucket["total_pnl"] = round(bucket["total_pnl"] + t["pnl"], 2)
    return {
        "trades": total,
        "win_rate": round(len(wins) / total, 3) if total else None,
        "total_pnl": round(sum(t["pnl"] for t in trades), 2),
        "by_emotion": by_emotion,
    }


def build_prompt(question: str, matches, stats: dict) -> str:
    excerpts = "\n".join(
        f"[{i}] ({m.metadata['symbol']}, {m.metadata['emotion']}) {m.metadata['text']}"
        for i, m in enumerate(matches, start=1)
    )
    return (
        f"STATISTICS (computed from the database):\n{json.dumps(stats, indent=2)}\n\n"
        f"JOURNAL EXCERPTS (most relevant first):\n{excerpts}\n\n"
        f"QUESTION: {question}"
    )


def ask(question: str, namespace: str) -> str:
    matches = search(question, namespace, top_k=4)
    stats = compute_stats(load_trades())
    response = config.openai_client.chat.completions.create(
        model=os.environ["CHAT_MODEL"],
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": build_prompt(question, matches, stats)},
        ],
    )
    return response.choices[0].message.content


if __name__ == "__main__":
    print(ask("When do I sell winners out of fear, and what does it cost me?", "user-alex"))

A few notes on this file.

• Chat model names change often. Take CHAT_MODEL from your provider’s current documentation, not from a tutorial. Use the provider’s streaming option for a chat-like feel.
• Follow-up questions need the conversation history sent with each request. Keep all of it until it nears the model’s context window, then use the provider’s recommended context-management approach instead of inventing your own truncation.
• Retrieved text is untrusted input. If users can write text that ends up in a prompt, like a journal note saying “ignore previous instructions,” treat it as data. Delimit it clearly, and never give the model powers you would not give the author of that text.
• Say when data is missing. If retrieval finds nothing relevant, the honest answer is “I don’t have that in your journal.” The system prompt above asks for exactly that.

Every AI feature in Tradion follows this shape: question, relevant data, model, streamed answer. Prices, returns and ratios are computed in code or fetched from data sources, then handed to the model to interpret. Charts render from real values, never from numbers the model typed. Vector search covers only the “what has this user said and done before” part. I hold the same rule for everything I build with AI, not just Tradion.

How do you design a Pinecone vector database for a real app?

You now know the API. The harder skill is design. These are the decisions that mattered, with Tradion’s real choices and RAGnetic’s real mistakes.

How should you split namespaces?

The rule is that a user must never see another user’s vectors. Pinecone gives you two tools.

• A namespace per tenant is strong, simple isolation. A query physically cannot cross namespaces.
• A metadata filter ({"user_id": {"$eq": "..."}}) depends on every code path remembering to add it. One forgotten filter is a data leak.

Namespaces are the safer default, and they help with cost, since a query is priced by the size of the namespace it hits.

Tradion uses one namespace per user per kind of data: memory, trades, post-trade reviews, analyses, alerts, conversations and broker positions each get their own, named {kind}-{userId}. Retrieval queries the context namespaces in parallel and merges by score. Erasing a user, or resetting just their memory, becomes a matter of deleting namespaces.

There is a tradeoff to know before you copy that. Seven namespaces per user multiplies fast against the per-index caps in the limits table. A simpler design is one namespace per user plus a type metadata field, filtering by type when you need to. Per-kind namespaces buy you an independent top-k per kind, so every kind gets a voice in the merged results. Decide on purpose.

RAGnetic did the opposite. Its permanent Pinecone store is opened by index name alone, with no namespace and no metadata filter, so isolation depends on which index name each agent’s config points to. Isolation that depends on convention breaks the first time someone forgets the convention. Add tenancy on day one, because retrofitting it means rewriting every record.

One more detail: records with no namespace go to the default namespace, the empty string. Tradion keeps its shared, non-user data there, such as the code-generation references every user’s request may need. Records are told apart by a type metadata field and queried with a filter like {"type": {"$eq": "example"}}.

Where should the text live?

You have two ways to get from a vector hit back to displayable content.

Approach	Pros	Cons
Text in metadata	One round trip, simple code	Costs storage and write units. Duplicates data. 40 KB per-record cap. Readable by anyone holding the API key
Only the ID in Pinecone, text in your database	One source of truth. Edits and deletes happen in one place	An extra database lookup per query, and IDs that no longer exist to handle

Tradion does both, on purpose. For most context types the readable summary sits in metadata and goes straight into the prompt. For its per-user memory entries, the vector search returns IDs only (include_metadata=False), and the full, current rows load from Postgres, so the database stays authoritative.

Either way, treat Pinecone metadata as plaintext that anyone with your API key can read. Do not store secrets or personal data you do not need.

Is top-k enough, or do you need re-ranking?

The nearest vector is not always the best answer, so real systems add their own ranking after the search. Tradion has two retrieval paths that do different jobs.

1. Context retrieval: query each context namespace for its top 3, merge, sort by similarity score and keep the best 6. No score threshold.
2. Memory retrieval: fetch the top 8, drop anything below a cosine floor of 0.30, then re-rank by blending three signals: 0.7 × similarity + 0.2 × confidence + 0.1 × recency. Recency decays with the time since the memory was last used and never drops below 0.3. If the vector service is down, it falls back to keyword matching.

Here is the shape of that blend. The 90-day window is an example value.

def recency(days_since_use: float) -> float:
    return max(0.3, 1 - days_since_use / 90)

def blended_score(similarity: float, confidence: float, days_since_use: float) -> float:
    return 0.7 * similarity + 0.2 * confidence + 0.1 * recency(days_since_use)

# A fresher, higher-confidence memory beats a slightly more similar stale one:
print(round(blended_score(0.62, 0.90, 5), 3))    # 0.708
print(round(blended_score(0.66, 0.40, 80), 3))   # 0.572

Call this custom re-ranking or score blending. “Hybrid search” means something more specific, which is combining dense semantic search with sparse keyword search. Pinecone also offers hosted reranking models, a slower, more accurate second pass over your top results, if you would rather not hand-tune weights.

How do you keep vectors in sync with your database?

Every write path that changes a source row must change its vector too.

Event in your app	Vector action
Row created	Embed, upsert
Row text edited	Re-embed, upsert (same ID overwrites)
Row metadata edited	update(set_metadata=...) or re-upsert
Row deleted	Delete the vector by ID, in the same code path
User deletes account	Delete the user’s namespaces

The delete rows are the ones people forget. A deleted trade whose vector survives keeps showing up in search results and in AI answers. That is a bug, and if it is someone’s personal data it is also a privacy problem.

RAGnetic forgot them twice. Its “delete agent” endpoint removes the database rows and the local vector-store folder, and nothing in the app deletes the remote vectors, so they stay in the index forever. It is the software equivalent of moving out and leaving the furniture.

The second miss is quieter. Its chunk IDs are derived from content hashes, which makes upserts idempotent, but nothing removes chunks that vanish from the source. Upsert handles additions and changes. Only an explicit delete handles removals.

# journal/maintenance.py
import logging

from pinecone import NotFoundError, PineconeError

from journal import config

log = logging.getLogger(__name__)


def delete_trade(db, trade_id: str, namespace: str) -> None:
    """Delete the row AND its vector together."""
    db.delete_trade_row(trade_id)                       # your database call
    config.get_index().delete(ids=[trade_id], namespace=namespace)


def erase_user(namespaces: list[str]) -> bool:
    """Right-to-erasure: FAIL CLOSED. True only if every namespace is confirmed gone."""
    # "Not found" can mean the INDEX is missing (wrong name, wrong project, deleted),
    # not just that a namespace is already gone. Never read that as "erased".
    if not config.pc.indexes.exists(config.INDEX_NAME):
        log.error("index %s not found, cannot confirm erasure", config.INDEX_NAME)
        return False

    index = config.get_index()
    all_gone = True
    for ns in namespaces:
        try:
            index.delete_namespace(name=ns)
        except NotFoundError:
            pass                       # namespace already gone: deleting twice must be safe
        except PineconeError:
            log.exception("could not erase namespace %s", ns)
            all_gone = False           # anything else: report failure, retry later

    # Confirm, don't assume. Stats are eventually consistent, so a False here
    # just means "try again on the next run".
    still_there = index.describe_index_stats().namespaces
    return all_gone and not any(ns in still_there for ns in namespaces)

Tradion treats erasure the same way. Resetting a user’s AI memory and deleting an account both erase that user’s namespaces. The account-deletion job only removes the account record after every external erasure, vectors and stored chart images included, has succeeded. If any step fails it keeps the account, reports the failure and retries on the next run.

Treating “not found” as “already erased” makes the operation safe to repeat, but only once you know the index itself exists. I hit the trap while testing. The emulator answered delete_namespace with a 404 for an operation it does not support, and the records were still there. A naive “404 means gone” rule would have reported the data erased when it was not.

Search does the opposite. It fails open: if the vector service is unreachable, the assistant still answers, just without personalization.

What happens when your embedding logic changes?

Change the embedding model, the dimension or how you build the text, and every old vector is inconsistent with new queries. Plan for it.

1. Create a new index, or a new namespace, for the new setup.
2. Re-embed everything from the source of truth into it.
3. Switch reads over, then delete the old one.

To detect whether source content changed, hash the inputs. The catch is that the hash must live somewhere that survives deployments, like your database or the vectors’ own metadata, as in ingest.py. A hash file on a container’s local disk vanishes on the next deploy, and your “skip if unchanged” check then re-embeds everything or wrongly decides nothing changed.

What should you check before going to production?

How should each feature fail?

Every network call can fail, so decide per feature whether it fails open or closed.

• Search for personalization: fail open. Log it, return no matches and let the app work without them.
• Erasure and compliance: fail closed. Report failure and retry. Never claim data is gone when you did not confirm it.
• Writes that must not be lost: put them on a queue with retries and idempotent IDs, so a retry is harmless.

The SDK helps. It retries transient failures for you, three retries by default with backoff, configurable through RetryConfig, and every data-plane call takes a timeout= argument. Beyond that, respect rate limits and put a hard time budget on anything in a user-facing request so a slow vector query cannot hang your page.

import logging
from pinecone import PineconeError

def safe_search(question: str, namespace: str):
    try:
        return search(question, namespace)
    except PineconeError:
        logging.getLogger(__name__).warning("vector search unavailable", exc_info=True)
        return []       # fail open

Never present partial or failed output as complete. If retrieval failed and you answered anyway, consider telling the user the answer had less context.

How do you measure retrieval quality?

Most RAG disappointment is a retrieval problem, and you cannot fix what you do not measure. Build a tiny evaluation set first: real questions paired with the record ID that should come back.

# journal/evaluate.py
from journal.search import search

CASES = [
    {"question": "When did I panic and sell?",          "expected_id": "trade-001"},
    {"question": "Trades where I followed my plan",      "expected_id": "trade-003"},
    {"question": "Times I chased a breakout",            "expected_id": "trade-004"},
]


def hit_rate(namespace: str, k: int = 3) -> float:
    """Fraction of questions whose expected record appears in the top k."""
    hits = 0
    for case in CASES:
        ids = [m.id for m in search(case["question"], namespace, top_k=k)]
        hits += case["expected_id"] in ids
    return hits / len(CASES)


if __name__ == "__main__":
    print(f"hit rate @3: {hit_rate('user-alex'):.2f}")

Use it to compare chunk sizes, embedding models, dimensions and re-ranking weights with evidence instead of vibes.

RAGnetic’s evaluation code compared the ground-truth chunk ID, a content hash, against retrieved IDs that were database integers. The two ID spaces could never match, so a strict retrieval metric read zero on every question. I wrote up the whole story in my post on the project. The fix: use one ID everywhere. The ID in your vector store should be the same one in your database, your evaluation set and your citations.

Where does the money go?

From most to least impactful:

1. Namespace size drives read cost, so partition by tenant.
2. Dimension: fewer dimensions mean less storage and smaller requests. Try dimensions= on newer OpenAI embedding models and measure recall.
3. Metadata size: every byte is stored and written. Keep only what you filter on or display.
4. Re-embedding: do not re-embed unchanged content. Hash it.
5. Delete-and-recreate: do not, when an upsert will do.

top_k, include_metadata and include_values are not levers, since they do not change the read cost. They do change response size and latency, so leave include_values off unless you need the numbers.

What is the security checklist?

• Keep API keys in a secrets manager or environment variables, never in git.
• Use one key per environment, ideally with the narrowest role that works.
• Assume anything in metadata is readable by anyone who obtains the key.
• Enforce tenant isolation in code with a namespace per tenant, not by convention.
• Test your erasure path. “We delete on request” is a claim you should be able to demonstrate.
• Never load untrusted files with unsafe deserializers.

That last one is personal. RAGnetic’s local FAISS store loads with allow_dangerous_deserialization=True, in five places. The flag has the word “dangerous” in its name, and I typed True five times anyway. Hosted stores like Pinecone avoid this class of risk, because you never load a pickle file at all.

Does your wrapper limit your SDK version?

If you use an integration package, it pins the Pinecone SDK version it supports. At the time of writing, the latest langchain-pinecone on PyPI requires pinecone below version 8, while the SDK itself is at 10. It also still exports the vector store class under the old Pinecone name, which is marked deprecated in favor of PineconeVectorStore. RAGnetic imports that old name.

Using the official SDK directly, as this tutorial does, keeps you on current features. If you do use a wrapper, know what it hides: batch sizes, namespace defaults and how it stores text in metadata.

What else can Pinecone do, and when do you not need it?

A few features go beyond the basics.

• Sparse and hybrid search. Dense vectors capture meaning but can miss exact terms like a ticker, an error code or a rare product name. Sparse vectors capture weighted keywords, and hybrid search combines both.
• Full-text search. The SDK’s index schema also accepts string fields with full_text_search for keyword matching without a model.
• Reranking. pc.inference.rerank(model=..., query=..., documents=..., top_n=...) rescoring your top 20 candidates can lift quality noticeably. Check pc.inference.list_models() for current model names.
• Integrated embedding. pc.indexes.create_for_model(...) binds an index to a hosted embedding model, so you upsert raw text with index.upsert_records(...). Fewer moving parts, but you are tied to the models Pinecone hosts.

Now the heretical part. You may not need a vector database at all.

• Fewer than tens of thousands of vectors in one process: a NumPy array and a cosine function is exact, free and fast.
• You already run PostgreSQL at modest scale: the pgvector extension keeps vectors next to your relational data and lets you join and filter in SQL.
• You need exact keyword lookups: use a search engine or full-text search.

A managed vector database earns its keep when you have millions of vectors, many tenants, low-latency requirements and no wish to operate the infrastructure.

(Optional) Using Claude Code? Give it these Pinecone rules first

Skip this section if you don’t use Claude Code. Nothing after it depends on it, and the cheat sheet comes next.

A coding agent writes what its training data remembers, and that can be an older SDK than the one you installed. The langchain-pinecone pin above is the same problem in miniature: code that was right for pinecone below version 8 and wrong for 10. So tell the agent the version and the traps before it types anything.

In Claude Code, the place for that is CLAUDE.md, the rules file it reads at the start of every session. This snippet restates the traps from this post. Adjust the names to your project.

## Pinecone rules
- Use the official `pinecone` SDK, version 10. Check docs.pinecone.io before writing any call.
- One namespace per tenant. Never filter by tenant in metadata instead.
- Upsert batches are limited by bytes (2 MB), not just record count. Measure real records first.
- After every batched upsert, check `has_errors` and retry `failed_items`. A failed batch does not raise.
- Keep the content hash in the vector's metadata, never in a local file.
- Every number shown to a user is computed in code and passed to the model, never typed by the model.
- Retrieved text is untrusted input. Delimit it in the prompt.
- Search fails open. Erasure fails closed, and a 404 does not prove data is gone.

Now try it yourself. Below are three prompts to run against the journal project you built in this tutorial. Each one is also a small lesson: run it, then check what the agent did against the section it links to. If the agent’s answer disagrees with the section, one of them is wrong, and finding out which is the learning.

The first is an upsert audit, from the batch size trap.

Audit the upsert code in journal/ingest.py. Measure the JSON size of five real records at our
embedding dimension, then pick a batch_size that keeps each request under 1 MB. Make sure a
failed batch is retried and then raises, so ingest never reports success with failed_items left.
Show me the measurements before you change anything.

Check that it measured before it changed anything, and that the retry path still looks at has_errors. Then compare its byte counts with the chart earlier in this post.

The second replaces tests that assert straight after a write, which fail because of eventual consistency.

Find every test that queries Pinecone right after an upsert. Replace the immediate assertion with
a poll on describe_index_stats() that times out after 30 seconds and raises a clear error.
Do not add sleep() calls.

Check that the timeout raises instead of passing quietly, and that the poll reads the namespace it wrote to.

The third plans a change of embedding model, following the migration steps.

We are moving from text-embedding-3-small to a different embedding model. Write a plan first, no code:
a new namespace, re-embedding from the database as the source of truth, switching reads, then
deleting the old namespace. Keep the content hash in vector metadata. Tell me what could
go wrong at each step.

Check that the plan deletes the old namespace last, after reads have switched. Then run evaluate.py against both namespaces before you trust the new one.

These prompts are starting points I wrote from this tutorial, not recipes I have benchmarked. Read the diff and compare hit rates before you merge anything. An agent that writes vector code nobody reviews is vibe coding with extra steps, and an automated code review is a cheap second pair of eyes on what it wrote.

SDK 10 cheat sheet and three exercises

Every call below appears earlier in this post. Keep it next to your editor.

from pinecone import Pinecone, Field
pc = Pinecone()                                   # PINECONE_API_KEY from env

# Control plane
pc.indexes.create(name=..., schema={"fields": {"_values": {"type": "dense_vector", "dimension": 1536, "metric": "cosine"}}},
                  deployment={"deployment_type": "managed", "cloud": "aws", "region": "us-east-1"})
pc.indexes.exists(name); pc.indexes.describe(name); pc.indexes.delete(name)

# Data plane
index = pc.index(name)                            # or pc.index(host=...)
index.upsert(vectors=[{"id": "a", "values": [...], "metadata": {...}}], namespace="ns", batch_size=25)
index.query(vector=[...], top_k=5, namespace="ns", filter={...}, include_metadata=True)
index.query_namespaces(vector=[...], namespaces=[...], metric="cosine", top_k=5)
index.fetch(ids=["a"], namespace="ns")
index.update(id="a", set_metadata={"k": "v"}, namespace="ns")
index.delete(ids=["a"], namespace="ns")           # also: delete_all=True, filter=...
index.delete_namespace(name="ns")
index.list(prefix="trade-", namespace="ns")       # iterate pages of IDs
index.describe_index_stats()                      # .namespaces[ns].vector_count

Then try these.

1. Add a date filter. Give search() a since parameter that filters ts >= since, using Field("ts").gte(...).
2. Prove idempotency. Run ingest twice and use describe_index_stats() to show the record count did not change. Then edit one trade’s text and show that exactly one record is rewritten.
3. Measure a dimension cut. Grow CASES in evaluate.py to 20 questions, then compare text-embedding-3-small at 1,536 dimensions against the same model at 512 (the dimensions parameter) using hit_rate.

Vector database APIs and model names change quickly. Before you build on any snippet here, check the docs for the version you install. The Pinecone docs and the Python SDK reference are the places to start.

What is the first pile of notes, tickets or logs in your own life that you wish you could search by meaning instead of by keyword?