FluxVector Python SDK

Semantic search in 4 lines. No OpenAI key. No embedding pipelines. Just text in, results out.

pip install fluxvector

from fluxvector import FluxVector

fv = FluxVector(api_key="fv_live_...")
fv.collections.create("docs")
fv.vectors.upsert("docs", [{"id": "1", "text": "Your text here"}])
results = fv.search("docs", "find similar content")

That's it. FluxVector embeds your text server-side with multilingual models (e5-large, BGE-M3). No OpenAI API key, no embedding code, no vector math.

vs Pinecone: Pinecone requires you to generate embeddings yourself (usually via OpenAI at $0.13/1M tokens), then send raw vectors. FluxVector does it all in one call.

Why FluxVector

• Built-in embeddings — send text, get search results. No external embedding API needed.
• Hybrid search — vector + BM25 keyword scoring combined, zero config.
• Self-hostable — one Docker image, your server, your data never leaves.
• Simple — 20 files, not 20,000. One endpoint to upsert, one to search.

Installation

pip install fluxvector

Quick Start

from fluxvector import FluxVector

fv = FluxVector(api_key="fv_live_abc123")

# Create a collection (embeddings handled automatically)
fv.collections.create("products")

# Just send text — FluxVector embeds it for you
fv.vectors.upsert("products", [
    {"id": "p1", "text": "Red running shoes", "metadata": {"price": 89, "brand": "Nike"}},
    {"id": "p2", "text": "Blue hiking boots", "metadata": {"price": 149, "brand": "Merrell"}},
    {"id": "p3", "text": "White tennis sneakers", "metadata": {"price": 65, "brand": "Adidas"}},
])

# Semantic search — returns results ranked by meaning, not keywords
results = fv.search("products", "comfortable shoes for running", top_k=5)
for r in results:
    print(f"{r.id}: {r.score:.2f} — {r.text}")

# Filter by metadata
results = fv.search("products", "shoes", filter={"price": {"$lt": 100}})

Async Support

from fluxvector import AsyncFluxVector

async with AsyncFluxVector(api_key="fv_live_abc123") as fv:
    results = await fv.search("products", "comfortable shoes")
    for r in results:
        print(f"{r.id}: {r.score:.2f}")

Configuration

fv = FluxVector(
    api_key="fv_live_abc123",       # or set FLUXVECTOR_API_KEY env var
    base_url="https://custom.host", # default: https://fluxvector.dev
    timeout=30.0,                   # request timeout in seconds
    max_retries=3,                  # retries on 429 / 5xx with exponential backoff
)

API Reference

Collections

# Create
col = fv.collections.create("products", dimension=1024, metric="cosine", description="Product catalog")

# List (cursor pagination)
page = fv.collections.list(limit=10)
for col in page:
    print(col.name)
# Next page
if page.has_more:
    next_page = fv.collections.list(cursor=page.next_cursor)

# Get
col = fv.collections.get("products")

# Delete
fv.collections.delete("products")

Vectors

# Upsert (auto-chunks at 1000 vectors per request)
fv.vectors.upsert("products", [
    {"id": "p1", "text": "Red shoes", "metadata": {"price": 89}},
    {"id": "p2", "text": "Blue hat", "values": [0.1, 0.2, ...]},  # raw vector
])

# Query by text
results = fv.vectors.query("products", text="shoes", top_k=10)

# Query by raw vector
results = fv.vectors.query("products", vector=[0.1, 0.2, ...], top_k=5)

# Query with filter
results = fv.vectors.query(
    "products",
    text="shoes",
    filter={"brand": {"$in": ["Nike", "Adidas"]}},
    include_metadata=True,
    include_text=True,
)

# Fetch by IDs
vectors = fv.vectors.fetch("products", ["p1", "p2"])

# Delete by IDs
fv.vectors.delete("products", ids=["p1", "p2"])

# Delete by filter
fv.vectors.delete("products", filter={"brand": {"$eq": "discontinued"}})

Search

The signature method — one-line semantic search:

results = fv.search("products", "comfortable running shoes", top_k=10)
for r in results:
    print(f"{r.id}: {r.score:.4f} — {r.text}")
    print(f"  metadata: {r.metadata}")

Embeddings

# Single text
resp = fv.embeddings.create("Hello world")
print(resp.embedding)  # [0.012, -0.034, ...]
print(resp.dimension)  # 1024

# Batch
resp = fv.embeddings.batch(["Hello", "World", "Foo"])
for emb in resp.embeddings:
    print(len(emb))  # 1536

API Keys

# Create
key = fv.api_keys.create("Production Key", env="live")
print(key.key)  # fv_live_... (only shown once)

# List
keys = fv.api_keys.list()
for k in keys:
    print(f"{k.name}: {k.prefix}...")

# Rename
fv.api_keys.update("key_id", name="New Name")

# Delete
fv.api_keys.delete("key_id")

Usage

# Current usage
usage = fv.usage.get()
print(f"Plan: {usage.plan}")
print(f"Requests: {usage.requests}")
print(f"Vectors stored: {usage.vectors_stored}")

# Historical usage
history = fv.usage.history(days=30)
for day in history:
    print(f"{day.date}: {day.requests} requests")

Filter Operators

Use metadata filters with any search or query method:

Error Handling

from fluxvector import FluxVector, FluxVectorError, AuthenticationError, RateLimitError, NotFoundError

fv = FluxVector(api_key="fv_live_abc123")

try:
    results = fv.search("products", "shoes")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited: {e.message}")
except NotFoundError:
    print("Collection not found")
except FluxVectorError as e:
    print(f"API error {e.status_code}: {e.message}")

Environment Variables

License

MIT