Managing Data Quality
1. Validating Vector Data
Example: Validate
import math
def valid(v):
return (len(v) == DIM
and all(isinstance(x, float) for x in v)
and not any(math.isnan(x) or math.isinf(x) for x in v))2. Detecting Duplicate Vectors
Example: Cosine Threshold
def is_dup(qv, threshold=0.99):
r = index.query(vector=qv, top_k=1)
return r["matches"] and r["matches"][0]["score"] >= threshold3. Handling Missing Embeddings
| Source | Action |
|---|---|
| Embedding API failure | Retry with backoff; DLQ on permanent fail |
| Empty input text | Skip; don't upsert |
4. Verifying Upsert Success
Example: Verify by Fetch
r = index.upsert(vectors=batch)
ids = [v["id"] for v in batch]
fetched = index.fetch(ids=ids)["vectors"]
assert len(fetched) == len(ids)5. Auditing Vector Counts
Example: Compare to Source
src = db.count("documents")
idx = index.describe_index_stats()["total_vector_count"]
assert idx == src, f"Drift: {src - idx}"6. Detecting Data Drift
| Signal | Detection |
|---|---|
| Embedding model change | Score distribution shifts |
| Stale content | Timestamps grow old |
| Bad upserts | Counts diverge from source |
7. Maintaining Data Consistency
Consistency Recipe
- Source DB is ground truth.
- Use CDC / outbox to enqueue change events.
- Worker upserts/deletes in Pinecone idempotently.
- Nightly reconcile job compares counts + samples.
8. Cleaning Stale Data
Example: TTL Cleanup
cutoff = int(time.time()) - 90 * 86400
index.delete(filter={"created_at": {"$lt": cutoff}})9. Implementing Data Validation Pipelines
| Stage | Check |
|---|---|
| Pre-embed | Text non-empty, language detect |
| Post-embed | Dim correct, no NaN |
| Post-upsert | Fetch sample, verify |
10. Monitoring Data Quality Metrics
| Metric | Why |
|---|---|
| % failed embeddings | Upstream issues |
| Score distribution | Catch model drift |
| Recall@k on eval set | Retrieval quality |