Nexus Enterprise AI (veloxs-nexus)

A high-performance, headless, layered data intelligence, format-aware chunking, configurable PII sanitization, and 3072-dimensional vector projection engine for enterprise AI applications.

Installation

# Standard in-memory installation
pip install veloxs-nexus

# With PostgreSQL + pgvector support
pip install veloxs-nexus[postgres]

# With YAML configuration support
pip install veloxs-nexus[yaml]

Key Features

• Format-Aware Structural Chunking: Converts CSV spreadsheets into row narratives ([Row ID: x] col: val | ...), JSON into structured objects, and text into semantic paragraph blocks.
• 3072-Dimensional Vector Projections: Multi-gram vector projection (unigrams 1.5x, bigrams 2.0x, trigrams 2.5x) normalized to exact L2 unit length (1.0) under IEEE 754 precision.
• 5-Stage Execution Trace Telemetry: Real-time stage durations, itemized summaries, and status logs returned with every payload for frontend rendering.
• Configurable Guardrails: Toggle PII redaction (enable_guardrails=True/False) to choose between compliance sanitization and raw verbatim fidelity.
• Multi-Tenant Cryptographic Isolation: Dynamic tenant-bound salt derivation (HKDF-SHA256("nexus-salt-" + tenant_id + "-" + key_id)) preventing cross-tenant correlation attacks.
• Serverless and Thread-Safe: Pure in-memory mode (in_memory_only=True) eliminates disk I/O, protected by threading.Lock mutexes across all indexes.

Code Examples

1. Standard Tabular CSV Processing with Guardrails

import nexus

# Initialize client in in-memory serverless mode
client = nexus.NexusClient(tenant_id="org-finance", in_memory_only=True)

csv_data = """employee_id,department,salary_usd,contact_email
101,Engineering,145000,john.doe@company.corp
102,Security,160000,jane.smith@company.corp"""

# Process document through 5-stage pipeline with PII redaction
doc = client.process_document(
    document_id="doc-ledger-01",
    name="salaries.csv",
    text=csv_data,
    file_type="csv",
    enable_guardrails=True
)

print(f"Document: {doc.name} | Total Chunks: {len(doc.chunks)}")
print(f"Chunk 0 Text: {doc.chunks[0].text}")
# Output: [Row ID: 1] employee_id: 101 | department: Engineering | salary_usd: 145000 | contact_email: [EMAIL]

# Inspect 5-Stage Execution Trace
for step in doc.execution_trace:
    print(f"[{step.step_number}/5] {step.stage_name} ({step.duration_ms}ms) -> {step.summary}")

2. Raw Fidelity Processing (Guardrails Bypassed)

When you need to index documents containing raw account numbers, code tokens, or verbatim records without redaction:

import nexus

client = nexus.NexusClient(in_memory_only=True)

raw_doc = client.process_document(
    document_id="doc-audit-02",
    name="audit_logs.txt",
    text="Transaction 9842 authorized by admin@bank.corp with key 4532-8901-2345-6789",
    file_type="txt",
    enable_guardrails=False  # Preserves verbatim text
)

print(f"Raw Chunk: {raw_doc.chunks[0].text}")
# Output: Transaction 9842 authorized by admin@bank.corp with key 4532-8901-2345-6789
print(f"Guardrails Status: {raw_doc.execution_trace[3].summary}")
# Output: Safety guardrails bypassed: preserving raw verbatim text without redaction.

3. PostgreSQL Table Sync and pgvector Ingestion

Stream live rows from any PostgreSQL source table directly into the 2-Tier knowledge_documents and knowledge_chunks schema:

import json
import nexus
import psycopg2
from psycopg2.extras import RealDictCursor, execute_values

client = nexus.NexusClient(in_memory_only=True)

def sync_table_to_knowledge_base(db_conn, org_id: str, workspace_id: str, table_name: str):
    with db_conn.cursor(cursor_factory=RealDictCursor) as cur:
        cur.execute(f'SELECT * FROM "{table_name}"')
        rows = cur.fetchall()

    if not rows:
        return

    # Convert rows into CSV-style narrative text
    headers = list(rows[0].keys())
    csv_body = ",".join(headers) + "\n" + "\n".join(
        ",".join(f'"{str(v)}"' if "," in str(v) else str(v) for v in r.values())
        for r in rows
    )

    # Process through Nexus
    doc = client.process_document(
        document_id=f"table_{table_name}",
        name=f"Table: {table_name}",
        text=csv_body,
        file_type="csv"
    )

    with db_conn.cursor() as cur:
        # Upsert Master Document
        cur.execute(
            """
            INSERT INTO knowledge_documents (id, org_id, workspace_id, name, file_type, file_size, content_hash, status)
            VALUES (%s, %s, %s, %s, %s, %s, %s, 'indexed')
            ON CONFLICT (id) DO UPDATE SET updated_at = CURRENT_TIMESTAMP;
            """,
            (doc.document_id, org_id, workspace_id, doc.name, "database_table", doc.file_size_bytes, doc.content_hash)
        )

        # Batch Upsert 3072D Vector Chunks
        chunk_data = []
        for chunk in doc.chunks:
            pg_vector_str = "[" + ",".join(map(str, chunk.embedding)) + "]"
            meta = dict(chunk.metadata)
            meta.update({"org_id": org_id, "workspace_id": workspace_id, "source_table": table_name})
            chunk_data.append((chunk.chunk_id, chunk.document_id, org_id, workspace_id, chunk.chunk_index, chunk.text, pg_vector_str, json.dumps(meta)))

        execute_values(
            cur,
            """
            INSERT INTO knowledge_chunks (id, document_id, org_id, workspace_id, chunk_index, chunk_text, embedding, metadata)
            VALUES %s
            ON CONFLICT (id) DO UPDATE SET chunk_text = EXCLUDED.chunk_text, embedding = EXCLUDED.embedding, metadata = EXCLUDED.metadata;
            """,
            chunk_data,
            template="(%s, %s, %s, %s, %s, %s, CAST(%s AS vector), %s::jsonb)"
        )
        db_conn.commit()

4. Grounded Question Answering and In-Memory Indexing

import nexus

client = nexus.NexusClient(in_memory_only=True)

# Ingest documentation
doc = client.process_document(
    document_id="arch-01",
    name="architecture.md",
    text="# Infrastructure\nAll database connections require TLS 1.3 encryption and mutual certificate authentication.",
    file_type="md"
)
client.index_document(doc)

# Query the knowledge base with fail-closed safety guardrails
response = client.ask("What encryption is required for database connections?")
print(f"Decision: {response.decision}")
print(f"Answer: {response.answer}")

5. Modular Sub-Layer Usage

Each sub-layer can be imported and utilized independently:

# 1. Direct 3072D Vector Embedding
from nexus.retrieval.engine import RetrievalEngine
retrieval = RetrievalEngine()
vector_3072 = retrieval.embed("Enterprise cloud infrastructure")

# 2. Standalone PII Redaction
from nexus.guardrails.pii import mask_pii
clean_text = mask_pii("Customer email is user@domain.com, card: 4532-0123-4567-8901")

# 3. Dynamic Multi-Tenant Encryption
from nexus.security.encryption import encrypt_text, decrypt_text
from nexus.security.config import EncryptionConfig

cfg = EncryptionConfig(secret_key="master-key-xyz", tenant_id="org-acme")
cipher = encrypt_text("Confidential Record", cfg)
plain = decrypt_text(cipher, cfg)

# 4. Format-Aware Chunking Engine
from nexus.processing.engine import ProcessingEngine
processing = ProcessingEngine()
row_chunks = processing.chunk_document("id,val\n1,Alpha\n2,Beta", file_type="csv")

Modular Sub-Layer Architecture

veloxs-nexus exports 7 decoupled sub-layers under the nexus.* namespace:

PostgreSQL + pgvector Schema

For production database persistence, use the provided schema:

CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

CREATE TABLE knowledge_documents (
    document_id         VARCHAR(128) PRIMARY KEY,
    name                VARCHAR(255) NOT NULL,
    file_type           VARCHAR(32) NOT NULL,
    file_size_bytes     BIGINT NOT NULL,
    content_hash        VARCHAR(64) NOT NULL,
    classification      VARCHAR(64) DEFAULT 'general',
    created_at          TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    updated_at          TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE knowledge_chunks (
    chunk_id            VARCHAR(128) PRIMARY KEY,
    document_id         VARCHAR(128) NOT NULL REFERENCES knowledge_documents(document_id) ON DELETE CASCADE,
    source_job          VARCHAR(64) NOT NULL,
    chunk_index         INTEGER NOT NULL,
    chunk_text          TEXT NOT NULL,
    metadata            JSONB DEFAULT '{}'::jsonb,
    embedding           VECTOR(3072) NOT NULL,
    created_at          TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_knowledge_chunks_embedding_hnsw 
ON knowledge_chunks 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

Multi-Tenant Cryptographic Isolation

Each tenant encryption and tokenization uses dynamic salt derivation: salt = HKDF-SHA256("nexus-salt-" + tenant_id + "-" + key_id)

This guarantees that two different tenants processing identical sensitive data produce cryptographically distinct ciphertexts.

License

Proprietary and confidential software. Copyright (c) 2026 Veloxs AI Inc. All rights reserved. See LICENSE for license terms.