langchain-luma 1.2.3

Python Client SDK for Luma (RustKissVDB)

langchain-luma

langchain-luma is a production-ready Python client SDK for Luma (RustKissVDB), a high-performance, lightweight multi-model database written in Rust. Beyond vector storage, Luma supports Document Store, SQL (SQLite-compatible), Key-Value State Management, and Real-time Streams. This library provides both a direct HTTP client for low-level interaction and a fully compliant VectorStore implementation for seamless integration with the LangChain ecosystem.

System Architecture

The following diagram illustrates how the Python client interacts with the Luma Rust backend and LangChain workflows.

graph TD
    subgraph "Python Environment"
        UserCode[User Application / RAG Pipeline]
        LC[LangChain Integration]
        SDK[LumaClient SDK]
    end

    subgraph "Infrastructure"
        Server[Luma Server - Rust Binary]
        Storage[(Vector Storage)]
    end

    UserCode -->|Direct API Calls| SDK
    UserCode -->|Via VectorStore| LC
    LC -->|Wraps| SDK
    SDK -->|HTTP/REST| Server
    Server -->|Read/Write| Storage

Prerequisites: Luma Backend

This SDK acts as a client. To function, it requires a running instance of the Luma server (v0.2.2 or higher).

1. Download the Server Binary: Download the appropriate executable for your operating system from the Official Release Assets.

• Linux: luma-linux-amd64
• Windows: luma-windows-amd64.exe
• macOS: luma-macos-amd64

2. Start the Server: Run the binary in a terminal window. By default, it listens on port 1234.

# Linux/macOS
chmod +x luma-linux-amd64
./luma-linux-amd64

# Windows
.\luma-windows-amd64.exe

Installation

Install the package directly from PyPI.

Standard Client

For users who only need the direct API client without LangChain dependencies:

pip install langchain-luma

With LangChain Support

For users building RAG pipelines or using LangChain abstractions:

pip install "langchain-luma[langchain]"

Multi-Model SDK Usage

The LumaClient class provides granular control over the database operations. It is organized into namespaces (system, vectors, documents, sql, state, stream) for clarity.

import time
import logging
from langchain_luma import LumaClient

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration
LUMA_URL = "http://localhost:1234"
COLLECTION_NAME = "enterprise_docs"
VECTOR_DIMENSION = 384  # Must match your embedding model

def run_pipeline():
    # 1. Initialize Client
    client = LumaClient(url=LUMA_URL)

    # 2. Health Check
    try:
        health = client.system.health()
        logger.info(f"System Status: {health}")
    except Exception as e:
        logger.error(f"Failed to connect to Luma at {LUMA_URL}: {e}")
        return

    # 3. Create Collection
    # This defines the schema for the vector space.
    try:
        client.vectors.create_collection(
            name=COLLECTION_NAME,
            dim=VECTOR_DIMENSION,
            metric="cosine"
        )
        logger.info(f"Collection '{COLLECTION_NAME}' created successfully.")
    except Exception as e:
        logger.warning(f"Collection creation skipped (might already exist): {e}")

    # 4. Upsert Data
    # Simulating a document vector with metadata
    vector_id = "doc_ref_1024"
    embedding_vector = [0.05] * VECTOR_DIMENSION
    metadata = {
        "source": "internal_wiki",
        "author": "dev_ops",
        "created_at": time.time()
    }

    client.vectors.upsert(
        collection=COLLECTION_NAME,
        id=vector_id,
        vector=embedding_vector,
        meta=metadata
    )
    logger.info(f"Vector '{vector_id}' upserted.")

    # 5. Semantic Search
    search_results = client.vectors.search(
        collection=COLLECTION_NAME,
        vector=embedding_vector,
        k=5
    )

    logger.info("Search Results:")
    for result in search_results:
        print(f"ID: {result.id} | Score: {result.score:.4f} | Metadata: {result.meta}")

if __name__ == "__main__":
    run_pipeline()

Document Store

Luma provides a JSON document store with metadata filtering.

# Store a document
doc_id = "user_123"
client.documents.put(
    collection="users",
    id=doc_id,
    document={"name": "Alice", "role": "admin", "preferences": {"theme": "dark"}}
)

# Retrieve
doc = client.documents.get(collection="users", id=doc_id)
print(doc.doc)

# Find by specific field
users = client.documents.find(
    collection="users",
    filter={"role": "admin"}
)

SQL (SQLite)

Execute standard SQL queries against the internal SQLite engine.

# DDL/DML
client.sql.exec("CREATE TABLE IF NOT EXISTS logistics (id INTEGER PRIMARY KEY, status TEXT)")
client.sql.exec("INSERT INTO logistics (status) VALUES (?)", params=["shipped"])

# Select
rows = client.sql.query("SELECT * FROM logistics WHERE status = ?", params=["shipped"])
print(rows)

Key-Value State

Manage distributed state with Optimistic Concurrency Control (OCC) and TTL.

# Set state with TTL (Time To Live)
client.state.put(
    key="job_status:99",
    value={"progress": 45, "stage": "processing"},
    ttl_ms=60000 # Expires in 60s
)

# Get state
state = client.state.get("job_status:99")
print(state.value)

Real-time Streams

Subscribe to database events using Server-Sent Events (SSE).

# Listen for all events
for event in client.stream.events():
    print(f"New Event: {event}")

LangChain Integration

langchain-luma implements the standard LangChain VectorStore interface. This allows Luma to be swapped seamlessly into existing RAG architectures.

RAG Example

The following example demonstrates how to use Luma as a retrieval backend for a standard document search pipeline.

from langchain_core.documents import Document
from langchain_luma import LumaVectorStore
# NOTE: In production, use standard embeddings (OpenAI, HuggingFace, etc.)
from langchain_community.embeddings import FakeEmbeddings

# 1. Initialize Embeddings
# The dimension size must match the collection configuration in Luma
embeddings = FakeEmbeddings(size=384)

# 2. Connect to Vector Store
# If the collection does not exist, LumaVectorStore can create it automatically
# depending on the internal implementation of the 'add_documents' method.

vector_store = LumaVectorStore.from_params(
    base_url="http://localhost:1234",
    api_key="dev",
    collection_name="rag_knowledge_base",
    embedding=embeddings,
    dim=384, # Required if creating a new collection
    create_if_not_exists=True
)

# 3. Ingest Documents
documents = [
    Document(
        page_content="Luma is optimized for low-latency vector retrieval.",
        metadata={"category": "database", "priority": "high"}
    ),
    Document(
        page_content="LangChain provides the glue code for LLM applications.",
        metadata={"category": "framework", "priority": "medium"}
    )
]

vector_store.add_documents(documents)
print("Documents indexed.")

# 4. Retrieval
# Perform a similarity search
query = "latency optimization"
retriever = vector_store.as_retriever(search_kwargs={"k": 1})
result = retriever.invoke(query)

print("-" * 40)
print(f"Query: {query}")
print(f"Retrieved Content: {result[0].page_content}")
print(f"Retrieved Metadata: {result[0].metadata}")
print("-" * 40)

API Reference

LumaClient

The main entry point for the SDK.

• __init__(url: str, api_key: str = None)
• Initializes the connection settings.

LumaClient.system

• health() -> dict
• Returns the operational status of the Luma server.

LumaClient.vectors

• create_collection(name: str, dim: int, metric: str = "cosine")

• Creates a new vector space. Metric options: cosine, dot.

• upsert(collection: str, id: str, vector: list[float], meta: dict = None)

• Inserts or updates a vector with optional metadata meta.

• search(collection: str, vector: list[float], k: int = 10) -> list[Hit]

• Performs a k-Nearest Neighbors (k-NN) search. Returns a list of Hit objects containing id, score, and meta.

LumaClient.documents

• put(collection: str, id: str, document: dict) -> DocRecord
• Stores a JSON document.
• get(collection: str, id: str) -> DocRecord
• Retrieves a document by ID.
• find(collection: str, filter: dict, limit: int) -> list[DocRecord]
• Queries documents using a MongoDB-like filter syntax.

LumaClient.sql

• exec(sql: str, params: list = None) -> dict
• Executes DDL/DML statements.
• query(sql: str, params: list = None) -> list[dict]
• Executes SELECT statements and returns rows.

LumaClient.state

• put(key: str, value: dict, ttl_ms: int = None, if_revision: int = None)
• Sets a key-value pair with optional expiration and revision check (OCC).
• get(key: str) -> StateItem
• Retrieves the current value and revision.
• batch_put(operations: list[StateBatchOperation])
• Performs atomic batch updates.

LumaClient.stream

• events(since: int = 0, types: str = None, key_prefix: str = None, collection: str = None)
• Detailed generator yielding SSE strings.

License

This project is licensed under the MIT License.