tidevec

Python SDK for TideVec — the world's first temporally-aware, causally-indexed vector database.

Install

pip install tidevec

Quick Start

import os
from tidevec import TideVec, HalfLife

# Auth: pass api_key or set TIDEVEC_API_KEY env var
db = TideVec("localhost:6399", api_key=os.environ.get("TIDEVEC_API_KEY", ""))

db.create_collection(
    "docs",
    dim=768,
    half_life_ms=HalfLife.ONE_WEEK,
    temporal_blend=0.3,
)

db.upsert("docs", [
    {"id": "doc_001", "embedding": [...], "payload": {"source": "wiki"}},
])

results = db.search("docs", query_vector=[...], top_k=10)
for hit in results:
    print(f"{hit.id}  score={hit.score:.4f}  temporal={hit.temporal_score:.3f}")

Authentication

Production TideVec requires an API key on all /v1/* routes:

# Option 1: environment variable (recommended)
export TIDEVEC_API_KEY="your-secret-key"
db = TideVec("localhost:6399")  # reads env automatically

# Option 2: explicit
db = TideVec("localhost:6399", api_key="your-secret-key")

# Option 3: TLS
db = TideVec("localhost:6399", api_key="...", tls=True)

Errors:

from tidevec import UnauthorizedError, ForbiddenError, RateLimitError

try:
    db.search("docs", query)
except UnauthorizedError:   # 401 — missing/invalid API key
    ...
except ForbiddenError:      # 403 — tenant or SSRF blocked
    ...
except RateLimitError:       # 429 — auto-retried up to 3 times
    ...

DriftBridge — Model Migration

Upgrade embedding models without downtime:

import time

db.start_drift(
    "docs",
    reembed_url="https://embed.example.com/v1/embed",
)

while True:
    status = db.drift_status("docs")
    print(f"{status.phase}: {status.pct_complete:.0f}%")
    if status.phase in ("COMPLETE", "IDLE", "FAILED"):
        break
    time.sleep(5)

# Abort if needed
db.abort_drift("docs")

Backups

snapshot = db.trigger_backup()   # "tidevec_1712345678.tar.gz"
backups  = db.list_backups()     # list all snapshots

Observability

# Prometheus metrics
print(db.metrics())

# Per-query trace (OTel-compatible)
results = db.search("docs", query, include_trace=True)
if results.trace:
    print(results.trace["strategy"], results.trace["latency_ms"])

Async

import asyncio
from tidevec import AsyncTideVec

async def main():
    async with AsyncTideVec("localhost:6399", api_key="...") as db:
        await db.upsert("docs", [{"id": "v1", "embedding": [...]}])
        results = await db.search("docs", query_vector=[...], top_k=5)

asyncio.run(main())

API Reference

HalfLife presets

from tidevec import HalfLife

HalfLife.ONE_HOUR   # agent session memory
HalfLife.ONE_DAY    # news / feeds
HalfLife.ONE_WEEK   # support tickets
HalfLife.ONE_MONTH  # documents (default)
HalfLife.ONE_YEAR   # long-term knowledge base