wmilvus - Enterprise Type-Safe Milvus Vector Database Client Wrapper

Enterprise Features

Why Choose wmilvus?

Pydantic ORM Data Models

Define collection schemas as standard Pydantic models with vector annotations (FieldVector, MetricType.COSINE, IndexType.HNSW).

KNN & Range Similarity Search

Perform top-k similarity search (search_similar) or radius threshold search (search_range) returning typed models or distance scores.

Bulk Ingestion with Auto-Chunking

Insert thousands of vector embeddings (insert_batch) with automatic payload chunking to prevent gRPC size limits.

Non-Blocking Async Client

Native AsyncWMilvus client with async with context manager support for high-throughput asyncio and FastAPI apps.

Enterprise Ghost Audit Log

Automatic auditing (_forensic_audit_log) tracking INSERT, UPDATE, and DELETE operations across all forensic models.

Collection Schema Verification

Validate existing Milvus collection vector dimensions and field constraints (verify_schema) before executing queries.

Code Showcase

Interactive Usage Examples

from typing import List
from pydantic import BaseModel
from wmilvus import FieldVector, MetricType, WMilvus

class UserFace(BaseModel):
    id: str
    name: str
    face_embedding: List[float] = FieldVector(dim=128, metric_type=MetricType.COSINE)

# Context manager handles connection cleanup automatically
with WMilvus(UserFace, uri="http://localhost:19530") as db:
    user = UserFace(id="u_001", name="William Rodriguez", face_embedding=[0.1] * 128)
    db.insert(user)
    
    matches = db.search_similar(vector=[0.1] * 128, top_k=5)
    print(f"Top match: {matches[0].name}")

# Search similarity with distance scores
matches_with_scores = db.search_similar_with_scores(
    vector=[0.92, 0.08, 0.01, 0.05],
    top_k=2,
    filter_expr='category == "VIP"'
)

for match in matches_with_scores:
    print(f"ID={match.id}, Score={match.distance:.4f}, Name={match.metadata.get('person_name')}")

# Search vectors with COSINE similarity >= 0.95 threshold
range_matches = db.search_range_with_scores(
    vector=[0.91, 0.09, 0.01, 0.09],
    radius=0.95,
    top_k=5
)

for match in range_matches:
    print(f"Match ID={match.id}, Similarity Score={match.distance:.4f}")

# Bulk insert 2,500 vector records with chunking
records = [
    FrameEmbedding(id=f"frame_{i:05d}", vector=np.random.rand(128).tolist())
    for i in range(2500)
]

result = db.insert_batch(records, batch_size=1000)
print(f"Ingested {result['inserted_count']} records across {result['batches_processed']} batches")

import asyncio
from wmilvus import AsyncWMilvus, FieldVector, MetricType

async def main():
    async with AsyncWMilvus(TelemetryModel, uri="http://localhost:19530") as db:
        await db.insert(TelemetryModel(id="t1", vector=[0.5] * 64))
        results = await db.search_similar(vector=[0.5] * 64, top_k=1)
        print(results[0].id)

asyncio.run(main())

class BankAccountVector(ForensicModel):
    id: str
    holder_name: str
    balance: float
    account_vec: List[float] = FieldVector(dim=128)

# Automatically records entries in _forensic_audit_log collection
db.insert(account, user_id=100)
db.update("1", updated_account, user_id=200)

logs = db.get_ghost_audit_log()
print(f"Audit log entries: {len(logs)}")

Component	Technology	Description
Language	Python 3.9+	Modern Python with strict typing support
Vector Engine	Milvus 2.3+	Open-source cloud-native vector database
SDK Core	pymilvus 2.3+	Official high-performance Milvus Python client
Data Validation	Pydantic 2.x	Type hinting and schema generation
Numerical Core	NumPy	Array vector conversions and operations
Logging	Loguru	Structured diagnostic logging

Type-Safe Milvus Vector ORM for Python